Fix Room and Sensor requests and add G3 Sensor builds

Bound ACL replies for encrypted routes, authorize Sensor ACL access using the actual Admin role, and preserve request retries after failed queue admission. Validate stored paths for replies, while keeping Room keep-alives direct-only.

Add both G3 RF-slot Sensor profiles, suppress battery alarms on unavailable measurements, and remove the unsupported Sensor WebConfig capability. Add executable regression coverage and CI checks.

.github/workflows/run-unit-tests.yml

@@ -238,6 +238,8 @@ jobs:
           python3 -B test/test_client_acl_metadata.py -v
           python3 -B test/test_client_acl_cli.py -v
           python3 -B test/test_client_acl_response.py -v
+          python3 -B test/test_client_acl_infrastructure.py -v
+          python3 -B test/test_stored_client_reply_routes.py -v
 
       - name: Setup Node for official app compatibility tests
         uses: actions/setup-node@v4
@@ -285,6 +287,7 @@ jobs:
           python3 -B test/test_gps_sync_interval.py -v
           python3 -B test/test_gps_sync_preferences.py -v
           python3 -B test/test_t1000e_sensor_power.py -v
+          python3 -B test/test_sensor_battery_alerts.py -v
 
       - name: Verify Companion ASCII defaults and USB session cleanup
         run: |
@@ -380,6 +383,7 @@ jobs:
       - name: Verify Full logging compiler flags and packaged capabilities
         run: |
           python3 -B test/test_upstream_observer_recipes.py -v
+          python3 -B test/test_station_g3_sensor_recipes.py -v
           python3 -B test/test_full_build_flag_dedup.py
           python3 -B test/test_full_logging_compile.py
           python3 -B test/test_full_logging_resume.py
@@ -462,6 +466,7 @@ jobs:
           python3 -B test/test_webconfig_gzip.py
           python3 -B test/test_webconfig_secret_masking.py
           python3 -B test/test_webconfig_station_ownership.py -v
+          python3 -B test/test_sensor_webconfig_capabilities.py -v
           python3 -B test/test_webconfig_ui_runtime.py
 
       - name: Upload Test Results

build_legacy.sh

