fix(reliability): hop governance belongs to rfplan, not boot reconciler (I2)

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

@@ -1,10 +1,26 @@
 """Boot-time reconciliation of crash-prone PineAP settings."""
+import time
+
 from server import (_apply_uci_wanted, _monitor_down, _raise_monitors,
                     PINEAPD_SAFE_UCI, device_run)
 
 WANTED_EXTRA = {'pineapd.@pineapd[0].autossidpool': '0'}
 POOL_CLEAR_MAX = 20
 MONITORS = ('wlan0mon', 'wlan1mon')
+# pineapd.wlan1mon.hop is intentionally NOT part of the applied set: channel
+# hopping is owned by the RF role manager (_pause_hop/_resume_hop). Hop=1 is
+# the healthy recon baseline, so reconciling it here would silently disable a
+# pager-enabled setting at every boot and latch GUARD PENDING after any
+# attack-role switch.
+GR_TTL_SECONDS = 30
+_GR_CACHE = {'t': 0.0, 'data': None}
+
+
+def _wanted():
+    wanted = {k: v for k, v in PINEAPD_SAFE_UCI.items()
+              if k != 'pineapd.wlan1mon.hop'}
+    wanted.update(WANTED_EXTRA)
+    return wanted
 
 
 def _pool_size():
@@ -16,7 +32,7 @@ def _pool_size():
 
 
 def reconcile(clear_pool=True):
-    changed = _apply_uci_wanted(dict(PINEAPD_SAFE_UCI, **WANTED_EXTRA))
+    changed = _apply_uci_wanted(_wanted())
     pool_cleared = False
     if clear_pool and _pool_size() > POOL_CLEAR_MAX:
         device_run(['uci', 'delete', 'pineapd.@ssidpool[0].ssid'])
@@ -24,12 +40,20 @@ def reconcile(clear_pool=True):
     if changed or pool_cleared:
         device_run(['uci', 'commit', 'pineapd'])
     raised = _raise_monitors() if any(_monitor_down(m) for m in MONITORS) else []
+    _GR_CACHE['data'] = None
     return {'changed': changed, 'pool_cleared': pool_cleared,
             'monitors_raised': raised}
 
 
 def guard_report():
+    now = time.time()
+    cached = _GR_CACHE['data']
+    if cached is not None and now - _GR_CACHE['t'] < GR_TTL_SECONDS:
+        return cached
     from server import _pending_uci
-    pending = _pending_uci(dict(PINEAPD_SAFE_UCI, **WANTED_EXTRA))
-    return {'in_sync': not pending, 'pending': pending,
+    pending = _pending_uci(_wanted())
+    report = {'in_sync': not pending, 'pending': pending,
               'pool_size': _pool_size()}
+    _GR_CACHE['t'] = now
+    _GR_CACHE['data'] = report
+    return report

scripts/smoke.sh

@@ -166,7 +166,6 @@ pineapd.wlan2mon.disable=1
 pineapd.wlan2mon.hop=0
 pineapd.wlan1mon.bands=5
 pineapd.wlan0mon.bands=2
-pineapd.wlan1mon.hop=0
 pineapd.@pineapd[0].autossidpool=0
 EOF
     pool="$(uci_get 'pineapd.@ssidpool[0].ssid')"

tests/test_mk8_guard.py

@@ -3,6 +3,8 @@ sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'payload', 'use
 import server
 import mk8_guard
 
+HOP_KEY = 'pineapd.wlan1mon.hop'
+
 class GuardTest(unittest.TestCase):
     def setUp(self):
         self.calls = []
@@ -30,19 +32,33 @@ class GuardTest(unittest.TestCase):
             return (0, '', '')
         mk8_guard.device_run = fake_run
         server.device_run = fake_run
+        mk8_guard._GR_CACHE['data'] = None
 
     def tearDown(self):
         server._iface_up = self.old_iface_up
         server.device_run = self.old_server_device_run
         mk8_guard.device_run = self.old_guard_device_run
+        mk8_guard._GR_CACHE['data'] = None
+
+    def wanted_count(self):
+        # Safe set minus wlan1mon.hop (rfplan owns hop), plus WANTED_EXTRA.
+        return (len(server.PINEAPD_SAFE_UCI) - 1 +
+                len(mk8_guard.WANTED_EXTRA))
 
     def test_applies_all_wanted_when_missing(self):
         self.pool = 0
         result = mk8_guard.reconcile(clear_pool=False)
         sets = [c[2] for c in self.calls if c[:2] == ['uci', 'set']]
-        self.assertEqual(len(sets),
-                         len(server.PINEAPD_SAFE_UCI) + len(mk8_guard.WANTED_EXTRA))
+        self.assertEqual(len(sets), self.wanted_count())
         self.assertTrue(result['changed'])
+        self.assertNotIn(HOP_KEY, {c[2].partition('=')[0]
+                                   for c in self.calls
+                                   if c[:2] == ['uci', 'set']})
+
+    def test_wanted_excludes_hop(self):
+        wanted = mk8_guard._wanted()
+        self.assertNotIn(HOP_KEY, wanted)
+        self.assertEqual(len(wanted), self.wanted_count())
 
     def test_clears_large_pool_only(self):
         self.pool = 25
@@ -55,5 +71,28 @@ class GuardTest(unittest.TestCase):
         result = mk8_guard.reconcile(clear_pool=True)
         self.assertFalse(result['pool_cleared'])
 
+    def test_report_ignores_hop_and_caches(self):
+        for key, value in mk8_guard._wanted().items():
+            self.uci[key] = value
+        self.uci[HOP_KEY] = '1'  # attack-role baseline; must stay ignored
+        report = mk8_guard.guard_report()
+        self.assertTrue(report['in_sync'], report)
+        gets = [c[2] for c in self.calls if c[:2] == ['uci', 'get']]
+        self.assertNotIn(HOP_KEY, gets)
+        n_after_first = len(self.calls)
+        self.assertIs(mk8_guard.guard_report(), report)
+        self.assertEqual(len(self.calls), n_after_first,
+                         'guard_report must serve from cache within TTL')
+        mk8_guard._GR_CACHE['t'] -= mk8_guard.GR_TTL_SECONDS * 2
+        mk8_guard.guard_report()
+        self.assertGreater(len(self.calls), n_after_first)
+        # reconcile mutates live state; it must invalidate the cached report.
+        mk8_guard.guard_report()
+        n_cached = len(self.calls)
+        mk8_guard.reconcile(clear_pool=False)
+        self.assertIsNone(mk8_guard._GR_CACHE['data'])
+        mk8_guard.guard_report()
+        self.assertGreater(len(self.calls), n_cached)
+
 if __name__ == '__main__':
     unittest.main()

tests/test_reliability_api.py

@@ -32,6 +32,8 @@ class ReliabilityApiTest(unittest.TestCase):
 
     def tearDown(self):
         server.device_run = self.old_device_run
+        import mk8_guard
+        mk8_guard._GR_CACHE['data'] = None
 
     def test_h_health_exposes_reliability_feed(self):
         status, h = server.h_health(None)