fix(radio): restore upstream token-bucket duty-cycle enforcement

Commit 22eb9b87 ("Revert 'Merge remote-tracking branch origin/dev...'")
reverted an entire upstream merge to escape a bad merge state, dropping
860 lines across 66 files. Among the collateral never reconciled on a
later re-merge was eb4fa032's token-bucket duty-cycle enforcement — the
mechanism that keeps nodes under a configured airtime budget (and EU
868 MHz nodes under the legally-mandated duty cycle). The fork had fallen
back to fixed per-packet spacing (getAirtimeBudgetFactor reverted to 2.0),
losing the windowed enforcement.

This was never an intentional design choice, so restoring it re-aligns the
fork with upstream and REDUCES the merge-conflict surface: Dispatcher.{h,cpp}
now diverge from upstream by watchdog additions only (77 insertions, 0
deletions) instead of rewriting checkSend()/loop().

Restored from upstream: updateTxBudget/tx_budget_ms/duty_cycle_window_ms/
getRemainingTxBudget/getDutyCycleWindowMs and the windowed budget logic in
Dispatcher; getOutboundTotal() and the 0xFFFFFFFF count-all sentinel in
StaticPoolPacketManager; the getOutboundTotal() call in StatsFormatHelper.
Re-applied the fork's MQTT radio-watchdog on top as pure additions
(#ifdef WITH_MQTT_BRIDGE), keeping formatRadioDiag.

Stored airtime_factor settings keep their meaning: fork's t*factor spacing
and upstream's 1/(1+factor) windowed budget yield the same steady-state
duty cycle; upstream additionally allows short bursts within the window.

Phase 2 (CAD / radio_fem_rxgain, which touch NodePrefs persistence and
per-board FEM wiring) is documented in RESTORE_UPSTREAM_NOTES.md, not done
here. Builds: Heltec_v3 observer + plain repeater. NEEDS ON-DEVICE
duty-cycle validation before merge.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

RESTORE_UPSTREAM_NOTES.md

@@ -0,0 +1,67 @@
+# Restoring accidentally-reverted upstream features
+
+## Background
+
+On 2026-03-20, commit `22eb9b87` — *Revert "Merge remote-tracking branch 'origin/dev' into mqtt-bridge-implementation"* — reverted an entire upstream merge to escape a bad merge state, deleting 860 lines across 66 files. That was not intentional feature removal; it wholesale dropped a batch of upstream progress. When upstream was later re-merged, some collateral came back (MicroNMEA `claim()/release()`, the GAT562 board) but several upstream features were never reconciled and remained missing.
+
+This branch restores them. They are **pure upstream code the fork accidentally dropped**, so restoring re-aligns the fork with upstream and *reduces* the future merge-conflict surface rather than adding to it.
+
+## Phase 1 — duty-cycle enforcement (done on this branch)
+
+Restored the token-bucket duty-cycle enforcement and its cluster:
+
+- `src/Dispatcher.{h,cpp}` — restored upstream's `updateTxBudget()`, `tx_budget_ms`,
+  `duty_cycle_window_ms`, `getRemainingTxBudget()`, `getDutyCycleWindowMs()`, and the
+  windowed `checkSend()`/`loop()` budget logic. The fork's MQTT **radio watchdog** was
+  re-applied on top as pure additions (behind `#ifdef WITH_MQTT_BRIDGE`).
+- `src/helpers/StaticPoolPacketManager.{h,cpp}` — restored `getOutboundTotal()` and the
+  `0xFFFFFFFF` count-all sentinel in `countBefore()`.
+- `src/helpers/StatsFormatHelper.h` — restored the `getOutboundTotal()` call; kept the
+  fork's `formatRadioDiag` template.
+
+Net effect: these files now diverge from upstream by **watchdog additions only (0 deletions)**
+instead of rewriting upstream's TX logic.
+
+### Why this is safe (no reinterpretation of stored settings)
+
+`getAirtimeBudgetFactor()` changed *meaning* between fork and upstream, but the resulting
+duty cycle is the **same formula**:
+- Fork: `next_tx = t · factor` → duty ≈ `1/(1+factor)`
+- Upstream: `duty = 1/(1+factor)` enforced over a rolling window
+
+So a device's stored `airtime_factor` (a `NodePrefs` field, unchanged) keeps its meaning.
+The only behavioral difference is upstream enforces it over a 1-hour window (allowing
+short bursts) instead of rigid per-packet spacing — a strict improvement, and the
+mechanism that keeps EU 868 MHz nodes under the legally-mandated duty cycle.
+
+### Must be validated on-device before merge
+
+This touches core TX timing. Confirm on hardware:
+- Normal traffic still flows (repeater forwards, observer uplinks).
+- Sustained TX is throttled to the configured duty cycle (watch `get` airtime/queue stats;
+  verify `next_tx_time` spacing under load).
+- The observer radio watchdog still recovers a stuck radio (`radio_watchdog_minutes`).
+
+## Phase 2 — CAD and FEM RX gain (NOT done; needs care + device testing)
+
+Still missing at HEAD, also dropped by `22eb9b87`, still present upstream:
+
+- **`cad_enabled`** — hardware Channel Activity Detection (listen-before-talk) before TX.
+  The `Dispatcher`/`Radio` interface (`setCADEnabled`/`getCADEnabled`) is restored by
+  Phase 1, so CAD currently stays **off by default** (unchanged behavior). To make it
+  configurable again, restore:
+  - `NodePrefs.cad_enabled` field, its CLI get/set, and its persistence in
+    `CommonCLI.cpp` (`loadPrefsInt`/`savePrefs`). **Offset care:** this branch's
+    `/com_prefs` layout is carefully managed — add `cad_enabled` following the same
+    append-and-size-guard pattern used for `rx_boosted_gain`/`flood_max_*`, and add a
+    host-side round-trip test (see `scratchpad/migtest`).
+  - The `MyMesh::getCADEnabled()` override returning `_prefs.cad_enabled`.
+  - `RadioLibWrappers::setCADEnabled()` override so the hardware CAD is actually driven
+    (the fork's wrapper currently doesn't override it).
+- **`radio_fem_rxgain`** — LoRa front-end-module RX gain. Restore `NodePrefs.radio_fem_rxgain`
+  + CLI + persistence (same offset care), plus the per-board FEM wiring reverted across
+  ~20 `variants/*/target.cpp` and the `heltec_tracker_v2/LoRaFEMControl.{cpp,h}` files.
+  This is board-specific and only affects FEM-equipped hardware.
+
+Phase 2 is lower urgency than duty-cycle enforcement (CAD/FEM are capability gaps, not a
+compliance regression) and is best done as its own change with per-board hardware testing.

src/Dispatcher.cpp

@@ -8,7 +8,9 @@
 
 namespace mesh {
 
-#define MAX_RX_DELAY_MILLIS   32000  // 32 seconds
+#define MAX_RX_DELAY_MILLIS        32000  // 32 seconds
+#define MIN_TX_BUDGET_RESERVE_MS   100    // min budget (ms) required before allowing next TX
+#define MIN_TX_BUDGET_AIRTIME_DIV  2      // require at least 1/N of estimated airtime as budget before TX
 
 #ifndef NOISE_FLOOR_CALIB_INTERVAL
   #define NOISE_FLOOR_CALIB_INTERVAL   2000     // 2 seconds
@@ -20,12 +22,34 @@ void Dispatcher::begin() {
   _err_flags = 0;
   radio_nonrx_start = _ms->getMillis();
 
+  duty_cycle_window_ms = getDutyCycleWindowMs();
+  float duty_cycle = 1.0f / (1.0f + getAirtimeBudgetFactor());
+  tx_budget_ms = (unsigned long)(duty_cycle_window_ms * duty_cycle);
+  last_budget_update = _ms->getMillis();
+
   _radio->begin();
   prev_isrecv_mode = _radio->isInRecvMode();
 }
 
 float Dispatcher::getAirtimeBudgetFactor() const {
-  return 2.0;   // default, 33.3%  (1/3rd)
+  return 1.0;
+}
+
+void Dispatcher::updateTxBudget() {
+  unsigned long now = _ms->getMillis();
+  unsigned long elapsed = now - last_budget_update;
+
+  float duty_cycle = 1.0f / (1.0f + getAirtimeBudgetFactor());
+  unsigned long max_budget = (unsigned long)(getDutyCycleWindowMs() * duty_cycle);
+  unsigned long refill = (unsigned long)(elapsed * duty_cycle);
+  
+  if (refill > 0) {
+    tx_budget_ms += refill;
+    if (tx_budget_ms > max_budget) {
+      tx_budget_ms = max_budget;
+    }
+    last_budget_update = now;
+  }
 }
 
 int Dispatcher::calcRxDelay(float score, uint32_t air_time) const {
@@ -48,6 +72,7 @@ uint32_t Dispatcher::getRadioWatchdogMillis() const {
 void Dispatcher::loop() {
   if (millisHasNowPassed(next_floor_calib_time)) {
     _radio->triggerNoiseFloorCalibrate(getInterferenceThreshold());
+    _radio->setCADEnabled(getCADEnabled());
     next_floor_calib_time = futureMillis(NOISE_FLOOR_CALIB_INTERVAL);
   }
   _radio->loop();
@@ -93,11 +118,24 @@ void Dispatcher::loop() {
   if (outbound) {  // waiting for outbound send to be completed
     if (_radio->isSendComplete()) {
       long t = _ms->getMillis() - outbound_start;
-      total_air_time += t;  // keep track of how much air time we are using
+      total_air_time += t;
       //Serial.print("  airtime="); Serial.println(t);
 
-      // will need radio silence up to next_tx_time
-      next_tx_time = futureMillis(t * getAirtimeBudgetFactor());
+      updateTxBudget();
+
+      if (t > tx_budget_ms) {
+        tx_budget_ms = 0;
+      } else {
+        tx_budget_ms -= t;
+      }
+
+      if (tx_budget_ms < MIN_TX_BUDGET_RESERVE_MS) {
+        float duty_cycle = 1.0f / (1.0f + getAirtimeBudgetFactor());
+        unsigned long needed = MIN_TX_BUDGET_RESERVE_MS - tx_budget_ms;
+        next_tx_time = futureMillis((unsigned long)(needed / duty_cycle));
+      } else {
+        next_tx_time = _ms->getMillis();
+      }
 
       _radio->onSendFinished();
       last_radio_active_ms = _ms->getMillis();  // TX success → radio is alive
@@ -268,9 +306,20 @@ void Dispatcher::processRecvPacket(Packet* pkt) {
 }
 
 void Dispatcher::checkSend() {
-  if (_mgr->getOutboundCount(_ms->getMillis()) == 0) return;  // nothing waiting to send
-  if (!millisHasNowPassed(next_tx_time)) return;   // still in 'radio silence' phase (from airtime budget setting)
-  if (_radio->isReceiving()) {   // LBT - check if radio is currently mid-receive, or if channel activity
+  if (_mgr->getOutboundCount(_ms->getMillis()) == 0) return;
+  
+  updateTxBudget();
+  
+  uint32_t est_airtime = _radio->getEstAirtimeFor(MAX_TRANS_UNIT);
+  if (tx_budget_ms < est_airtime / MIN_TX_BUDGET_AIRTIME_DIV) {
+    float duty_cycle = 1.0f / (1.0f + getAirtimeBudgetFactor());
+    unsigned long needed = est_airtime / MIN_TX_BUDGET_AIRTIME_DIV - tx_budget_ms;
+    next_tx_time = futureMillis((unsigned long)(needed / duty_cycle));
+    return;
+  }
+  
+  if (!millisHasNowPassed(next_tx_time)) return;
+  if (_radio->isReceiving()) {
     if (cad_busy_start == 0) {
       cad_busy_start = _ms->getMillis();   // record when CAD busy state started
     }

src/Dispatcher.h

@@ -68,6 +68,8 @@ public:
 
   virtual void triggerNoiseFloorCalibrate(int threshold) { }
 
+  virtual void setCADEnabled(bool enable) { }
+
   virtual void resetAGC() { }
 
   virtual uint8_t getRadioState() const { return 0; }
@@ -102,6 +104,7 @@ public:
   virtual void queueOutbound(Packet* packet, uint8_t priority, uint32_t scheduled_for) = 0;
   virtual Packet* getNextOutbound(uint32_t now) = 0;    // by priority
   virtual int getOutboundCount(uint32_t now) const = 0;
+  virtual int getOutboundTotal() const = 0;
   virtual int getFreeCount() const = 0;
   virtual Packet* getOutboundByIdx(int i) = 0;
   virtual Packet* removeOutboundByIdx(int i) = 0;
@@ -141,8 +144,12 @@ class Dispatcher {
   bool  prev_isrecv_mode;
   uint32_t n_sent_flood, n_sent_direct;
   uint32_t n_recv_flood, n_recv_direct;
+  unsigned long tx_budget_ms;
+  unsigned long last_budget_update;
+  unsigned long duty_cycle_window_ms;
 
   void processRecvPacket(Packet* pkt);
+  void updateTxBudget();
 
 protected:
   PacketManager* _mgr;
@@ -155,12 +162,15 @@ protected:
   {
     outbound = NULL;
     total_air_time = rx_air_time = 0;
-    next_tx_time = 0;
+    next_tx_time = ms.getMillis();
     cad_busy_start = 0;
     next_floor_calib_time = next_agc_reset_time = 0;
     _err_flags = 0;
     radio_nonrx_start = 0;
     prev_isrecv_mode = true;
+    tx_budget_ms = 0;
+    last_budget_update = 0;
+    duty_cycle_window_ms = 3600000;
     last_watchdog_recovery = 0;
     last_radio_active_ms = 0;
   }
@@ -179,7 +189,9 @@ protected:
   virtual uint32_t getCADFailRetryDelay() const;
   virtual uint32_t getCADFailMaxDuration() const;
   virtual int getInterferenceThreshold() const { return 0; }    // disabled by default
+  virtual bool getCADEnabled() const { return false; }    // hardware CAD disabled by default
   virtual int getAGCResetInterval() const { return 0; }    // disabled by default
+  virtual unsigned long getDutyCycleWindowMs() const { return 3600000; }
 #ifdef WITH_MQTT_BRIDGE
   virtual uint32_t getRadioWatchdogMillis() const;  // observer-only radio recovery
 #endif
@@ -192,8 +204,9 @@ public:
   void releasePacket(Packet* packet);
   void sendPacket(Packet* packet, uint8_t priority, uint32_t delay_millis=0);
 
-  unsigned long getTotalAirTime() const { return total_air_time; }  // in milliseconds
+  unsigned long getTotalAirTime() const { return total_air_time; }
   unsigned long getReceiveAirTime() const {return rx_air_time; }
+  unsigned long getRemainingTxBudget() const { return tx_budget_ms; }
   uint32_t getNumSentFlood() const { return n_sent_flood; }
   uint32_t getNumSentDirect() const { return n_sent_direct; }
   uint32_t getNumRecvFlood() const { return n_recv_flood; }

src/helpers/StaticPoolPacketManager.cpp

@@ -9,6 +9,8 @@ PacketQueue::PacketQueue(int max_entries) {
 }
 
 int PacketQueue::countBefore(uint32_t now) const {
+  if (now == 0xFFFFFFFF) return _num;  // sentinel: count all entries regardless of schedule
+
   int n = 0;
   for (int j = 0; j < _num; j++) {
     if ((int32_t)(_schedule_table[j] - now) > 0) continue;   // scheduled for future... ignore for now
@@ -97,6 +99,10 @@ int  StaticPoolPacketManager::getOutboundCount(uint32_t now) const {
   return send_queue.countBefore(now);
 }
 
+int  StaticPoolPacketManager::getOutboundTotal() const {
+  return send_queue.count();
+}
+
 int StaticPoolPacketManager::getFreeCount() const {
   return unused.count();
 }

src/helpers/StaticPoolPacketManager.h

@@ -29,6 +29,7 @@ public:
   void queueOutbound(mesh::Packet* packet, uint8_t priority, uint32_t scheduled_for) override;
   mesh::Packet* getNextOutbound(uint32_t now) override;
   int getOutboundCount(uint32_t now) const override;
+  int getOutboundTotal() const override;
   int getFreeCount() const override;
   mesh::Packet* getOutboundByIdx(int i) override;
   mesh::Packet* removeOutboundByIdx(int i) override;

src/helpers/StatsFormatHelper.h

@@ -14,7 +14,7 @@ public:
       board.getBattMilliVolts(),
       ms.getMillis() / 1000,
       err_flags,
-      mgr->getOutboundCount(0xFFFFFFFF)
+      mgr->getOutboundTotal()
     );
   }