@@ -3289,7 +3289,7 @@ is_esp32_full_only_bulk_target() {
     heltec_v4_expansionkit_repeater|heltec_v4_expansionkit_repeater_observer_mqtt|heltec_v4_expansionkit_room_server_observer_mqtt|\
     heltec_v4_tft_repeater|heltec_v4_tft_repeater_bridge_espnow|heltec_v4_tft_room_server|heltec_v4_tft_sensor|\
     lilygo_teth_elite_sx1262_repeater|lilygo_teth_elite_sx1262_room_server|lilygo_teth_elite_sx1262_repeater_observer_mqtt|lilygo_teth_elite_sx1262_room_server_observer_mqtt|\
-    station_g3_esp32_repeater|station_g3_esp32_logging_repeater|station_g3_esp32_room_server|station_g3_esp32_repeater_observer_mqtt|station_g3_esp32_room_server_observer_mqtt|\
+    station_g3_esp32_repeater|station_g3_esp32_logging_repeater|station_g3_esp32_room_server|station_g3_esp32_sensor|station_g3_esp32_r2_sensor|station_g3_esp32_repeater_observer_mqtt|station_g3_esp32_room_server_observer_mqtt|\
     rak_3112_repeater|rak_3112_repeater_bridge_rs232|rak_3112_repeater_bridge_espnow|rak_3112_repeater_observer_mqtt|rak_3112_room_server|rak_3112_room_server_observer_mqtt|rak_3112_sensor|\
     xiao_s3_wio_repeater|xiao_s3_wio_repeater_bridge_espnow|xiao_s3_wio_repeater_observer_mqtt|xiao_s3_wio_room_server|xiao_s3_wio_room_server_observer_mqtt|xiao_s3_wio_sensor|\
     lilygo_tdeck_repeater|\
@@ -3504,9 +3504,12 @@ declare_build_capability_contract() {
     record_build_expectation "sensor.ina3221" "INA3221"
   fi
 
+  # SensorMesh keeps CommonCLI's unsupported WebConfig callbacks. Its generic
+  # command text is linked even though only the board's browser OTA is usable.
   if [ "$env_platform" = "ESP32_PLATFORM" ] \
       && [ "$BUILD_PROFILE_FOR_TARGET" = "full" ] \
-      && ! is_esp32_companion_build "$env_name"; then
+      && ! is_esp32_companion_build "$env_name" \
+      && ! is_sensor_role_target "$env_name"; then
     pio_env_name=$(get_pio_build_env "$env_name")
     if pio_env_option_contains "$pio_env_name" build_flags "ADMIN_PASSWORD"; then
       record_build_expectation "web.webconfig" "start webconfig"

examples/simple_room_server/MyMesh.cpp

@@ -10,6 +10,7 @@
 #include <helpers/radiolib/RxBoostedGainDefaults.h>
 #include <helpers/CLICommandUtils.h>
 #include <helpers/ClientACLCLI.h>
+#include <helpers/ClientACLResponse.h>
 #include <helpers/ClientLoginPersistence.h>
 #include <helpers/ClientPathPersistence.h>
 #include <helpers/LazyPersistence.h>
@@ -147,7 +148,7 @@ void MyMesh::pushPostToClient(ClientInfo *client, PostInfo &post) {
   bool sent = false;
   if (reply) {
     reply->radio_reply = true;  // asynchronous response to the room subscription
-    if (client->out_path_len == OUT_PATH_UNKNOWN) {
+    if (!mesh::Packet::isValidPathLen(client->out_path_len)) {
       unsigned long delay_millis = 0;
       sent = sendFloodScoped(default_scope, reply, delay_millis,
                              _prefs.path_hash_mode + 1); // REVISIT
@@ -216,7 +217,11 @@ File MyMesh::openAppend(const char *fname) {
 }
 
 int MyMesh::handleRequest(ClientInfo *sender, uint32_t sender_timestamp, uint8_t *payload,
-                          size_t payload_len) {
+                          size_t payload_len, size_t reply_capacity) {
+  if (sender == NULL || payload == NULL || payload_len == 0 || reply_capacity < 4) return 0;
+  if (reply_capacity > mesh::CLIENT_ACL_DIRECT_REPLY_CAPACITY) {
+    reply_capacity = mesh::CLIENT_ACL_DIRECT_REPLY_CAPACITY;
+  }
   // uint32_t now = getRTCClock()->getCurrentTimeUnique();
   // memcpy(reply_data, &now, 4);   // response packets always prefixed with timestamp
   memcpy(reply_data, &sender_timestamp, 4); // reflect sender_timestamp back in response packet (kind of like a 'tag')
@@ -246,6 +251,7 @@ int MyMesh::handleRequest(ClientInfo *sender, uint32_t sender_timestamp, uint8_t
     return 4 + sizeof(stats);
   }
   if (payload[0] == REQ_TYPE_GET_TELEMETRY_DATA) {
+    if (payload_len < 2) return 0;
     uint8_t perm_mask = ~(payload[1]); // NEW: first reserved byte (of 4), is now inverse mask to apply to permissions
 
     telemetry.reset();
@@ -267,11 +273,13 @@ int MyMesh::handleRequest(ClientInfo *sender, uint32_t sender_timestamp, uint8_t
     return 4 + tlen; // reply_len
   }
   if (payload[0] == REQ_TYPE_GET_ACCESS_LIST && sender->isAdmin()) {
+    if (payload_len < 3) return 0;
     uint8_t res1 = payload[1];   // reserved for future  (extra query params)
     uint8_t res2 = payload[2];
     if (res1 == 0 && res2 == 0) {
-      uint8_t ofs = 4;
-      for (int i = 0; i < acl.getNumClients() && ofs + 7 <= sizeof(reply_data) - 4; i++) {
+      size_t ofs = 4;
+      // Preserve the legacy seven-byte entries within the encrypted route budget.
+      for (int i = 0; i < acl.getNumClients() && ofs + 7 <= reply_capacity; i++) {
         auto c = acl.getClientByIdx(i);
         if (!c->isAdmin()) continue;  // skip non-Admin entries
         memcpy(&reply_data[ofs], c->id.pub_key, 6); ofs += 6;  // just 6-byte pub_key prefix
@@ -789,7 +797,7 @@ void MyMesh::onAnonDataRecv(mesh::Packet *packet, const uint8_t *secret, const m
     } else {
       mesh::Packet *reply = createDatagram(PAYLOAD_TYPE_RESPONSE, sender, client->shared_secret, reply_data, 13);
       if (reply) {
-        if (client->out_path_len != OUT_PATH_UNKNOWN) { // we have an out_path, so send DIRECT
+        if (mesh::Packet::isValidPathLen(client->out_path_len)) { // we have an out_path, so send DIRECT
           sendDirect(reply, client->out_path, client->out_path_len, SERVER_RESPONSE_DELAY);
         } else {
           sendFloodReply(reply, SERVER_RESPONSE_DELAY, packet->getPathHashSize());
@@ -986,7 +994,7 @@ void MyMesh::onPeerDataRecv(mesh::Packet *packet, uint8_t type, int sender_idx,
       mesh::Utils::sha256((uint8_t*)&ack_hash, 4, data, 5 + text_len,
                           client->id.pub_key, PUB_KEY_SIZE);
 
-      if (client->out_path_len == OUT_PATH_UNKNOWN) {
+      if (!mesh::Packet::isValidPathLen(client->out_path_len)) {
         mesh::Packet *ack = createAck(ack_hash);
         if (ack) sendFloodReply(ack, TXT_ACK_DELAY, packet->getPathHashSize());
         delay_millis = TXT_ACK_DELAY + REPLY_DELAY_MILLIS;
@@ -1021,7 +1029,7 @@ void MyMesh::onPeerDataRecv(mesh::Packet *packet, uint8_t type, int sender_idx,
       auto reply = createDatagram(PAYLOAD_TYPE_TXT_MSG, client->id, secret,
                                   temp, 5 + reply_text_len);
       if (reply) {
-        if (client->out_path_len == OUT_PATH_UNKNOWN) {
+        if (!mesh::Packet::isValidPathLen(client->out_path_len)) {
           sendFloodReply(reply, delay_millis + SERVER_RESPONSE_DELAY,
                          packet->getPathHashSize());
         } else {
@@ -1059,7 +1067,7 @@ void MyMesh::onPeerDataRecv(mesh::Packet *packet, uint8_t type, int sender_idx,
         // if client sends too quickly, evict()
 
         // RULE: only send keep_alive response DIRECT!
-        if (client->out_path_len != OUT_PATH_UNKNOWN) {
+        if (mesh::Packet::isValidPathLen(client->out_path_len)) {
           uint32_t ack_hash; // calc ACK to prove to sender that we got request
           mesh::Utils::sha256((uint8_t *)&ack_hash, 4, data, 9, client->id.pub_key, PUB_KEY_SIZE);
 
@@ -1070,7 +1078,9 @@ void MyMesh::onPeerDataRecv(mesh::Packet *packet, uint8_t type, int sender_idx,
           }
         }
       } else {
-        int reply_len = handleRequest(client, sender_timestamp, &data[4], len - 4);
+        const size_t reply_capacity =
+            mesh::clientACLReplyCapacity(packet->isRouteFlood(), packet->path_len);
+        int reply_len = handleRequest(client, sender_timestamp, &data[4], len - 4, reply_capacity);
         if (reply_len > 0) { // valid command
           if (packet->isRouteFlood()) {
             // let this sender know path TO here, so they can use sendDirect(), and ALSO encode the response
@@ -1080,7 +1090,7 @@ void MyMesh::onPeerDataRecv(mesh::Packet *packet, uint8_t type, int sender_idx,
           } else {
             mesh::Packet *reply = createDatagram(PAYLOAD_TYPE_RESPONSE, client->id, secret, reply_data, reply_len);
             if (reply) {
-              if (client->out_path_len != OUT_PATH_UNKNOWN) { // we have an out_path, so send DIRECT
+              if (mesh::Packet::isValidPathLen(client->out_path_len)) { // we have an out_path, so send DIRECT
                 sendDirect(reply, client->out_path, client->out_path_len, SERVER_RESPONSE_DELAY);
               } else {
                 sendFloodReply(reply, SERVER_RESPONSE_DELAY, packet->getPathHashSize());
@@ -2290,7 +2300,7 @@ void MyMesh::onUserGpioTimerCompleted(uint8_t pin, uint8_t state,
                                         5 + (size_t)text_len);
   if (!packet) return;
   packet->radio_reply = true;  // delayed GPIO command completion
-  if (client->out_path_len == OUT_PATH_UNKNOWN) {
+  if (!mesh::Packet::isValidPathLen(client->out_path_len)) {
     sendFlood(packet, SERVER_RESPONSE_DELAY, path_hash_size);
   } else {
     sendDirect(packet, client->out_path, client->out_path_len,

examples/simple_room_server/MyMesh.h

@@ -333,7 +333,8 @@ class MyMesh : public mesh::Mesh, public CommonCLICallbacks,
   bool processAck(const uint8_t *data);
   mesh::Packet* createSelfAdvert();
   File openAppend(const char* fname);
-  int handleRequest(ClientInfo* sender, uint32_t sender_timestamp, uint8_t* payload, size_t payload_len);
+  int handleRequest(ClientInfo* sender, uint32_t sender_timestamp, uint8_t* payload, size_t payload_len,
+                    size_t reply_capacity = MAX_PACKET_PAYLOAD - CIPHER_MAC_SIZE - (CIPHER_BLOCK_SIZE - 1));
 #if MESH_ENABLE_ROOM_FLOOD_RULE_ENGINE
   bool evaluateFloodRuleTiming(const mesh::Packet* packet,
                                bool& fast_track);

examples/simple_sensor/SensorMesh.cpp

@@ -1,6 +1,7 @@
 #include "SensorMesh.h"
 #include <helpers/CLICommandUtils.h>
 #include <helpers/ClientACLCLI.h>
+#include <helpers/ClientACLResponse.h>
 #include <helpers/ClientLoginPersistence.h>
 #include <helpers/ClientPathPersistence.h>
 #include <helpers/LazyPersistence.h>
@@ -198,7 +199,11 @@ void SensorMesh::updateGpsTelemetryPolicy() {
 
 uint8_t SensorMesh::handleRequest(ClientInfo* from, uint32_t sender_timestamp,
                                   uint8_t req_type, uint8_t* payload,
-                                  size_t payload_len) {
+                                  size_t payload_len, size_t reply_capacity) {
+  if (from == NULL || (payload == NULL && payload_len != 0) || reply_capacity < 4) return 0;
+  if (reply_capacity > mesh::CLIENT_ACL_DIRECT_REPLY_CAPACITY) {
+    reply_capacity = mesh::CLIENT_ACL_DIRECT_REPLY_CAPACITY;
+  }
   const uint8_t perms = from->isAdmin() ? 0xFF : from->permissions;
 
   memcpy(reply_data, &sender_timestamp, 4);   // reflect sender_timestamp back in response packet (kind of like a 'tag')
@@ -263,12 +268,13 @@ uint8_t SensorMesh::handleRequest(ClientInfo* from, uint32_t sender_timestamp,
     return ofs;
   }
   if (req_type == REQ_TYPE_GET_ACCESS_LIST && payload_len >= 2
-      && (perms & PERM_ACL_ROLE_MASK) == PERM_ACL_ADMIN) {
+      && from->isAdmin()) {
     uint8_t res1 = payload[0];   // reserved for future  (extra query params)
     uint8_t res2 = payload[1];
     if (res1 == 0 && res2 == 0) {
-      uint8_t ofs = 4;
-      for (int i = 0; i < acl.getNumClients() && ofs + 7 <= sizeof(reply_data) - 4; i++) {
+      size_t ofs = 4;
+      // Preserve the legacy seven-byte entries within the encrypted route budget.
+      for (int i = 0; i < acl.getNumClients() && ofs + 7 <= reply_capacity; i++) {
         auto c = acl.getClientByIdx(i);
         if (c->permissions == 0) continue;  // skip deleted entries
         memcpy(&reply_data[ofs], c->id.pub_key, 6); ofs += 6;  // just 6-byte pub_key prefix
@@ -351,7 +357,7 @@ void SensorMesh::sendAlert(const ClientInfo* c, Trigger* t) {
 
   auto pkt = createDatagram(PAYLOAD_TYPE_TXT_MSG, c->id, c->shared_secret, data, 5 + text_len);
   if (pkt) {
-    if (c->out_path_len != OUT_PATH_UNKNOWN) {  // we have an out_path, so send DIRECT
+    if (mesh::Packet::isValidPathLen(c->out_path_len)) {  // we have an out_path, so send DIRECT
       sendDirect(pkt, c->out_path, c->out_path_len);
     } else {
       unsigned long delay_millis = 0;
@@ -629,7 +635,7 @@ void SensorMesh::onUserGpioTimerCompleted(uint8_t pin, uint8_t state,
                                         5 + (size_t)text_len);
   if (!packet) return;
   packet->radio_reply = true;  // delayed GPIO command completion
-  if (client->out_path_len == OUT_PATH_UNKNOWN) {
+  if (!mesh::Packet::isValidPathLen(client->out_path_len)) {
     sendFlood(packet, CLI_REPLY_DELAY_MILLIS, path_hash_size);
   } else {
     sendDirect(packet, client->out_path, client->out_path_len,
@@ -843,7 +849,7 @@ void SensorMesh::getPeerSharedSecret(uint8_t* dest_secret, int peer_idx) {
 }
 
 void SensorMesh::sendAckTo(const ClientInfo& dest, uint32_t ack_hash, uint8_t path_hash_size) {
-  if (dest.out_path_len == OUT_PATH_UNKNOWN) {
+  if (!mesh::Packet::isValidPathLen(dest.out_path_len)) {
     mesh::Packet* ack = createAck(ack_hash);
     if (ack) sendFloodScoped(default_scope, ack, TXT_ACK_DELAY, path_hash_size);
   } else {
@@ -869,31 +875,36 @@ void SensorMesh::onPeerDataRecv(mesh::Packet* packet, uint8_t type, int sender_i
   ClientInfo* from = acl.getClientByIdx(i);
 
   if (type == PAYLOAD_TYPE_REQ) {  // request (from a known contact)
+    if (len < 5) return;  // timestamp plus request type must be present
     uint32_t timestamp;
     memcpy(&timestamp, data, 4);
 
     if (timestamp > from->last_timestamp) {  // prevent replay attacks
-      uint8_t reply_len = handleRequest(from, timestamp, data[4], &data[5], len - 5);
+      const size_t reply_capacity =
+          mesh::clientACLReplyCapacity(packet->isRouteFlood(), packet->path_len);
+      uint8_t reply_len = handleRequest(from, timestamp, data[4], &data[5], len - 5, reply_capacity);
       if (reply_len == 0) return;  // invalid command
 
-      from->last_timestamp = timestamp;
-      from->last_activity = getRTCClock()->getCurrentTime();
-
+      bool reply_queued = false;
       if (packet->isRouteFlood()) {
         // let this sender know path TO here, so they can use sendDirect(), and ALSO encode the response
         mesh::Packet* path = createPathReturn(from->id, secret, packet->path, packet->path_len,
                                               PAYLOAD_TYPE_RESPONSE, reply_data, reply_len);
-        if (path) sendFloodReply(path, SERVER_RESPONSE_DELAY, packet->getPathHashSize());
+        if (path) reply_queued = sendFloodReply(path, SERVER_RESPONSE_DELAY, packet->getPathHashSize());
       } else {
         mesh::Packet* reply = createDatagram(PAYLOAD_TYPE_RESPONSE, from->id, secret, reply_data, reply_len);
         if (reply) {
-          if (from->out_path_len != OUT_PATH_UNKNOWN) {  // we have an out_path, so send DIRECT
-            sendDirect(reply, from->out_path, from->out_path_len, SERVER_RESPONSE_DELAY);
+          if (mesh::Packet::isValidPathLen(from->out_path_len)) {  // we have an out_path, so send DIRECT
+            reply_queued = sendDirect(reply, from->out_path, from->out_path_len, SERVER_RESPONSE_DELAY);
           } else {
-            sendFloodReply(reply, SERVER_RESPONSE_DELAY, packet->getPathHashSize());
+            reply_queued = sendFloodReply(reply, SERVER_RESPONSE_DELAY, packet->getPathHashSize());
           }
         }
       }
+      if (reply_queued) {
+        from->last_timestamp = timestamp;
+        from->last_activity = getRTCClock()->getCurrentTime();
+      }
     } else {
       MESH_DEBUG_PRINTLN("onPeerDataRecv: possible replay attack detected");
     }
@@ -944,7 +955,7 @@ void SensorMesh::onPeerDataRecv(mesh::Packet* packet, uint8_t type, int sender_i
 
           auto reply = createDatagram(PAYLOAD_TYPE_TXT_MSG, from->id, secret, temp, 5 + text_len);
           if (reply) {
-            if (from->out_path_len == OUT_PATH_UNKNOWN) {
+            if (!mesh::Packet::isValidPathLen(from->out_path_len)) {
               sendFloodReply(reply, CLI_REPLY_DELAY_MILLIS, packet->getPathHashSize());
             } else {
               sendDirect(reply, from->out_path, from->out_path_len, CLI_REPLY_DELAY_MILLIS);
@@ -1225,17 +1236,17 @@ bool SensorMesh::scheduleNormalRadio() {
   return true;
 }
 
-void SensorMesh::sendFloodScoped(const TransportKey& scope, mesh::Packet* packet,
+bool SensorMesh::sendFloodScoped(const TransportKey& scope, mesh::Packet* packet,
                                  uint32_t delay_millis, uint8_t path_hash_size) {
   if (scope.isNull()) {
-    sendFlood(packet, delay_millis, path_hash_size);
+    return sendFlood(packet, delay_millis, path_hash_size);
   } else {
     uint16_t codes[2] = {scope.calcTransportCode(packet), 0};
-    sendFlood(packet, codes, delay_millis, path_hash_size);
+    return sendFlood(packet, codes, delay_millis, path_hash_size);
   }
 }
 
-void SensorMesh::sendFloodReply(mesh::Packet* packet,
+bool SensorMesh::sendFloodReply(mesh::Packet* packet,
                                 unsigned long delay_millis,
                                 uint8_t path_hash_size) {
   TransportKey request_scope;
@@ -1245,15 +1256,13 @@ void SensorMesh::sendFloodReply(mesh::Packet* packet,
   switch (mesh::chooseReplyScope(request_scope_known, wildcard,
                                  !default_scope.isNull())) {
     case mesh::REPLY_SCOPE_REQUEST:
-      sendFloodScoped(request_scope, packet, delay_millis, path_hash_size);
-      break;
+      return sendFloodScoped(request_scope, packet, delay_millis, path_hash_size);
     case mesh::REPLY_SCOPE_DEFAULT:
-      sendFloodScoped(default_scope, packet, delay_millis, path_hash_size);
-      break;
+      return sendFloodScoped(default_scope, packet, delay_millis, path_hash_size);
     case mesh::REPLY_SCOPE_NONE:
-      sendFlood(packet, delay_millis, path_hash_size);
-      break;
+      return sendFlood(packet, delay_millis, path_hash_size);
   }
+  return false;
 }
 
 void SensorMesh::sendSelfAdvertisement(int delay_millis, bool flood) {
@@ -1446,7 +1455,7 @@ void SensorMesh::loop() {
           PAYLOAD_TYPE_RESPONSE, client->id, client->shared_secret,
           reply_data, 8 + telemetry_len);
       if (reply) {
-        if (client->out_path_len != OUT_PATH_UNKNOWN) {
+        if (mesh::Packet::isValidPathLen(client->out_path_len)) {
           sendDirect(reply, client->out_path, client->out_path_len, 0);
         } else {
           sendFloodScoped(scope, reply, 0, _prefs.path_hash_mode + 1);

examples/simple_sensor/SensorMesh.h

@@ -121,8 +121,8 @@ public:
 
   float getTelemValue(uint8_t channel, uint8_t type);
 
-  void sendFloodScoped(const TransportKey& scope, mesh::Packet* pkt, uint32_t delay_millis, uint8_t path_hash_size);
-  void sendFloodReply(mesh::Packet* packet, unsigned long delay_millis, uint8_t path_hash_size);
+  bool sendFloodScoped(const TransportKey& scope, mesh::Packet* pkt, uint32_t delay_millis, uint8_t path_hash_size);
+  bool sendFloodReply(mesh::Packet* packet, unsigned long delay_millis, uint8_t path_hash_size);
 
 protected:
   bool isTempRadioActive() const override {
@@ -243,7 +243,8 @@ private:
   bool telemHasChanged(ClientInfo* c);
   uint8_t handleLoginReq(const mesh::Identity& sender, const uint8_t* secret, uint32_t sender_timestamp, const uint8_t* data, bool is_flood);
   uint8_t handleRequest(ClientInfo* from, uint32_t sender_timestamp,
-                        uint8_t req_type, uint8_t* payload, size_t payload_len);
+                        uint8_t req_type, uint8_t* payload, size_t payload_len,
+                        size_t reply_capacity = MAX_PACKET_PAYLOAD - CIPHER_MAC_SIZE - (CIPHER_BLOCK_SIZE - 1));
   bool hasLocationTelemetryClient();
   void updateGpsTelemetryPolicy();
   mesh::Packet* createSelfAdvert();

examples/simple_sensor/main.cpp

@@ -1,4 +1,5 @@
 #include "SensorMesh.h"
+#include <math.h>
 #include <helpers/IdentityGeneration.h>
 #include <helpers/ui/StartupScreen.h>
 #include <helpers/ui/DisplayPowerSettings.h>
@@ -35,10 +36,12 @@ protected:
 
   void onSensorDataRead() override {
     float batt_voltage = getVoltage(TELEM_CHANNEL_SELF);
+    // Boards without battery measurement return zero; do not alarm on it.
+    const bool battery_available = batt_voltage > 0.0f && isfinite(batt_voltage);
 
     battery_data.recordData(getRTCClock(), batt_voltage);   // record battery
-    alertIf(batt_voltage < 3.4f, critical_batt, HIGH_PRI_ALERT, "Battery is critical!");
-    alertIf(batt_voltage < 3.6f, low_batt, LOW_PRI_ALERT, "Battery is low");
+    alertIf(battery_available && batt_voltage < 3.4f, critical_batt, HIGH_PRI_ALERT, "Battery is critical!");
+    alertIf(battery_available && batt_voltage < 3.6f, low_batt, LOW_PRI_ALERT, "Battery is low");
   }
 
   int querySeriesData(uint32_t start_secs_ago, uint32_t end_secs_ago, MinMaxAvg dest[], int max_num) override {

test/fixtures/client_acl_infrastructure/test_client_acl_infrastructure.cpp

@@ -0,0 +1,341 @@
+#include <algorithm>
+#include <cassert>
+#include <cstring>
+#include <iostream>
+#include <vector>
+#include <helpers/ClientACLResponse.h>
+#include <helpers/RoutingPolicy.h>
+
+#define REQ_TYPE_GET_ACCESS_LIST 0x05
+#define REQ_TYPE_GET_TELEMETRY_DATA 0x04
+#define REQ_TYPE_KEEP_ALIVE 0x02
+#define SERVER_RESPONSE_DELAY 300
+#define OUT_PATH_UNKNOWN 0xFF
+#define PERM_ACL_ROLE_MASK 7
+#define PERM_ACL_ADMIN 3
+
+namespace mesh {
+struct Identity {
+  uint8_t pub_key[PUB_KEY_SIZE] = {};
+  size_t copyHashTo(uint8_t* destination) const {
+    *destination = pub_key[0];
+    return PATH_HASH_SIZE;
+  }
+};
+struct Utils {
+  // Exercise the real packet-size admission and layout; replace crypto only.
+  static int encryptThenMAC(const uint8_t*, uint8_t* destination,
+                            const uint8_t* source, int length) {
+    const int encrypted = (length + CIPHER_BLOCK_SIZE - 1) / CIPHER_BLOCK_SIZE * CIPHER_BLOCK_SIZE;
+    memset(destination, 0, encrypted + CIPHER_MAC_SIZE);
+    memcpy(destination + CIPHER_MAC_SIZE, source, length);
+    return encrypted + CIPHER_MAC_SIZE;
+  }
+  static void sha256(uint8_t* out, int size, const uint8_t*, int, const uint8_t*, int) {
+    memset(out, 0, size);
+  }
+};
+struct Random {
+  void random(uint8_t* data, size_t length) { memset(data, 0x99, length); }
+};
+class Mesh {
+  std::vector<Packet*> allocations;
+public:
+  Identity self_id;
+  Random rng;
+  struct Tables { void markSent(Packet*) {} } tables;
+  Tables* _tables = &tables;
+  unsigned released = 0;
+  void releasePacket(Packet*) { ++released; }
+  void maybeScheduleFloodRetry(Packet*, uint8_t) {}
+  void maybeScheduleDirectRetry(Packet*, uint8_t) {}
+  uint8_t getTraceDirectPriority(Packet*) { return 0; }
+  void replaceQueuedSelfAdvertRetries(Packet*) {}
+  virtual bool sendPacket(Packet*, uint8_t, uint32_t) = 0;
+  bool sendFlood(Packet*, uint32_t, uint8_t);
+  bool sendFlood(Packet*, uint16_t*, uint32_t, uint8_t);
+  bool sendDirect(Packet*, const uint8_t*, uint8_t, uint32_t);
+  bool refuse_allocation = false;
+  ~Mesh() { for (auto* packet : allocations) delete packet; }
+  Packet* obtainNewPacket() {
+    if (refuse_allocation) return nullptr;
+    auto* packet = new Packet;
+    allocations.push_back(packet);
+    return packet;
+  }
+  uint8_t getContactTxRadio(const Identity&) { return RADIO_TX_AUTO; }
+  Random* getRNG() { return &rng; }
+  Packet* createPathReturn(const Identity&, const uint8_t*, const uint8_t*, uint8_t,
+                           uint8_t, const uint8_t*, size_t);
+  Packet* createPathReturn(const uint8_t*, const uint8_t*, const uint8_t*, uint8_t,
+                           uint8_t, const uint8_t*, size_t);
+  Packet* createDatagram(uint8_t, const Identity&, const uint8_t*, const uint8_t*, size_t);
+};
+}
+struct TransportKey {
+  bool null = true;
+  bool isNull() const { return null; }
+  uint16_t calcTransportCode(mesh::Packet*) const { return 7; }
+};
+struct RegionEntry { bool wildcard = false; bool isWildcard() const { return wildcard; } };
+struct RegionMap {
+  bool known = false;
+  int getTransportKeysFor(const RegionEntry&, TransportKey* key, int) {
+    key->null = !known; return known ? 1 : 0;
+  }
+};
+struct ClientInfo {
+  mesh::Identity id;
+  uint8_t permissions = 3;
+  uint8_t out_path_len = 0, out_path[MAX_PATH_SIZE] = {};
+  uint32_t last_timestamp = 0, last_activity = 0;
+  struct { struct { uint32_t sync_since = 0, pending_ack = 0; uint8_t push_failures = 0; } room; } extra;
+  bool isAdmin() const { return (permissions & PERM_ACL_ROLE_MASK) == PERM_ACL_ADMIN; }
+};
+struct ACL {
+  std::vector<ClientInfo> clients;
+  int getNumClients() const { return clients.size(); }
+  ClientInfo* getClientByIdx(int index) { return &clients.at(index); }
+};
+struct Clock : mesh::RTCClock {
+  uint32_t getCurrentTime() override { return 100; }
+  void setCurrentTime(uint32_t) override {}
+};
+#ifdef TEST_SENSOR
+#define TARGET_CLASS SensorMesh
+#else
+#define TARGET_CLASS MyMesh
+#endif
+class TARGET_CLASS : public mesh::Mesh {
+public:
+  uint8_t reply_data[MAX_PACKET_PAYLOAD] = {};
+  ACL acl;
+  ClientInfo sender;
+  Clock clock;
+  unsigned queued = 0, direct_sent = 0, flood_sent = 0;
+  bool refuse_queue = false;
+  TransportKey default_scope;
+  RegionEntry region;
+  RegionEntry* recv_pkt_region = nullptr;
+  RegionMap region_map;
+  mesh::Packet* last_reply = nullptr;
+  Clock* getRTCClock() { return &clock; }
+#ifdef TEST_SENSOR
+  uint8_t handleRequest(ClientInfo*, uint32_t, uint8_t, uint8_t*, size_t,
+                    size_t = mesh::CLIENT_ACL_DIRECT_REPLY_CAPACITY);
+#else
+  int handleRequest(ClientInfo*, uint32_t, uint8_t*, size_t,
+                    size_t = mesh::CLIENT_ACL_DIRECT_REPLY_CAPACITY);
+  int telemetryPrefix(uint8_t*, size_t);
+#endif
+  void receive(mesh::Packet*, uint8_t*, size_t);
+  bool admit(mesh::Packet* packet) {
+    if (!packet || refuse_queue) return false;
+    assert(packet->payload_len <= MAX_PACKET_PAYLOAD);
+    last_reply = packet;
+    ++queued;
+    if (packet->isRouteDirect()) ++direct_sent; else ++flood_sent;
+    return true;
+  }
+  bool sendPacket(mesh::Packet* packet, uint8_t, uint32_t) override { return admit(packet); }
+#ifdef TEST_SENSOR
+  bool sendFloodScoped(const TransportKey&, mesh::Packet*, uint32_t, uint8_t);
+  bool sendFloodReply(mesh::Packet*, unsigned long, uint8_t);
+#else
+  bool sendFloodReply(mesh::Packet* packet, unsigned long delay, uint8_t hash_size) {
+    return sendFlood(packet, delay, hash_size);
+  }
+#endif
+  mesh::Packet* createAck(uint32_t) { return nullptr; }
+  uint8_t getUnsyncedCount(ClientInfo*) { return 0; }
+};
+#include "production.inc"
+
+static int request(TARGET_CLASS& target, ClientInfo* sender, uint8_t* query, size_t length,
+                   size_t capacity = mesh::CLIENT_ACL_DIRECT_REPLY_CAPACITY) {
+#ifdef TEST_SENSOR
+  const uint8_t type = length ? query[0] : REQ_TYPE_GET_ACCESS_LIST;
+  return target.handleRequest(sender, 51, type, length ? query + 1 : nullptr,
+                              length ? length - 1 : 0, capacity);
+#else
+  return target.handleRequest(sender, 51, query, length, capacity);
+#endif
+}
+static bool included(const ClientInfo& client) {
+#ifdef TEST_SENSOR
+  return client.permissions != 0;
+#else
+  return client.isAdmin();
+#endif
+}
+static void fill(ACL& acl, unsigned count, bool mixed = false) {
+  acl.clients.assign(count, ClientInfo{});
+  for (unsigned i = 0; i < count; ++i) {
+    for (unsigned j = 0; j < PUB_KEY_SIZE; ++j) acl.clients[i].id.pub_key[j] = (i + j) % 255 + 1;
+    if (mixed) acl.clients[i].permissions = (i % 4 == 0) ? 0 : (i % 4 == 1) ? 1 : (i % 4 == 2) ? 3 : 4;
+  }
+}
+static unsigned verify(TARGET_CLASS& target, int length, size_t capacity) {
+  assert(length >= 4 && size_t(length) <= capacity && (length - 4) % 7 == 0);
+  uint32_t tag;
+  memcpy(&tag, target.reply_data, 4);
+  assert(tag == 51);
+  unsigned offset = 4;
+  for (const auto& client : target.acl.clients) {
+    if (!included(client)) continue;
+    if (offset + 7 > size_t(length)) break;
+    assert(!memcmp(target.reply_data + offset, client.id.pub_key, 6));
+    assert(target.reply_data[offset + 6] == client.permissions);
+    offset += 7;
+  }
+  assert(offset == unsigned(length));
+  return (length - 4) / 7;
+}
+static void print_legacy(const uint8_t* body, unsigned count, size_t length) {
+  static const char hex[] = "0123456789abcdef";
+  std::cout << "LEGACY:" << count << ':';
+  for (size_t i = 0; i < length; ++i) std::cout << hex[body[i] >> 4] << hex[body[i] & 15];
+  std::cout << '\n';
+}
+int main() {
+  uint8_t query[] = {REQ_TYPE_GET_ACCESS_LIST, 0, 0};
+  uint8_t secret[PUB_KEY_SIZE] = {};
+  unsigned paths = 0;
+  for (bool flood : {false, true}) {
+    for (unsigned path_len = 0; path_len < 256; ++path_len) {
+      if (!mesh::Packet::isValidPathLen(path_len)) {
+        assert(mesh::clientACLReplyCapacity(flood, path_len) == 0);
+        continue;
+      }
+      ++paths;
+      for (unsigned clients : {0, 1, 22, 23, 24, 25, 32, 256}) {
+        for (bool mixed : {false, true}) {
+          TARGET_CLASS target;
+          fill(target.acl, clients, mixed);
+          const size_t capacity = mesh::clientACLReplyCapacity(flood, path_len);
+          const int length = request(target, &target.sender, query, sizeof(query), capacity);
+          const unsigned count = verify(target, length, capacity);
+          const unsigned live = std::count_if(target.acl.clients.begin(), target.acl.clients.end(), included);
+          assert(count == std::min(live, unsigned((capacity - 4) / 7)));
+          uint8_t path[MAX_PATH_SIZE] = {};
+          mesh::Packet* packet = flood
+              ? target.createPathReturn(target.sender.id, secret, path, path_len,
+                                      PAYLOAD_TYPE_RESPONSE, target.reply_data, length)
+              : target.createDatagram(PAYLOAD_TYPE_RESPONSE, target.sender.id, secret, target.reply_data, length);
+          assert(packet && packet->payload_len <= MAX_PACKET_PAYLOAD);
+          const size_t cipher_bytes = packet->payload_len - 2 * PATH_HASH_SIZE - CIPHER_MAC_SIZE;
+          const size_t path_prefix = flood ? 2 + (path_len & 63) * ((path_len >> 6) + 1) : 0;
+          assert(cipher_bytes >= path_prefix);
+          assert(cipher_bytes - path_prefix + 2 <= MAX_FRAME_SIZE);
+          if (path_len == 0 && !mixed) {
+            const size_t offset = 2 * PATH_HASH_SIZE + CIPHER_MAC_SIZE + 4 + (flood ? 2 : 0);
+            print_legacy(packet->payload + offset, count, packet->payload_len - offset);
+          }
+        }
+      }
+    }
+  }
+  assert(paths > 200);
+  TARGET_CLASS target;
+  fill(target.acl, 32, true);
+  for (uint8_t permissions : {0, 1, 2, 4, 5, 6, 7, 255}) {
+    target.sender.permissions = permissions;
+    assert(request(target, &target.sender, query, sizeof(query)) == 0);
+  }
+  for (uint8_t permissions : {3, 131, 195}) {
+    target.sender.permissions = permissions;
+    assert(request(target, &target.sender, query, sizeof(query)) >= 4);
+  }
+  target.sender.permissions = 3;
+  assert(request(target, nullptr, query, sizeof(query)) == 0);
+  for (size_t length : {0, 1, 2}) {
+    // Exact-sized buffers make a reserved-byte overread visible to ASan.
+    std::vector<uint8_t> truncated(query, query + length);
+    assert(request(target, &target.sender, truncated.data(), length) == 0);
+  }
+  for (unsigned index : {1, 2}) {
+    query[index] = 1;
+    assert(request(target, &target.sender, query, sizeof(query)) == 0);
+    query[index] = 0;
+  }
+  for (size_t capacity = 0; capacity <= MAX_PACKET_PAYLOAD + 10; ++capacity) {
+    const int length = request(target, &target.sender, query, sizeof(query), capacity);
+    if (capacity < 4) assert(length == 0);
+    else verify(target, length, std::min(capacity, mesh::CLIENT_ACL_DIRECT_REPLY_CAPACITY));
+  }
+#ifndef TEST_SENSOR
+  for (size_t length : {0, 1}) {
+    std::vector<uint8_t> truncated(length, REQ_TYPE_GET_TELEMETRY_DATA);
+    assert(target.telemetryPrefix(truncated.data(), length) == 0);
+  }
+#endif
+  for (bool flood : {false, true}) {
+    for (uint8_t stored_path : {0, 1, 0xFE, 0xFF, 0xC0, 0x41}) {
+      TARGET_CLASS receiver;
+      fill(receiver.acl, 256, true);
+      receiver.sender.out_path_len = stored_path;
+      mesh::Packet packet;
+      packet.header = flood ? ROUTE_TYPE_FLOOD : ROUTE_TYPE_DIRECT;
+      packet.path_len = 0x60; // Maximum 64 bytes in the flood-return route.
+      const uint8_t data[] = {51, 0, 0, 0, REQ_TYPE_GET_ACCESS_LIST, 0, 0};
+      for (size_t length = 0; length < sizeof(data); ++length) {
+        std::vector<uint8_t> truncated(data, data + length);
+        receiver.receive(&packet, truncated.data(), length);
+        assert(receiver.queued == 0);
+      }
+      receiver.receive(&packet, const_cast<uint8_t*>(data), sizeof(data));
+      assert(receiver.queued == 1 && receiver.last_reply != nullptr);
+      assert(receiver.sender.last_timestamp == 51 && receiver.sender.last_activity == 100);
+      assert(receiver.last_reply->payload_len <= MAX_PACKET_PAYLOAD);
+      const bool direct = !flood && mesh::Packet::isValidPathLen(stored_path);
+      assert(receiver.direct_sent == unsigned(direct));
+      assert(receiver.flood_sent == unsigned(!direct));
+      const size_t capacity = mesh::clientACLReplyCapacity(flood, packet.path_len);
+      assert(verify(receiver, 4 + 7 * std::min(192U,
+          unsigned((capacity - 4) / 7)), capacity) > 0);
+    }
+  }
+  for (bool flood : {false, true}) {
+    for (uint8_t stored_path : {0, 1, 0xFE, 0xFF, 0xC0}) {
+      for (int scope = 0; scope < 4; ++scope) {
+        TARGET_CLASS receiver;
+        fill(receiver.acl, 32, true);
+        receiver.sender.last_timestamp = 50;
+        receiver.sender.last_activity = 12;
+        receiver.sender.out_path_len = stored_path;
+        if (scope == 1) receiver.default_scope.null = false;
+        if (scope >= 2) receiver.recv_pkt_region = &receiver.region;
+        receiver.region_map.known = scope == 2;
+        receiver.region.wildcard = scope == 3;
+        mesh::Packet packet;
+        packet.header = flood ? ROUTE_TYPE_FLOOD : ROUTE_TYPE_DIRECT;
+        packet.path_len = 0x60;
+        uint8_t data[] = {51, 0, 0, 0, REQ_TYPE_GET_ACCESS_LIST, 0, 0};
+        receiver.refuse_allocation = true;
+        receiver.receive(&packet, data, sizeof(data));
+        assert(receiver.queued == 0);
+#ifdef TEST_SENSOR
+        assert(receiver.sender.last_timestamp == 50 && receiver.sender.last_activity == 12);
+#endif
+        receiver.refuse_allocation = false;
+        receiver.refuse_queue = true;
+        receiver.receive(&packet, data, sizeof(data));
+        assert(receiver.queued == 0);
+#ifdef TEST_SENSOR
+        assert(receiver.sender.last_timestamp == 50 && receiver.sender.last_activity == 12);
+#endif
+        receiver.refuse_queue = false;
+        receiver.receive(&packet, data, sizeof(data));
+        assert(receiver.queued == 1 && receiver.sender.last_timestamp == 51);
+        assert(receiver.sender.last_activity == 100);
+        receiver.receive(&packet, data, sizeof(data));
+#ifdef TEST_SENSOR
+        assert(receiver.queued == 1); // Admission commits the Sensor replay floor.
+#else
+        assert(receiver.queued == 2); // Preserve Room's equal-tag retry semantics.
+#endif
+      }
+    }
+  }
+  std::cout << "Room/Sensor ACL route, input, and permission checks passed\n";
+}

test/test_client_acl_infrastructure.py

@@ -0,0 +1,168 @@
+#!/usr/bin/env python3
+"""Execute Room/Sensor ACL handlers and encrypted packet assembly with bounds."""
+from pathlib import Path
+import re
+import shutil
+import subprocess
+import sys
+import tempfile
+import unittest
+
+from test_replay_reset_integration import extract_braced
+from test_client_acl_response import parse_acl
+
+ROOT = Path(__file__).resolve().parents[1]
+FIXTURE = ROOT / "test/fixtures/client_acl_infrastructure/test_client_acl_infrastructure.cpp"
+
+
+def production_methods(role, transform=None):
+    sensor = role == "sensor"
+    path = ROOT / ("examples/simple_sensor/SensorMesh.cpp" if sensor
+                   else "examples/simple_room_server/MyMesh.cpp")
+    source = path.read_text()
+    if transform:
+        source = transform(source)
+    name = "SensorMesh" if sensor else "MyMesh"
+    handler = extract_braced(source, ("uint8_t " if sensor else "int ") + name + "::handleRequest(")
+    begin = handler.index("  if (req_type == REQ_TYPE_GET_TELEMETRY_DATA" if sensor
+                          else "  if (payload[0] == REQ_TYPE_GET_STATUS)")
+    prefix = handler[:begin]
+    acl = extract_braced(handler, "if (req_type == REQ_TYPE_GET_ACCESS_LIST" if sensor
+                         else "if (payload[0] == REQ_TYPE_GET_ACCESS_LIST")
+    generated = prefix + ("  (void)perms;\n" if sensor else "") + acl + "\nreturn 0;\n}\n"
+    signature = "if (type == PAYLOAD_TYPE_REQ) {  // request" if sensor else "if (type == PAYLOAD_TYPE_REQ && len >= 5)"
+    receive = extract_braced(source, signature)
+    generated += "void " + name + "::receive(mesh::Packet* packet, uint8_t* data, size_t len) {\n"
+    generated += "ClientInfo* " + ("from" if sensor else "client") + " = &sender;\n"
+    generated += "const uint8_t type = PAYLOAD_TYPE_REQ; uint8_t secret[PUB_KEY_SIZE] = {};\n" + receive + "\n}\n"
+    if not sensor:
+        telemetry = extract_braced(handler, "if (payload[0] == REQ_TYPE_GET_TELEMETRY_DATA)")
+        telemetry_prefix = telemetry[:telemetry.index("    telemetry.reset();")]
+        generated += "int MyMesh::telemetryPrefix(uint8_t* payload, size_t payload_len) {\n"
+        generated += "ClientInfo* sender = &this->sender; uint32_t sender_timestamp = 51;\n"
+        generated += "size_t reply_capacity = mesh::CLIENT_ACL_DIRECT_REPLY_CAPACITY;\n"
+        generated += prefix[prefix.index("{") + 1:] + telemetry_prefix + "return perm_mask;\n}\nreturn 0;\n}\n"
+    if sensor:
+        generated += extract_braced(source, "bool SensorMesh::sendFloodScoped(") + "\n"
+        generated += extract_braced(source, "bool SensorMesh::sendFloodReply(") + "\n"
+    mesh_source = (ROOT / "src/Mesh.cpp").read_text()
+    generated += "namespace mesh {\n#define MAX_COMBINED_PATH (MAX_PACKET_PAYLOAD - 2 - CIPHER_BLOCK_SIZE)\n"
+    for signature in ("Packet* Mesh::createPathReturn(const Identity& dest,",
+                      "Packet* Mesh::createPathReturn(const uint8_t* dest_hash,",
+                      "Packet* Mesh::createDatagram(", "bool Mesh::sendFlood(Packet* packet, uint32_t",
+                      "bool Mesh::sendFlood(Packet* packet, uint16_t*", "bool Mesh::sendDirect("):
+        generated += extract_braced(mesh_source, signature) + "\n"
+    packet_source = (ROOT / "src/Packet.cpp").read_text()
+    for signature in ("Packet::Packet()", "bool Packet::isValidPathLen(", "uint8_t Packet::copyPath(", "size_t Packet::writePath("):
+        generated += extract_braced(packet_source, signature) + "\n"
+    return generated + "}\n"
+
+
+class ClientAclInfrastructureTest(unittest.TestCase):
+    def execute(self, role, transform=None):
+        compiler = shutil.which("g++") or shutil.which("clang++")
+        self.assertIsNotNone(compiler, "host C++17 compiler required")
+        with tempfile.TemporaryDirectory(prefix="infrastructure-acl-") as directory:
+            work = Path(directory)
+            (work / "production.inc").write_text(production_methods(role, transform), encoding="ascii")
+            binary = work / "acl.exe"
+            command = [compiler, "-std=c++17", "-Wall", "-Wextra", "-Werror", "-Wno-unused-parameter",
+                       "-I" + str(work), "-I" + str(ROOT / "src"), str(FIXTURE), "-o", str(binary)]
+            if role == "sensor": command.insert(1, "-DTEST_SENSOR=1")
+            if sys.platform.startswith("linux"):
+                command[1:1] = ["-fsanitize=address,undefined", "-fno-sanitize-recover=all", "-fno-pie", "-no-pie"]
+            built = subprocess.run(command, capture_output=True, text=True, timeout=60)
+            self.assertEqual(built.returncode, 0, built.stdout + built.stderr)
+            run = subprocess.run([str(binary)], capture_output=True, text=True, timeout=30)
+            return run
+
+    def check_role(self, role):
+        run = self.execute(role)
+        self.assertEqual(run.returncode, 0, run.stdout + run.stderr)
+        self.assertIn("Room/Sensor ACL route, input, and permission checks passed", run.stdout)
+        decoded = 0
+        for line in run.stdout.splitlines():
+            if not line.startswith("LEGACY:"): continue
+            _, expected, body = line.split(":")
+            entries = parse_acl(bytes.fromhex(body))
+            self.assertEqual(len(entries), int(expected))
+            self.assertTrue(all(entry["perm"] == 3 for entry in entries))
+            decoded += 1
+        self.assertGreaterEqual(decoded, 10)
+
+    def test_room_actual_route_capacity_input_and_admin_filter(self):
+        self.check_role("room")
+
+    def test_sensor_actual_route_capacity_input_and_true_admin_permission(self):
+        self.check_role("sensor")
+
+    def test_old_acl_loop_is_rejected_for_both_roles(self):
+        def old(source):
+            self.assertIn("ofs + 7 <= reply_capacity", source)
+            return source.replace("ofs + 7 <= reply_capacity", "ofs + 7 <= sizeof(reply_data) - 4", 1)
+        for role in ("room", "sensor"):
+            with self.subTest(role=role): self.assertNotEqual(self.execute(role, old).returncode, 0)
+
+    def test_old_room_reserved_byte_overread_is_rejected(self):
+        def old(source):
+            self.assertIn("    if (payload_len < 3) return 0;", source)
+            return source.replace("    if (payload_len < 3) return 0;", "", 1)
+        self.assertNotEqual(self.execute("room", old).returncode, 0)
+
+    def test_old_sensor_truncated_packet_overread_is_rejected(self):
+        def old(source):
+            self.assertIn("    if (len < 5) return;", source)
+            return source.replace("    if (len < 5) return;", "", 1)
+        self.assertNotEqual(self.execute("sensor", old).returncode, 0)
+
+    def test_old_sensor_effective_permissions_break_genuine_admin(self):
+        def old(source):
+            self.assertIn("      && from->isAdmin())", source)
+            return source.replace("      && from->isAdmin())", "      && (perms & PERM_ACL_ROLE_MASK) == PERM_ACL_ADMIN)", 1)
+        self.assertNotEqual(self.execute("sensor", old).returncode, 0)
+
+    def test_old_force_flood_route_selection_is_rejected(self):
+        for role, client in (("room", "client"), ("sensor", "from")):
+            def old(source):
+                guard = "mesh::Packet::isValidPathLen(" + client + "->out_path_len)"
+                signature = ("if (type == PAYLOAD_TYPE_REQ && len >= 5)" if role == "room"
+                             else "if (type == PAYLOAD_TYPE_REQ) {  // request")
+                receive = extract_braced(source, signature)
+                self.assertIn(guard, receive)
+                altered = receive.replace(guard, client + "->out_path_len != OUT_PATH_UNKNOWN")
+                return source.replace(receive, altered, 1)
+            with self.subTest(role=role):
+                self.assertNotEqual(self.execute(role, old).returncode, 0)
+
+    def test_sensor_early_replay_commit_prevents_retry_after_rejection(self):
+        def old(source):
+            commit = """      if (reply_queued) {
+        from->last_timestamp = timestamp;
+        from->last_activity = getRTCClock()->getCurrentTime();
+      }"""
+            self.assertIn(commit, source)
+            return source.replace(commit, "      (void)reply_queued;", 1).replace(
+                "      bool reply_queued = false;",
+                "      from->last_timestamp = timestamp;\n"
+                "      from->last_activity = getRTCClock()->getCurrentTime();\n"
+                "      bool reply_queued = false;", 1)
+        self.assertNotEqual(self.execute("sensor", old).returncode, 0)
+
+    def test_sensor_scope_wrappers_must_propagate_failed_queue_admission(self):
+        def old(source):
+            changed, count = re.subn(r"return (sendFlood(?:Scoped)?\([^;]+\));",
+                                     r"\1; return true;", source)
+            self.assertGreaterEqual(count, 3)
+            return changed
+        self.assertNotEqual(self.execute("sensor", old).returncode, 0)
+
+    def test_default_capacity_declarations_and_ci_are_wired(self):
+        for path in ("examples/simple_room_server/MyMesh.h", "examples/simple_sensor/SensorMesh.h"):
+            self.assertIn("size_t reply_capacity = MAX_PACKET_PAYLOAD - CIPHER_MAC_SIZE - (CIPHER_BLOCK_SIZE - 1)",
+                          (ROOT / path).read_text())
+        self.assertIn("          python3 -B test/test_client_acl_infrastructure.py -v\n",
+                      (ROOT / ".github/workflows/run-unit-tests.yml").read_text())
+
+
+if __name__ == "__main__":
+    unittest.main()

test/test_esp32_hwcdc_recipes.py

@@ -34,7 +34,8 @@ USB_CDC = "ARDUINO_USB_CDC_ON_BOOT"
 USB_MACROS = (USB_MODE, USB_CDC)
 
 # These are the eight native-USB families that used TinyUSB before migration.
-# G2 already used HWCDC for its observer repeater and KISS roles.
+# G2 already used HWCDC for its observer repeater and KISS roles. New
+# Sensor roles inherit the same reviewed backend rather than migrating it.
 MIGRATED_BOARD_COUNTS = {
     "heltec_e213": 8,
     "heltec_e290": 9,
@@ -42,7 +43,7 @@ MIGRATED_BOARD_COUNTS = {
     "heltec_tracker_v1_1": 2,
     "heltec_tracker_v2": 15,
     "station-g2": 12,
-    "station-g3-esp32": 14,
+    "station-g3-esp32": 16,
     "t_beam_1w": 11,
 }
 PREVIOUS_G2_HWCDC = {"Station_G2_repeater_observer_mqtt", "Station_G2_kiss_modem"}
@@ -215,7 +216,7 @@ class Esp32HwcdcRecipeTests(unittest.TestCase):
                 "esp32": esp32, "kind": kind, "manifest": manifest,
             }
 
-    def test_all_78_migrated_roles_select_hwcdc_and_its_sdk_patch(self):
+    def test_all_reviewed_family_roles_select_hwcdc_and_its_sdk_patch(self):
         counts = Counter()
         migrated = set()
         for name, row in self.rows.items():
@@ -237,7 +238,7 @@ class Esp32HwcdcRecipeTests(unittest.TestCase):
                     migrated.add(name)
         self.assertEqual(counts, MIGRATED_BOARD_COUNTS)
         self.assertTrue(PREVIOUS_G2_HWCDC <= self.rows.keys())
-        self.assertEqual(len(migrated), 78)
+        self.assertEqual(len(migrated), 80)
 
     def test_migrated_uploads_use_hardware_reset_without_touch_or_port_wait(self):
         for name, row in self.rows.items():
@@ -265,7 +266,7 @@ class Esp32HwcdcRecipeTests(unittest.TestCase):
                 self.assertEqual(self.sdk["hwcdc_primary_enabled"](row["env"]),
                                  row["kind"] == "hwcdc")
                 self.assertFalse(self.sdk["native_primary_enabled"](row["env"]))
-        self.assertEqual(counts, {"hwcdc": 317, "uart": 122, "classic_uart": 61})
+        self.assertEqual(counts, {"hwcdc": 319, "uart": 122, "classic_uart": 61})
 
     def test_uart_existing_hwcdc_and_newer_mcu_recipes_keep_their_transports(self):
         counts = Counter()

test/test_full_infrastructure_selection.sh

@@ -119,11 +119,12 @@ for target in "${!PIO_ENV_PLATFORM_BY_NAME[@]}"; do
     full_only_count=$((full_only_count + 1))
   fi
 done
-[ "$full_only_count" -eq 147 ] || fail "expected 147 same-partition FULL-only targets, found $full_only_count"
+[ "$full_only_count" -eq 149 ] || fail "expected 149 same-partition FULL-only targets, found $full_only_count"
 
 for target in heltec_rc32_repeater Station_G2_repeater_observer_mqtt \
     Heltec_v3_repeater heltec_v4_r8_repeater \
-    heltec_v4_tft_repeater RAK_3112_repeater; do
+    heltec_v4_tft_repeater RAK_3112_repeater \
+    Station_G3_ESP32_sensor Station_G3_ESP32_r2_sensor; do
   is_esp32_full_only_bulk_target "$target" || fail "missed same-partition FULL-only target $target"
 done
 for target in LilyGo_TLora_V2_1_1_6_repeater_observer_mqtt_ \

test/test_sensor_battery_alerts.py

@@ -0,0 +1,178 @@
+"""Exercise the Sensor example's actual battery callback with absent readings."""
+from pathlib import Path
+import shutil
+import subprocess
+import tempfile
+import unittest
+
+ROOT = Path(__file__).resolve().parents[1]
+
+HARNESS = r'''
+#include <cassert>
+#include <cmath>
+#include <cstdint>
+#include <cstring>
+#include <limits>
+#include <vector>
+using std::isfinite;
+namespace mesh {
+struct MainBoard {};
+struct Radio {};
+struct MillisecondClock {};
+struct RNG {};
+struct RTCClock {};
+struct MeshTables {};
+}
+constexpr uint8_t TELEM_CHANNEL_SELF = 0;
+constexpr uint8_t LPP_VOLTAGE = 2;
+struct MinMaxAvg {};
+struct TimeSeriesData {
+  std::vector<float> values;
+  TimeSeriesData(int, int) {}
+  void recordData(mesh::RTCClock*, float voltage) { values.push_back(voltage); }
+  void calcMinMaxAvg(mesh::RTCClock*, uint32_t, uint32_t,
+                    MinMaxAvg*, uint8_t, uint8_t) {}
+};
+struct SensorMesh {
+  enum AlertPriority { LOW_PRI_ALERT, HIGH_PRI_ALERT };
+  struct Trigger { bool active = false; };
+  float voltage = 0;
+  mesh::RTCClock clock;
+  SensorMesh(mesh::MainBoard&, mesh::Radio&, mesh::MillisecondClock&,
+             mesh::RNG&, mesh::RTCClock&, mesh::MeshTables&) {}
+  virtual ~SensorMesh() = default;
+  virtual void onSensorDataRead() = 0;
+  virtual int querySeriesData(uint32_t, uint32_t, MinMaxAvg*, int) = 0;
+  virtual bool handleCustomCommand(uint32_t, char*, char*) = 0;
+  float getVoltage(uint8_t channel) {
+    assert(channel == TELEM_CHANNEL_SELF);
+    return voltage;
+  }
+  mesh::RTCClock* getRTCClock() { return &clock; }
+  void alertIf(bool condition, Trigger& trigger, AlertPriority, const char*) {
+    trigger.active = condition;
+  }
+};
+// PRODUCTION_CLASS
+struct TestMesh : MyMesh {
+  using MyMesh::MyMesh;
+  using MyMesh::onSensorDataRead;
+  using MyMesh::battery_data;
+  using MyMesh::critical_batt;
+  using MyMesh::low_batt;
+};
+int main(int argc, char** argv) {
+  assert(argc == 2);
+  mesh::MainBoard board;
+  mesh::Radio radio;
+  mesh::MillisecondClock milliseconds;
+  mesh::RNG random;
+  mesh::RTCClock clock;
+  mesh::MeshTables tables;
+  TestMesh sensor(board, radio, milliseconds, random, clock, tables);
+  const auto read = [&](float voltage) {
+    sensor.voltage = voltage;
+    sensor.onSensorDataRead();
+  };
+  if (strcmp(argv[1], "unavailable") == 0) {
+    // Zero is the board API's unsupported measurement and the missing-field
+    // fallback. Neither is a discharged battery. Keep history cadence intact.
+    read(0.0f);
+    assert(!sensor.critical_batt.active && !sensor.low_batt.active);
+    assert(sensor.battery_data.values.size() == 1);
+    assert(sensor.battery_data.values[0] == 0.0f);
+  } else if (strcmp(argv[1], "invalid") == 0) {
+    for (float voltage : {-1.0f, std::numeric_limits<float>::quiet_NaN(),
+                          std::numeric_limits<float>::infinity(),
+                          -std::numeric_limits<float>::infinity()}) {
+      read(voltage);
+      assert(!sensor.critical_batt.active && !sensor.low_batt.active);
+    }
+    assert(sensor.battery_data.values.size() == 4);
+  } else if (strcmp(argv[1], "healthy") == 0) {
+    read(4.1f);
+    assert(!sensor.critical_batt.active && !sensor.low_batt.active);
+    assert(sensor.battery_data.values.size() == 1);
+    assert(sensor.battery_data.values[0] == 4.1f);
+  } else if (strcmp(argv[1], "low") == 0) {
+    read(3.5f);
+    assert(!sensor.critical_batt.active && sensor.low_batt.active);
+    assert(sensor.battery_data.values.size() == 1);
+  } else if (strcmp(argv[1], "critical") == 0) {
+    read(3.3f);
+    assert(sensor.critical_batt.active && sensor.low_batt.active);
+    assert(sensor.battery_data.values.size() == 1);
+  } else if (strcmp(argv[1], "boundaries") == 0) {
+    read(3.4f);
+    assert(!sensor.critical_batt.active && sensor.low_batt.active);
+    read(3.6f);
+    assert(!sensor.critical_batt.active && !sensor.low_batt.active);
+    assert(sensor.battery_data.values.size() == 2);
+  } else if (strcmp(argv[1], "lost_measurement") == 0) {
+    read(3.3f);
+    assert(sensor.critical_batt.active && sensor.low_batt.active);
+    read(0.0f);
+    assert(!sensor.critical_batt.active && !sensor.low_batt.active);
+    assert(sensor.battery_data.values.size() == 2);
+    assert(sensor.battery_data.values[1] == 0.0f);
+    read(3.3f);
+    assert(sensor.critical_batt.active && sensor.low_batt.active);
+    assert(sensor.battery_data.values.size() == 3);
+  } else {
+    assert(false && "unknown case");
+  }
+}
+'''
+
+
+class SensorBatteryAlertTests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        compiler = shutil.which("g++") or shutil.which("clang++")
+        if compiler is None:
+            raise RuntimeError("host C++ compiler required")
+        source = (ROOT / "examples/simple_sensor/main.cpp").read_text()
+        start = source.index("class MyMesh : public SensorMesh {")
+        end = source.index("\n};", start) + len("\n};")
+        cls.directory = tempfile.TemporaryDirectory(prefix="sensor-battery-")
+        cls.addClassCleanup(cls.directory.cleanup)
+        work = Path(cls.directory.name)
+        harness = work / "test.cpp"
+        harness.write_text(HARNESS.replace("// PRODUCTION_CLASS", source[start:end]))
+        cls.binary = work / "test"
+        compiled = subprocess.run([
+            compiler, "-std=c++11", "-Wall", "-Wextra", "-Wno-unused-parameter",
+            str(harness), "-o", str(cls.binary),
+        ], capture_output=True, text=True)
+        if compiled.returncode:
+            raise AssertionError(compiled.stderr)
+
+    def run_case(self, case):
+        completed = subprocess.run([str(self.binary), case],
+                                   capture_output=True, text=True)
+        self.assertEqual(completed.returncode, 0, completed.stderr)
+
+    def test_unsupported_or_missing_voltage(self):
+        self.run_case("unavailable")
+
+    def test_invalid_voltage(self):
+        self.run_case("invalid")
+
+    def test_healthy_voltage(self):
+        self.run_case("healthy")
+
+    def test_low_battery(self):
+        self.run_case("low")
+
+    def test_critical_battery(self):
+        self.run_case("critical")
+
+    def test_existing_threshold_boundaries(self):
+        self.run_case("boundaries")
+
+    def test_invalid_reading_clears_pending_alarm_and_preserves_history_cadence(self):
+        self.run_case("lost_measurement")
+
+
+if __name__ == "__main__":
+    unittest.main()

test/test_sensor_webconfig_capabilities.py

@@ -0,0 +1,224 @@
+"""Keep the Sensor release contract consistent with its actual CLI callbacks."""
+import json
+from pathlib import Path
+import struct
+import subprocess
+import sys
+import tempfile
+import unittest
+
+from test_replay_reset_integration import extract_braced
+import test_wifi_ota_start as wifi_start
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+class SensorWebConfigCapabilityTest(unittest.TestCase):
+    compile_and_run = wifi_start.WiFiOtaStartTest.compile_and_run
+
+    def contract(self, target, profile="full", platform="ESP32_PLATFORM", admin=True,
+                 builder=None):
+        # Supply an already-resolved environment snapshot. Execute the actual
+        # builder functions; never initialize PlatformIO or invoke a build.
+        config = [["env:" + target, [["build_flags", ["-DADMIN_PASSWORD=fixture"]
+                                    if admin else []]]]]
+        script = r'''
+source "$1"
+pio() { printf 'PlatformIO is forbidden in this host fixture\n' >&2; exit 99; }
+TARGET=$2
+BUILD_PROFILE_FOR_TARGET=$3
+PIO_ENV_PLATFORM_BY_NAME[$TARGET]=$4
+PIO_ENV_FULL_BUILD_BY_NAME[$TARGET]=1
+PIO_ENV_FULL_WIFI_OTA_BY_NAME[$TARGET]=1
+PIO_ENV_OTA_BY_NAME[$TARGET]=1
+PIO_CONFIG_JSON=$5
+ESP32_FULL_BUILD=1
+BUILD_CAPABILITIES=()
+BUILD_EXPECTATIONS=()
+BUILD_APPLICATION_EXPECTATIONS=()
+declare_build_capability_contract "$TARGET" "${PIO_ENV_PLATFORM_BY_NAME[$TARGET]}"
+status=$?
+[ "$status" = 0 ] || exit "$status"
+for value in "${BUILD_CAPABILITIES[@]}"; do printf 'C:%s\0' "$value"; done
+for value in "${BUILD_EXPECTATIONS[@]}"; do printf 'E:%s\0' "$value"; done
+for value in "${BUILD_APPLICATION_EXPECTATIONS[@]}"; do printf 'A:%s\0' "$value"; done
+'''
+        result = subprocess.run(["bash", "-c", script, "sensor-capability-fixture",
+            str(builder or ROOT / "build_legacy.sh"), target, profile, platform,
+            json.dumps(config)], capture_output=True, timeout=30)
+        self.assertEqual(result.returncode, 0, result.stderr.decode())
+        records = {key: [] for key in "CEA"}
+        for value in result.stdout.split(b"\0"):
+            if value:
+                key, text = value.decode().split(":", 1)
+                records[key].append(text)
+        return records
+
+    def test_full_sensors_do_not_claim_an_unimplemented_portal(self):
+        for target in ("heltec_v3_sensor", "heltec_ct62_sensor", "mke_s3_sensor",
+                       "meshnology_w12_sensor", "Station_G3_ESP32_sensor",
+                       "Station_G3_ESP32_r2_sensor", "test_sensor_",
+                       "test_sensor_lora_ota_no_external_sensors"):
+            with self.subTest(target=target):
+                contract = self.contract(target)
+                self.assertNotIn("web.webconfig=start webconfig", contract["E"])
+                self.assertIn("ota.update.wifi=Started: http://%s/update", contract["E"])
+                if not target.endswith("lora_ota_no_external_sensors"):
+                    self.assertIn("ota.update.lora=image_hash MISMATCH after decode",
+                                  contract["E"])
+
+    def test_existing_infrastructure_and_full_companion_portals_remain(self):
+        for target in ("heltec_v3_repeater", "Station_G2_repeater",
+                       "test_repeater_observer_mqtt", "heltec_v3_room_server",
+                       "test_room_server_observer_mqtt", "heltec_v3_companion_radio_full"):
+            with self.subTest(target=target):
+                self.assertIn("web.webconfig=start webconfig", self.contract(target)["E"])
+        for profile, platform, admin in (("standard", "ESP32_PLATFORM", True),
+                                         ("full", "NRF52_PLATFORM", True),
+                                         ("full", "ESP32_PLATFORM", False)):
+            with self.subTest(profile=profile, platform=platform, admin=admin):
+                self.assertNotIn("web.webconfig=start webconfig",
+                                 self.contract("test_repeater", profile, platform, admin)["E"])
+
+    def test_packaged_sensor_manifest_keeps_real_ota_without_false_webconfig(self):
+        contract = self.contract("heltec_v3_sensor")
+        with tempfile.TemporaryDirectory(prefix="sensor-capabilities-") as directory:
+            work = Path(directory)
+            image = work / "firmware.elf"
+            application = work / "firmware.bin"
+            partitions = work / "partitions.bin"
+            manifest = work / "capabilities.json"
+            # Generic CLI text must not qualify a portal without callbacks.
+            markers = [value.split("=", 1)[1] for value in contract["E"] + contract["A"]]
+            image.write_bytes("\0".join(["start webconfig", *markers]).encode())
+            application.write_bytes(image.read_bytes())
+            def entry(kind, subtype, address, size):
+                return struct.pack("<HBBII16sI", 0x50AA, kind, subtype, address, size, b"test", 0)
+            partitions.write_bytes(entry(1, 0, 0xe000, 0x2000)
+                + entry(0, 0x10, 0x10000, 0x640000)
+                + entry(0, 0x11, 0x650000, 0x640000))
+            args = [sys.executable, "-B", str(ROOT / "scripts/check_firmware_capabilities.py"),
+                    "--image", str(image), "--output", str(manifest),
+                    "--target", "heltec_v3_sensor", "--platform", "ESP32_PLATFORM",
+                    "--build-profile", "full", "--require-ota", "--firmware-bin", str(application),
+                    "--partitions", str(partitions)]
+            for option, key in (("--capability", "C"), ("--expect", "E"), ("--expect-application", "A")):
+                for value in contract[key]:
+                    args.extend([option, value])
+            result = subprocess.run(args, capture_output=True, text=True, timeout=30)
+            self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+            produced = json.loads(manifest.read_text())
+            self.assertTrue(produced["verified"])
+            self.assertTrue(produced["ota_update_verified"])
+            self.assertEqual(set(produced["ota_update_methods"]), {"wifi", "lora"})
+            self.assertNotIn("web.webconfig", produced["capabilities"])
+
+    def test_actual_sensor_callbacks_reject_portal_but_keep_browser_ota(self):
+        sensor = (ROOT / "examples/simple_sensor/SensorMesh.h").read_text()
+        self.assertIn("public CommonCLICallbacks", sensor)
+        for path in (ROOT / "examples/simple_sensor").glob("*"):
+            if path.suffix in (".h", ".cpp"):
+                self.assertNotIn("startWebConfig(", path.read_text())
+                self.assertNotIn("stopWebConfig(", path.read_text())
+        header = (ROOT / "src/helpers/CommonCLI.h").read_text()
+        defaults = "\n".join(extract_braced(header, signature) for signature in (
+            "virtual bool startWebConfig(", "virtual bool stopWebConfig(",
+            "virtual bool stopWebConfigForOTA(", "virtual bool isWebConfigActive("))
+        source = (ROOT / "src/helpers/CommonCLI.cpp").read_text()
+        start = source.index('    } else if (memcmp(command, "start webconfig", 15)')
+        web = source[start:source.index("\n#endif", start)]
+        start = source.index('    } else if (memcmp(command, "start ota", 9)')
+        ota = source[start:source.index('    } else if (memcmp(command, "clock", 5)', start)]
+        fixture = r'''
+#include <cassert>
+#include <cstdio>
+#include <cstring>
+#include <initializer_list>
+struct Callbacks {
+@DEFAULTS@
+};
+struct SensorCallbacks : Callbacks {};
+struct PortalCallbacks : Callbacks {
+  bool startWebConfig(bool ap, char* reply) override {
+    strcpy(reply, ap ? "portal AP started" : "portal started"); return true;
+  }
+  bool stopWebConfig(char* reply) override { strcpy(reply, "portal stopped"); return true; }
+};
+struct Board {
+  int starts = 0, stops = 0;
+  bool forced = false;
+  bool startOTAUpdate(const char* name, char* reply, bool ap) {
+    assert(strcmp(name, "sensor") == 0); ++starts; forced = ap;
+    strcpy(reply, "Started: http://fixture/update"); return true;
+  }
+  bool stopOTAUpdate(char* reply) { ++stops; strcpy(reply, "Stopped OTA"); return true; }
+};
+struct Prefs { const char* node_name = "sensor"; };
+struct CLI {
+  Callbacks* _callbacks;
+  Board* _board;
+  Prefs* _prefs;
+  void run(const char* input, char* reply) {
+    char command[160] = {};
+    strcpy(command, input);
+    reply[0] = 0;
+    if (false) {
+@WEB@
+@OTA@
+    }
+  }
+};
+int main() {
+  SensorCallbacks sensor;
+  Board board;
+  Prefs prefs;
+  CLI cli{&sensor, &board, &prefs};
+  char reply[160] = {};
+  for (const char* command : {"start webconfig", "start webconfig ap", "stop webconfig"}) {
+    cli.run(command, reply);
+    assert(strcmp(reply, "ERR: webconfig not supported on this build") == 0);
+    assert(!sensor.isWebConfigActive());
+    assert(board.starts == 0 && board.stops == 0);
+  }
+  cli.run("start webconfig invalid", reply);
+  assert(strcmp(reply, "ERR: usage start webconfig [ap]") == 0);
+  cli.run("start ota", reply);
+  assert(strcmp(reply, "Started: http://fixture/update") == 0);
+  assert(board.starts == 1 && !board.forced);
+  cli.run("start ota ap", reply);
+  assert(strcmp(reply, "Started: http://fixture/update") == 0);
+  assert(board.starts == 2 && board.forced);
+  cli.run("start ota invalid", reply);
+  assert(strcmp(reply, "ERR: usage start ota [ap]") == 0 && board.starts == 2);
+  cli.run("stop ota", reply);
+  assert(strcmp(reply, "Stopped OTA") == 0 && board.stops == 1);
+  PortalCallbacks portal;
+  cli._callbacks = &portal;
+  cli.run("start webconfig", reply);
+  assert(strcmp(reply, "portal started") == 0);
+  cli.run("start webconfig ap", reply);
+  assert(strcmp(reply, "portal AP started") == 0);
+  cli.run("stop webconfig", reply);
+  assert(strcmp(reply, "portal stopped") == 0);
+}
+'''
+        fixture = fixture.replace("@DEFAULTS@", defaults).replace("@WEB@", web).replace("@OTA@", ota)
+        self.compile_and_run(fixture, "-DESP_PLATFORM=1", "-DADMIN_PASSWORD=fixture")
+
+    def test_unpatched_builder_is_a_meaningful_negative_control(self):
+        source = (ROOT / "build_legacy.sh").read_text()
+        guard = '      && ! is_esp32_companion_build "$env_name" \\\n      && ! is_sensor_role_target "$env_name"; then'
+        self.assertIn(guard, source)
+        old = source.replace(guard, '      && ! is_esp32_companion_build "$env_name"; then', 1)
+        with tempfile.TemporaryDirectory(prefix="sensor-old-contract-") as directory:
+            builder = Path(directory) / "build_legacy.sh"
+            builder.write_text(old)
+            self.assertIn("web.webconfig=start webconfig", self.contract("heltec_v3_sensor", builder=builder)["E"])
+
+    def test_suite_is_explicitly_wired_in_ci(self):
+        workflow = (ROOT / ".github/workflows/run-unit-tests.yml").read_text()
+        self.assertIn("          python3 -B test/test_sensor_webconfig_capabilities.py -v\n", workflow)
+
+
+if __name__ == "__main__":
+    unittest.main()

test/test_station_g3_sensor_recipes.py

@@ -0,0 +1,239 @@
+#!/usr/bin/env python3
+"""Host contracts for the Station G3 Sensor role on each RF daughterboard slot.
+
+Resolve tracked source recipes without starting PlatformIO, building firmware,
+flashing hardware, or changing a release catalog. These fixtures check routing
+and metadata; linked-image and RAM qualification still require real builds.
+"""
+
+import json
+from pathlib import Path
+import re
+import subprocess
+import sys
+import tempfile
+import unittest
+
+from test_upstream_observer_recipes import Recipes
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT / "scripts"))
+import generate_mesh_america_release as catalogs
+import generate_picker_controls as controls
+from package_cascade_release import category
+from package_esp32_partition_migration import BOARDS
+
+SLOTS = {
+    "Station_G3_ESP32": (48, 11, 21, 47, 12, 14, 13, 10),
+    "Station_G3_ESP32_r2": (2, 43, 44, 1, 39, 41, 40, 42),
+}
+RADIO_PINS = (
+    "P_LORA_DIO_1", "P_LORA_NSS", "P_LORA_RESET", "P_LORA_BUSY",
+    "P_LORA_SCLK", "P_LORA_MISO", "P_LORA_MOSI", "P_PRIMARY_LNA_EN",
+)
+SENSORS = tuple(slot + "_sensor" for slot in SLOTS)
+SOURCE = "a" * 40
+FAMILY = "v1.17.1.9-halo-keymind-cascade-dev-" + SOURCE[:8]
+
+
+def definitions(flags):
+    """Preserve definition histories so a second slot's flags cannot shadow pins."""
+    result = {}
+    for name, value in re.findall(r"(?m)^\s*-D\s+(\w+)=([^\s;]+)", flags):
+        result.setdefault(name, []).append(value)
+    return result
+
+
+def manifest(target):
+    profile = target + "-full-logging-ota"
+    stem = profile + "-" + FAMILY
+    capabilities = ["profile.full", "ota.update.lora", "ota.update.wifi"]
+    return dict(target=target, artifact_target=profile, platformio_env=target,
+                platform="ESP32_PLATFORM", build_profile="full", verified=True,
+                ota_update_verified=True, source_commit=SOURCE,
+                capabilities=capabilities, ota_update_methods=["lora", "wifi"],
+                verification=[dict(capability=capability, present=True,
+                                   source="linked image") for capability in capabilities],
+                files=[stem + suffix for suffix in (
+                    ".bin", "-merged.bin", ".capabilities.json", ".memory.json")])
+
+
+class StationG3SensorRecipesTests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.recipes = Recipes()
+
+    def test_each_sensor_inherits_its_own_slot_and_only_one_pin_set(self):
+        text = (ROOT / "variants/station_g3_esp32/platformio.ini").read_text()
+        for slot, pins in SLOTS.items():
+            section = "env:" + slot + "_sensor"
+            with self.subTest(slot=slot):
+                self.assertEqual(text.count("[" + section + "]"), 1)
+                self.assertEqual(self.recipes.config.get(section, "extends"), slot)
+                values = definitions(self.recipes.value(section, "build_flags"))
+                for name, pin in zip(RADIO_PINS, pins):
+                    self.assertEqual(values[name], [str(pin)], name)
+                for name, pin in {"P_PA1_EN": 9, "PIN_BOARD_SDA": 5,
+                                  "PIN_BOARD_SCL": 6, "PIN_USER_BTN": 38,
+                                  "PIN_GPS_RX": 15, "PIN_GPS_TX": 7}.items():
+                    self.assertEqual(values[name], [str(pin)], name)
+                all_pins = (*pins, 9, 5, 6, 38, 15, 7)
+                self.assertEqual(len(all_pins), len(set(all_pins)))
+                self.assertEqual(values["P_PRIMARY_LNA_EN_ACTIVE"], ["LOW"])
+                self.assertEqual(values["P_PA1_EN_ACTIVE"], ["HIGH"])
+
+    def test_sensor_entrypoints_board_ui_and_ota_sources_are_selected(self):
+        for target in SENSORS:
+            section = "env:" + target
+            with self.subTest(target=target):
+                for path in ("../examples/simple_sensor/main.cpp",
+                             "../examples/simple_sensor/SensorMesh.cpp",
+                             "../examples/simple_sensor/UITask.cpp",
+                             "../examples/simple_sensor/TimeSeriesData.cpp",
+                             "../variants/station_g3_esp32/target.cpp",
+                             "../variants/station_g3_esp32/StationG3Board.cpp",
+                             "../variants/station_g3_esp32/LoRaFEMControl.cpp",
+                             "helpers/ui/SH1106Display.cpp", "helpers/ui/MomentaryButton.cpp",
+                             "helpers/sensors/EnvironmentSensorManager.cpp",
+                             "helpers/esp32/WiFiOtaSeeder.cpp", "helpers/ota/OtaTinf.c"):
+                    self.assertTrue(self.recipes.includes(section, path), path)
+                for role in ("simple_repeater", "simple_room_server", "companion_radio", "kiss_modem"):
+                    self.assertFalse(self.recipes.includes(section, f"../examples/{role}/main.cpp"))
+                flags = self.recipes.value(section, "build_flags")
+                for flag in ("ESP32_PLATFORM", "ENABLE_OTA=1", "OTA_FLASH_STORE=1",
+                             "ENV_INCLUDE_GPS=1", "ENV_INCLUDE_INA219=1",
+                             "ENV_INCLUDE_BME280=1", "DISPLAY_CLASS=SH1106Display"):
+                    self.assertIn(flag, flags)
+                self.assertNotIn("-UENV_INCLUDE_", flags)
+
+    def test_native_usb_partition_and_qualification_hooks_match_existing_slot(self):
+        board = json.loads((ROOT / "boards/station-g3-esp32.json").read_text())
+        self.assertEqual(board["build"]["mcu"], "esp32s3")
+        self.assertEqual(board["upload"]["flash_size"], "16MB")
+        self.assertIn("-DARDUINO_USB_MODE=1", board["build"]["extra_flags"])
+        self.assertIn("-DARDUINO_USB_CDC_ON_BOOT=1", board["build"]["extra_flags"])
+        for target in SENSORS:
+            section = "env:" + target
+            with self.subTest(target=target):
+                self.assertEqual(self.recipes.value(section, "board"), "station-g3-esp32")
+                self.assertEqual(self.recipes.value(section, "board_build.partitions"),
+                                 "default_16MB.csv")
+                for hook in ("esp32_usb_session_fix.py", "meshcore_image_identity.py",
+                             "tools/mota/pio_endf.py", "check_firmware_ram.py",
+                             "check_esp32_dram.py"):
+                    self.assertIn(hook, self.recipes.value(section, "extra_scripts"))
+                self.assertIn("file://arch/esp32/AsyncElegantOTA",
+                              self.recipes.value(section, "lib_deps"))
+
+    def test_release_discovery_and_full_matrix_keep_both_sensor_identities(self):
+        # The fixture rows come from the source-resolved recipes above. Any
+        # accidental PIO or compilation call fails before doing work.
+        program = r'''
+set -euo pipefail
+source build.sh
+pio() { echo "unexpected PlatformIO" >&2; return 99; }
+prepare_esp32_arduino3_framework() { return 99; }
+run_pio_with_size_detection() { return 99; }
+collect_build_artifacts() { return 99; }
+SUPPORTED_PIO_ENVS=(Station_G3_ESP32_sensor Station_G3_ESP32_r2_sensor)
+for target in "${SUPPORTED_PIO_ENVS[@]}"; do
+  PIO_ENV_PLATFORM_BY_NAME[$target]=ESP32_PLATFORM
+  PIO_ENV_BOARD_BY_NAME[$target]=station-g3-esp32
+  PIO_ENV_FULL_BUILD_BY_NAME[$target]=1
+  PIO_ENV_OTA_BY_NAME[$target]=1
+  PIO_ENV_FULL_WIFI_OTA_BY_NAME[$target]=1
+  PIO_ENV_MQTT_BY_NAME[$target]=0
+  is_sensor_role_target "$target"
+  is_esp32_full_only_bulk_target "$target"
+  ! is_esp32_partition_migration_full_target "$target"
+  [ "$(get_exact_identity_full_pio_env "$target")" = "$target" ]
+done
+[ "$(get_pio_envs_for_variant_role sensor)" = "$(printf '%s\n' "${SUPPORTED_PIO_ENVS[@]}")" ]
+REQUIRE_OTA_UPDATES=0
+calls=()
+run_logged_build_targets() {
+  [ "$BUILD_PROFILE_EFFECTIVE:$ESP32_FULL_BUILD:$FIRMWARE_FILENAME_INFIX" = full:1:full-logging ]
+  for target in "$@"; do is_lora_ota_build "$target"; done
+  calls+=("$*")
+}
+run_logging_matrix_build_targets "${SUPPORTED_PIO_ENVS[@]}" >/dev/null
+[ "${#calls[@]}" = 1 ]
+[ "${calls[0]}" = "${SUPPORTED_PIO_ENVS[*]}" ]
+'''
+        result = subprocess.run(["bash", "-c", program], cwd=ROOT,
+                                text=True, capture_output=True, timeout=20)
+        self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+
+    def test_full_sensor_packages_are_not_legacy_partition_migrations(self):
+        migrated = {spec["target"] for spec in BOARDS.values()}
+        for target in SENSORS:
+            with self.subTest(target=target):
+                self.assertNotIn(target, migrated)
+                self.assertEqual(category(dict(manifest=manifest(target))), "full-profiles")
+
+    def test_picker_parses_distinct_slot_hardware(self):
+        profiles = catalogs.target_profiles(target + "-full-logging-ota" for target in SENSORS)
+        for slot in SLOTS:
+            profile = profiles[slot + "_sensor-full-logging-ota"]
+            with self.subTest(slot=slot):
+                self.assertEqual(profile["hardware"], slot)
+                self.assertEqual(profile["role"], "sensor")
+                self.assertEqual(profile["feature"], "full")
+
+    def test_both_mesh_america_catalogs_map_each_sensor_to_its_own_card(self):
+        rows = [manifest(target) for target in SENSORS]
+        records = [(row, "full-profiles-" + FAMILY,
+                    [Path(name) for name in row["files"] if name.endswith(".bin")])
+                   for row in rows]
+        plan = dict(version="1.17.1.9", source=SOURCE, groups=[dict(key="full-profiles")])
+        names = {"Station_G3_ESP32": "UnitEng/BQ Voyage Station G3",
+                 "Station_G3_ESP32_r2": "UnitEng/BQ Voyage Station G3 ESP32 R2"}
+        for filename in ("keymind-cascade-v1.16.0-provider.json",
+                         "keymind-cascade-logging-v1.16.0-provider.json"):
+            catalog = json.loads((ROOT / "mesh-america" / filename).read_text())
+            targets = {row["artifact_target"] for row in rows}
+            targets.update(catalogs.identity(file["name"]) for device in catalog["device"]
+                           for firmware in device["firmware"]
+                           for version in firmware["version"].values() for file in version["files"])
+            result = catalogs.generate(catalog, plan, FAMILY, records,
+                                       catalogs.target_profiles(targets), "mikecarper/MeshCore")
+            self.assertEqual(len(result["device"]), 2)
+            for slot, name in names.items():
+                with self.subTest(catalog=filename, slot=slot):
+                    card, = [device for device in result["device"] if device["name"] == name]
+                    firmware, = card["firmware"]
+                    self.assertEqual(firmware["role"], "sensor")
+                    self.assertEqual(firmware["subTitle"], slot + "_sensor-full-logging-ota - Full profile")
+                    files = firmware["version"][FAMILY]["files"]
+                    self.assertEqual({file["type"] for file in files}, {"flash-wipe", "flash-update"})
+                    self.assertTrue(all(file["name"].startswith(slot + "_sensor-") for file in files))
+
+    def test_controls_expose_qualified_ota_without_inventing_a_sensor_web_portal(self):
+        rows = [manifest(target) for target in SENSORS]
+        config = [("env:" + target, [("build_flags", self.recipes.value(
+            "env:" + target, "build_flags").splitlines())]) for target in SENSORS]
+        with tempfile.TemporaryDirectory(prefix="mesh-g3-sensor-controls-") as temporary:
+            stage = Path(temporary)
+            for group in ("companion", "full-profiles"):
+                (stage / group).mkdir()
+            (stage / "release-plan.json").write_text(json.dumps(dict(source=SOURCE, groups=[
+                dict(key="companion", tag=FAMILY), dict(key="full-profiles", tag="full-profiles-" + FAMILY)])))
+            (stage / "companion/TARGET-MANIFEST.json").write_text("[]")
+            (stage / "full-profiles/TARGET-MANIFEST.json").write_text(json.dumps(rows))
+            generated = controls.generate(stage, config)
+        self.assertEqual(set(generated["profiles"]), {target + "-full-logging" for target in SENSORS})
+        for target in SENSORS:
+            profile = generated["profiles"][target + "-full-logging"]
+            with self.subTest(target=target):
+                self.assertEqual(profile["otaRole"], "lora-receiver")
+                self.assertEqual(profile["updateMethods"], ["lora", "wifi"])
+                self.assertTrue(profile["gps"])
+                self.assertTrue(profile["display"])
+                self.assertTrue(profile["femRx"])
+                self.assertTrue(profile["femTx"])
+                self.assertFalse(profile["webconfig"])
+                self.assertNotIn(target, generated["partitionMigrations"])
+
+
+if __name__ == "__main__":
+    unittest.main()

test/test_stored_client_reply_routes.py

@@ -0,0 +1,173 @@
+#!/usr/bin/env python3
+"""Exercise actual Room CLI/Sensor ACK routes through real Mesh admission."""
+from pathlib import Path
+import shutil
+import subprocess
+import sys
+import tempfile
+import unittest
+
+from test_replay_reset_integration import extract_braced
+from test_client_acl_infrastructure import production_methods
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+class StoredClientReplyRoutesTest(unittest.TestCase):
+    def execute(self, role, old=False, old_keepalive=False):
+        compiler = shutil.which("g++") or shutil.which("clang++")
+        self.assertIsNotNone(compiler, "host C++17 compiler required")
+        sensor = role == "sensor"
+        name = "SensorMesh" if sensor else "MyMesh"
+        source = (ROOT / ("examples/simple_sensor/SensorMesh.cpp" if sensor
+                          else "examples/simple_room_server/MyMesh.cpp")).read_text()
+        if old:
+            guard = "!mesh::Packet::isValidPathLen(" + ("dest." if sensor else "client->") + "out_path_len)"
+            self.assertIn(guard, source)
+            source = source.replace(guard, ("dest." if sensor else "client->") + "out_path_len == OUT_PATH_UNKNOWN")
+        if sensor:
+            route = extract_braced(source, "void SensorMesh::sendAckTo(")
+        else:
+            ack_start = source.index("    if (send_ack) {")
+            branch = extract_braced(source[ack_start:], "if (" + (
+                "client->out_path_len == OUT_PATH_UNKNOWN" if old else
+                "!mesh::Packet::isValidPathLen(client->out_path_len)") + ")")
+            # Extract_braced returns the first if block; retain its actual
+            # matching else through the end of this send_ack branch too.
+            full = extract_braced(source, "if (send_ack) {")
+            start = full.index(branch)
+            route_body = full[start:full.rfind("}")]
+            route = """void MyMesh::sendCliAck(const ClientInfo& dest) {
+const ClientInfo* client = &dest;
+mesh::Packet incoming;
+mesh::Packet* packet = &incoming;
+uint32_t ack_hash = 7, delay_millis = 0;
+""" + route_body + "\n(void)delay_millis;\n}\n"
+        transform = None
+        if old_keepalive:
+            self.assertFalse(sensor)
+            def transform(text):
+                receive = extract_braced(text, "if (data[4] == REQ_TYPE_KEEP_ALIVE && packet->isRouteDirect())")
+                guard = "mesh::Packet::isValidPathLen(client->out_path_len)"
+                self.assertIn(guard, receive)
+                altered = receive.replace(guard, "client->out_path_len != OUT_PATH_UNKNOWN")
+                return text.replace(receive, altered, 1)
+        generated = production_methods(role, transform) + route
+        mesh_source = (ROOT / "src/Mesh.cpp").read_text()
+        generated += "namespace mesh {\n"
+        for signature in ("Packet* Mesh::createAck(const uint8_t*", "Packet* Mesh::createAck(uint32_t",
+                          "Packet* Mesh::createMultiAck(const uint8_t*", "Packet* Mesh::createMultiAck(uint32_t"):
+            generated += extract_braced(mesh_source, signature) + "\n"
+        generated += "}\n"
+        fixture = (ROOT / "test/fixtures/client_acl_infrastructure/test_client_acl_infrastructure.cpp").read_text()
+        fixture = fixture.replace("#define SERVER_RESPONSE_DELAY 300", "#define SERVER_RESPONSE_DELAY 300\n#define TXT_ACK_DELAY 300\n#define REPLY_DELAY_MILLIS 1500")
+        declaration = "  Packet* createDatagram(uint8_t, const Identity&, const uint8_t*, const uint8_t*, size_t);"
+        self.assertIn(declaration, fixture)
+        fixture = fixture.replace(declaration, declaration + """
+  Packet* createAck(const uint8_t*, uint8_t);
+  Packet* createAck(uint32_t);
+  Packet* createMultiAck(const uint8_t*, uint8_t, uint8_t);
+  Packet* createMultiAck(uint32_t, uint8_t);
+""")
+        fixture = fixture.replace("  mesh::Packet* createAck(uint32_t) { return nullptr; }", """
+  uint8_t getExtraAckTransmitCount() const { return 0; }
+#ifdef TEST_SENSOR
+  void sendAckTo(const ClientInfo&, uint32_t, uint8_t);
+#else
+  void sendCliAck(const ClientInfo&);
+#endif
+""")
+        fixture = fixture.replace("  unsigned released = 0;", "  unsigned released = 0, obtained = 0;")
+        fixture = fixture.replace("    auto* packet = new Packet;", "    ++obtained;\n    auto* packet = new Packet;")
+        fixture = fixture.replace("int main() {", "int original_main() {", 1)
+        marker = '  std::cout << "Room/Sensor ACL route, input, and permission checks passed\\n";'
+        fixture = fixture.replace(marker, marker + "\n  return 0;")
+        fixture += r'''
+int main() {
+  original_main();
+  for (uint8_t stored_path : {0, 1, 0x41, 0xFE, 0xFF, 0xC0, 0xBF}) {
+    for (int refusal = 0; refusal < 3; ++refusal) {
+      TARGET_CLASS target;
+      target.sender.out_path_len = stored_path;
+      target.refuse_allocation = refusal == 1;
+      target.refuse_queue = refusal == 2;
+#ifdef TEST_SENSOR
+      target.sendAckTo(target.sender, 7, 1);
+#else
+      target.sendCliAck(target.sender);
+#endif
+      if (refusal) { assert(target.queued == 0); continue; }
+      assert(target.queued == 1 && target.last_reply != nullptr);
+      const bool direct = mesh::Packet::isValidPathLen(stored_path);
+      assert(target.direct_sent == unsigned(direct));
+      assert(target.flood_sent == unsigned(!direct));
+      assert(target.last_reply->getPayloadType() == PAYLOAD_TYPE_ACK);
+      assert(target.last_reply->payload_len == 4);
+    }
+#ifndef TEST_SENSOR
+    for (int refusal = 0; refusal < 3; ++refusal) {
+      TARGET_CLASS target;
+      target.sender.out_path_len = stored_path;
+      target.refuse_allocation = refusal == 1;
+      target.refuse_queue = refusal == 2;
+      mesh::Packet incoming;
+      incoming.header = (PAYLOAD_TYPE_REQ << PH_TYPE_SHIFT) | ROUTE_TYPE_DIRECT;
+      uint8_t request[9] = {};
+      uint32_t timestamp = 51;
+      memcpy(request, &timestamp, sizeof(timestamp));
+      request[4] = REQ_TYPE_KEEP_ALIVE;
+      target.receive(&incoming, request, sizeof(request));
+      const bool direct = mesh::Packet::isValidPathLen(stored_path);
+      assert(target.flood_sent == 0);
+      assert(target.sender.last_timestamp == timestamp);
+      assert(target.sender.last_activity == 100);
+      if (!direct) {
+        assert(target.queued == 0 && target.obtained == 0);
+      } else if (refusal) {
+        assert(target.queued == 0);
+      } else {
+        assert(target.queued == 1 && target.direct_sent == 1);
+        assert(target.last_reply->payload_len == 5);
+      }
+    }
+#endif
+  }
+}
+'''
+        with tempfile.TemporaryDirectory(prefix="stored-reply-route-") as directory:
+            work = Path(directory)
+            (work / "production.inc").write_text(generated, encoding="ascii")
+            (work / "test.cpp").write_text(fixture, encoding="ascii")
+            binary = work / "test"
+            # Actual Mesh::createMultiAck has a pre-existing signed-size comparison.
+            # Keep other fixture warnings fatal without changing production code.
+            args = [compiler, "-std=c++17", "-Wall", "-Wextra", "-Werror", "-Wno-unused-parameter", "-Wno-sign-compare",
+                    "-I" + str(work), "-I" + str(ROOT / "src"), str(work / "test.cpp"), "-o", str(binary)]
+            if sensor: args.insert(1, "-DTEST_SENSOR=1")
+            if sys.platform.startswith("linux"):
+                args[1:1] = ["-fsanitize=address,undefined", "-fno-sanitize-recover=all", "-fno-pie", "-no-pie"]
+            built = subprocess.run(args, capture_output=True, text=True, timeout=60)
+            self.assertEqual(built.returncode, 0, built.stdout + built.stderr)
+            return subprocess.run([str(binary)], capture_output=True, text=True, timeout=30)
+
+    def test_actual_room_cli_and_sensor_ack_route_and_pool_admission(self):
+        for role in ("room", "sensor"):
+            with self.subTest(role=role):
+                run = self.execute(role)
+                self.assertEqual(run.returncode, 0, run.stdout + run.stderr)
+
+    def test_legacy_unknown_only_guard_drops_force_flood_ack(self):
+        for role in ("room", "sensor"):
+            with self.subTest(role=role):
+                self.assertNotEqual(self.execute(role, old=True).returncode, 0)
+
+    def test_legacy_keepalive_guard_attempts_invalid_direct_allocation(self):
+        self.assertNotEqual(self.execute("room", old_keepalive=True).returncode, 0)
+
+    def test_suite_is_explicitly_wired_in_ci(self):
+        self.assertIn("          python3 -B test/test_stored_client_reply_routes.py -v\n",
+                      (ROOT / ".github/workflows/run-unit-tests.yml").read_text())
+
+
+if __name__ == "__main__":
+    unittest.main()

variants/station_g3_esp32/platformio.ini

@@ -135,6 +135,20 @@ lib_deps =
   ${Station_G3_ESP32.lib_deps}
   ${esp32_ota.lib_deps}
 
+[env:Station_G3_ESP32_sensor]
+extends = Station_G3_ESP32
+build_flags =
+  ${Station_G3_ESP32.build_flags}
+  -D ADVERT_NAME='"Station G3 ESP32 Sensor"'
+  -D ADVERT_LAT=0.0
+  -D ADVERT_LON=0.0
+  -D ADMIN_PASSWORD='"password"'
+build_src_filter = ${Station_G3_ESP32.build_src_filter}
+  +<../examples/simple_sensor>
+lib_deps =
+  ${Station_G3_ESP32.lib_deps}
+  ${esp32_ota.lib_deps}
+
 [env:Station_G3_ESP32_companion_radio_usb]
 extends = Station_G3_ESP32
 build_flags =
@@ -339,6 +353,20 @@ lib_deps =
   ${Station_G3_ESP32_r2.lib_deps}
   ${esp32_ota.lib_deps}
 
+[env:Station_G3_ESP32_r2_sensor]
+extends = Station_G3_ESP32_r2
+build_flags =
+  ${Station_G3_ESP32_r2.build_flags}
+  -D ADVERT_NAME='"Station G3 ESP32 Slot 2 Sensor"'
+  -D ADVERT_LAT=0.0
+  -D ADVERT_LON=0.0
+  -D ADMIN_PASSWORD='"password"'
+build_src_filter = ${Station_G3_ESP32_r2.build_src_filter}
+  +<../examples/simple_sensor>
+lib_deps =
+  ${Station_G3_ESP32_r2.lib_deps}
+  ${esp32_ota.lib_deps}
+
 [env:Station_G3_ESP32_r2_repeater_observer_mqtt]
 extends = Station_G3_ESP32_r2
 board_build.partitions = default_16MB.csv ; standard 16MB partition table