Preserve BLE command replies and respect negotiated mOTA MTU

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

@@ -622,6 +622,7 @@ jobs:
           python3 -B test/test_local_release_capacity_archive.py
           python3 -B test/test_t096_full_memory.py
           python3 -B test/test_nrf52_ble_startup.py
+          python3 -B test/test_nrf52_ble_backpressure.py
           python3 -B test/test_nrf52_ble_trace.py
           python3 -B test/test_nrf52_ble_debug_tools.py
           python3 -B test/test_nrf52_usb_power.py

examples/companion_radio/MyMesh.cpp

@@ -25,6 +25,9 @@
 #endif
 #include <Mesh.h>
 #include <helpers/CompanionHardwareCommandCompat.h>
+#if defined(MESH_HIL_BLE_BENCH) && MESH_HIL_BLE_BENCH
+#include <helpers/CompanionBleBench.h>
+#endif
 #include <helpers/CompanionStatusResponse.h>
 #include <helpers/CompanionReader.h>
 #include <helpers/IdentityGeneration.h>
@@ -5742,6 +5745,19 @@ void MyMesh::handleCmdFrame(size_t len) {
     return;
   }
 
+#if defined(MESH_HIL_BLE_BENCH) && MESH_HIL_BLE_BENCH
+  if (cmd_frame[0] == mesh::companion::kBleBenchCommand) {
+    // Direct Companion transport benchmark only: no RF dispatch, settings,
+    // identity reads or persistent state. Reuse the normal reply queue and
+    // its required-frame admission/backpressure, including the source route.
+    const size_t echoed = mesh::companion::buildBleBenchEcho(
+        cmd_frame, len, out_frame, MAX_FRAME_SIZE);
+    if (echoed) _serial->writeFrame(out_frame, echoed);
+    else writeErrFrame(ERR_CODE_ILLEGAL_ARG);
+    return;
+  }
+#endif
+
   servicePendingSerialReply();
   const bool starts_long_lived_request =
       (cmd_frame[0] == CMD_SEND_LOGIN && len >= 1 + PUB_KEY_SIZE)

examples/companion_radio/main.cpp

@@ -469,7 +469,17 @@ public:
   }
 
 private:
-  mesh::ota::SerialMotaSource _source;
+  class Source : public mesh::ota::SerialMotaSource {
+  public:
+    using mesh::ota::SerialMotaSource::SerialMotaSource;
+  protected:
+    uint16_t responseFrameCapacity() const override {
+      return bluetooth_interface.motaResponseCapacity();
+    }
+  };
+  static_assert(sizeof(Source) == sizeof(mesh::ota::SerialMotaSource),
+                "BLE source MTU policy must not allocate additional state");
+  Source _source;
   uint32_t _packets_sent_at_start;
   uint32_t _last_packets_sent;
 };

src/helpers/CompanionBleBench.h

@@ -0,0 +1,39 @@
+#pragma once
+
+// Explicit lab images only. Normal release builds expose no echo command,
+// allocate no benchmark state and retain their existing command behavior.
+#if defined(MESH_HIL_BLE_BENCH) && MESH_HIL_BLE_BENCH
+#if MESH_HIL_BLE_BENCH != 1
+#error "MESH_HIL_BLE_BENCH must be explicitly 0 or 1"
+#endif
+
+#include <stdint.h>
+#include <stddef.h>
+#include <string.h>
+#include "CompanionFrameLimits.h"
+
+namespace mesh {
+namespace companion {
+
+// 0x78..0x7F already have legacy hardware aliases. 0x70 is lab-only and is
+// distinct from normal responses/pushes; nonce and sequence belong to the host.
+static constexpr uint8_t kBleBenchCommand = 0x70;
+static constexpr size_t kBleBenchHeaderBytes = 15;
+
+// Wire: 70, "BLE1", runIdLE32, sequenceLE32, payloadLengthLE16, payload.
+// Echo the entire request exactly once, without generic OK or radio traffic.
+// The host also limits frame size to its negotiated ATT MTU minus three.
+inline size_t buildBleBenchEcho(const uint8_t* request, size_t length,
+                               uint8_t* response, size_t capacity) {
+  if (!request || !response || length < kBleBenchHeaderBytes
+      || length > MAX_FRAME_SIZE || length > capacity
+      || request[0] != kBleBenchCommand || memcmp(request + 1, "BLE1", 4)) return 0;
+  const size_t payload_length = size_t(request[13]) | (size_t(request[14]) << 8);
+  if (payload_length != length - kBleBenchHeaderBytes) return 0;
+  memcpy(response, request, length);
+  return length;
+}
+
+}  // namespace companion
+}  // namespace mesh
+#endif

src/helpers/esp32/SerialBLEInterface.cpp

@@ -977,14 +977,38 @@ size_t SerialBLEInterface::writeFrame(const uint8_t src[], size_t len) {
   return 0;
 }
 
-#define  BLE_WRITE_MIN_INTERVAL   60
+#if defined(MESH_HIL_BLE_INTERVAL_MS)
+#if !defined(MESH_HIL_BLE_BENCH) || MESH_HIL_BLE_BENCH != 1
+#error "MESH_HIL_BLE_INTERVAL_MS requires the explicit MESH_HIL_BLE_BENCH=1 lab image"
+#endif
+#if MESH_HIL_BLE_INTERVAL_MS < 7 || MESH_HIL_BLE_INTERVAL_MS > 60
+#error "MESH_HIL_BLE_INTERVAL_MS must be in 7..60 milliseconds"
+#endif
+#define BLE_WRITE_MIN_INTERVAL MESH_HIL_BLE_INTERVAL_MS
+#else
+// Preserve the shipped pacing until a separate lab image has measured a safe
+// replacement. One value governs admission and both successful/failed sends.
+#define BLE_WRITE_MIN_INTERVAL 60
+#endif
 
 bool SerialBLEInterface::isReadBusy() const {
   return uxQueueMessagesWaiting(recv_queue) > 0;
 }
 
+bool SerialBLEInterface::hasRequiredReplyCapacity() const {
+  if (send_queue_len < FRAME_QUEUE_SIZE) return true;
+  // Match enqueueCompanionFrame's admission rule: a required response can
+  // displace best-effort traffic, but must never displace another required one.
+  for (int i = 0; i < send_queue_len; ++i) {
+    if (!mesh::companionFrameRequiresDelivery(send_queue[i].buf, send_queue[i].len))
+      return true;
+  }
+  return false;
+}
+
 bool SerialBLEInterface::isWriteBusy() const {
-  return !mesh::bleElapsedAtLeast((uint32_t)millis(), _last_write,
+  return !hasRequiredReplyCapacity()
+      || !mesh::bleElapsedAtLeast((uint32_t)millis(), _last_write,
                                   BLE_WRITE_MIN_INTERVAL);
 }
 
@@ -1052,7 +1076,11 @@ size_t SerialBLEInterface::checkRecvFrame(uint8_t dest[]) {
   }
 
   Frame frame;
-  if (deviceConnected && xQueueReceive(recv_queue, &frame, 0) == pdTRUE) {
+  // Service TX above even when full, then retain a command until its required
+  // reply can be admitted. Dequeuing while all TX slots are required causes
+  // one-shot replies (including CMD66) to be silently lost by writeFrame().
+  if (deviceConnected && hasRequiredReplyCapacity()
+      && xQueueReceive(recv_queue, &frame, 0) == pdTRUE) {
     memcpy(dest, frame.buf, frame.len);
     BLE_DEBUG_PRINTLN("readBytes: sz=%d, hdr=%d", (uint32_t) frame.len, (uint32_t) dest[0]);
     return frame.len;

src/helpers/esp32/SerialBLEInterface.h

@@ -67,6 +67,7 @@ class SerialBLEInterface : public BaseSerialInterface,
   Frame send_queue[FRAME_QUEUE_SIZE];
 
   void clearBuffers();
+  bool hasRequiredReplyCapacity() const;
   void servicePendingTxReset();
   void scheduleAdvertisingRestart(uint32_t now);
   void recoverStalledTx(const char* cause);

src/helpers/nrf52/SerialBLEInterface.cpp

@@ -1194,6 +1194,17 @@ void SerialBLEInterface::loop() {
   }
 }
 
+bool SerialBLEInterface::hasRequiredReplyCapacity() const {
+  if (send_queue_len < FRAME_QUEUE_SIZE) return true;
+  // Required responses may evict an unsent packet log, never another required
+  // reply. Match enqueueCompanionFrame without growing either bounded queue.
+  for (uint8_t i = 0; i < send_queue_len; ++i) {
+    if (!mesh::companionFrameRequiresDelivery(send_queue[i].buf, send_queue[i].len))
+      return true;
+  }
+  return false;
+}
+
 size_t SerialBLEInterface::writeFrame(const uint8_t src[], size_t len) {
   if (len > MAX_FRAME_SIZE) {
     BLE_DEBUG_PRINTLN("writeFrame(), frame too big, len=%u", (unsigned)len);
@@ -1285,7 +1296,9 @@ size_t SerialBLEInterface::checkRecvFrame(uint8_t dest[]) {
     _tx_stall_watchdog.reset();
   }
   
-  if (recv_queue_len > 0) {
+  // TX/recovery above must run even when full. Retain RX until a required
+  // command reply can be admitted instead of silently losing that reply.
+  if (recv_queue_len > 0 && hasRequiredReplyCapacity()) {
     size_t len = recv_queue[0].len;
     memcpy(dest, recv_queue[0].buf, len);
     
@@ -1366,6 +1379,17 @@ bool SerialBLEInterface::isMotaChannelReady() {
       && _conn_handle != BLE_CONN_HANDLE_INVALID
       && _mota_request.notifyEnabled(_conn_handle);
 }
+
+uint16_t SerialBLEInterface::motaResponseCapacity() {
+  if (!isMotaChannelReady() || !_mota_stream.isActive()) return 0;
+  BLEConnection* connection = Bluefruit.Connection(_conn_handle);
+  if (!connection || !connection->connected()) return 0;
+  const uint16_t mtu = connection->getMtu();
+  if (mtu < 23) return 0;
+  const uint16_t capacity = (uint16_t)(mtu - 3u);
+  return capacity < mesh::ota::BLE_MOTA_RESPONSE_MAX
+      ? capacity : mesh::ota::BLE_MOTA_RESPONSE_MAX;
+}
 #endif
 
 bool SerialBLEInterface::isReadBusy() const {

src/helpers/nrf52/SerialBLEInterface.h

@@ -109,6 +109,7 @@ class SerialBLEInterface : public BaseSerialInterface {
   Frame recv_queue[FRAME_QUEUE_SIZE];
 
   void clearBuffers();
+  bool hasRequiredReplyCapacity() const;
   void shiftSendQueueLeft();
   void shiftRecvQueueLeft();
   size_t writeBleUartFrame(const Frame& frame);
@@ -209,6 +210,7 @@ public:
 #if COMPANION_FEATURE_BLE_MOTA_SOURCE
   Stream& motaStream() { return _mota_stream; }
   bool isMotaChannelReady();
+  uint16_t motaResponseCapacity();
   bool isMotaStreamActive() const { return _mota_stream.isActive(); }
   void setMotaStreamActive(bool active) {
     _mota_stream.setActive(active);

src/helpers/ota/MotaSourceSerial.cpp

@@ -202,13 +202,21 @@ bool SerialMotaSource::describe(uint8_t idx, MotaDesc& out) {
   return true;
 }
 
+uint16_t SerialMotaSource::responsePayloadCapacity() const {
+  uint16_t frame = responseFrameCapacity();
+  if (frame < 8) return 0;  // even a nonempty DEFLATE response must fit
+  if (frame > MOTA_SEEDER_READ_MAX + 5u) frame = MOTA_SEEDER_READ_MAX + 5u;
+  return (uint16_t)(frame - 5u);
+}
+
 bool SerialMotaSource::read(uint8_t idx, uint32_t off, uint8_t* buf, uint32_t len) {
   if (!buf && len != 0) return false;
   if (len > UINT32_MAX - off) return false;
   uint32_t done = 0;
   while (done < len) {
-    uint16_t chunk = (uint16_t)(len - done > MOTA_SEEDER_READ_MAX
-        ? MOTA_SEEDER_READ_MAX : len - done);
+    const uint16_t maximum = responsePayloadCapacity();
+    if (maximum == 0) return false;  // disconnected/invalid BLE session
+    uint16_t chunk = (uint16_t)(len - done > maximum ? maximum : len - done);
     uint8_t args[7];
     args[0] = idx;
     wr_u32le(args + 1, off + done);
@@ -223,6 +231,7 @@ bool SerialMotaSource::read_deflated_block(uint8_t idx, uint16_t block, uint8_t*
                                            uint16_t cap, uint16_t* len) {
   if (!buf || !len || cap == 0) return false;
   *len = 0;
+  if (responsePayloadCapacity() < 3) return false;
 
   uint8_t args[7] = { idx };
   args[1] = (uint8_t)(block & 0xFF); args[2] = (uint8_t)(block >> 8);
@@ -235,8 +244,11 @@ bool SerialMotaSource::read_deflated_block(uint8_t idx, uint16_t block, uint8_t*
 
   uint16_t done = 0;
   while (done < total) {
-    const uint16_t chunk = (uint16_t)((total - done > MOTA_SEEDER_DEFLATE_CHUNK_MAX)
-        ? MOTA_SEEDER_DEFLATE_CHUNK_MAX : total - done);
+    const uint16_t payload_capacity = responsePayloadCapacity();
+    if (payload_capacity < 3) return false;
+    const uint16_t maximum = (uint16_t)(payload_capacity - 2u);
+    const uint16_t chunk = (uint16_t)((total - done > maximum)
+        ? maximum : total - done);
     args[3] = (uint8_t)(done & 0xFF); args[4] = (uint8_t)(done >> 8);
     args[5] = (uint8_t)(chunk & 0xFF); args[6] = (uint8_t)(chunk >> 8);
     uint8_t response[2 + MOTA_SEEDER_DEFLATE_CHUNK_MAX];

src/helpers/ota/MotaSourceSerial.h

@@ -38,7 +38,14 @@ public:
   bool hasPendingResponse() const { return _response_pending; }
   void resetSessionState();
 
+protected:
+  // Full framed response capacity, including magic/op/status/checksum. The
+  // default preserves USB/UART/TCP geometry. A BLE source can override this
+  // using its current ATT MTU without storing another buffer or callback.
+  virtual uint16_t responseFrameCapacity() const { return 197; }
+
 private:
+  uint16_t responsePayloadCapacity() const;
   // Send a request (op+args) and read its response header; on OK, `payload` (if non-null) receives
   // `payload_len` bytes. Returns true iff a well-formed OK response for `op` arrived in time.
   bool txn(uint8_t op, const uint8_t* args, uint8_t arglen, uint8_t* payload, uint32_t payload_len);

test/fixtures/nimble_companion/mocks/NimBLEDevice.h

@@ -61,6 +61,7 @@ inline uint32_t pin = 0;
 inline bool bonding = false, mitm = false, sc = false;
 inline std::vector<std::vector<uint8_t>> transmitted;
 inline unsigned disconnects = 0;
+inline unsigned notification_attempts = 0;
 }
 inline int ble_gap_wl_set(const ble_addr_t* peers, uint8_t count) {
   assert(count == 1);
@@ -91,6 +92,7 @@ struct NimBLECharacteristic {
   void setValue(const uint8_t* data, size_t size) { value.assign(data, data + size); }
   bool notify(uint16_t handle = 0xffff) const {
     assert(handle == 7);
+    ++fake_nimble::notification_attempts;
     if (!fake_nimble::accept_notification) return false;
     fake_nimble::transmitted.push_back(value);
     return true;

test/fixtures/nimble_companion/test_nimble_companion.cpp

@@ -10,7 +10,7 @@ using namespace fake_nimble;
 static void reset() {
   NimBLEDevice::deinit(true);
   failure.clear(); calls.clear(); transmitted.clear();
-  persisted = true; accept_notification = true; disconnects = 0;
+  persisted = true; accept_notification = true; disconnects = 0; notification_attempts = 0;
   test_millis = 100;
 }
 
@@ -199,11 +199,171 @@ static void test_real_transport_frames() {
   NimBLEDevice::deinit(true);
 }
 
+static void test_required_reply_backpressure_services_tx_before_rx() {
+  reset(); SerialBLEInterface interface;
+  assert(interface.begin("MeshCore-", "backpressure", 123456));
+  interface.enable();
+  NimBLEConnInfo info;
+  authenticate(interface, info);
+  auto* server = NimBLEDevice::server.get();
+  auto* tx = server->service->characteristics[0].get();
+  auto* rx = server->service->characteristics[1].get();
+  tx->callbacks->onSubscribe(tx, info, 1);
+  mtu = 179;
+  for (uint8_t i = 0; i < 4; ++i) {
+    const uint8_t reply[] = {0x0d, i};
+    assert(interface.writeFrame(reply, sizeof(reply)) == sizeof(reply));
+  }
+  const uint8_t command[] = {66, 'A', '7', '|', 'g', 'e', 't', ' ', 'l', 'a', 'b', '.', 't', 'x'};
+  rx->setValue(command, sizeof(command));
+  rx->callbacks->onWrite(rx, info);
+  uint8_t received[MAX_FRAME_SIZE] = {};
+  accept_notification = false;
+  // A command must retain its RX slot until its required reply can fit.
+  assert(interface.checkRecvFrame(received) == 0);
+  assert(interface.isReadBusy() && transmitted.empty());
+  test_millis += 61;
+  assert(interface.isWriteBusy()); // Capacity, not just the 60 ms TX throttle.
+  assert(interface.checkRecvFrame(received) == 0);
+  assert(interface.isReadBusy() && transmitted.empty());
+
+  accept_notification = true;
+  test_millis += 61;
+  // TX must be serviced BEFORE the full-queue RX gate, or this deadlocks.
+  assert(interface.checkRecvFrame(received) == sizeof(command));
+  assert(memcmp(received, command, sizeof(command)) == 0);
+  assert(!interface.isReadBusy() && transmitted.size() == 1);
+  const uint8_t cli_reply[] = {29, 'A', '7', '|', 'O', 'K'};
+  assert(interface.writeFrame(cli_reply, sizeof(cli_reply)) == sizeof(cli_reply));
+  for (unsigned i = 0; i < 6; ++i) {
+    test_millis += 61;
+    assert(interface.checkRecvFrame(received) == 0);
+  }
+  assert(transmitted.size() == 5);
+  for (uint8_t i = 0; i < 4; ++i) {
+    assert((transmitted[i] == std::vector<uint8_t>{0x0d, i}));
+  }
+  assert(transmitted[4] == std::vector<uint8_t>(cli_reply, cli_reply + sizeof(cli_reply)));
+  assert(!interface.hasPendingIO());
+  NimBLEDevice::deinit(true);
+}
+
+static void test_full_queue_best_effort_eviction_still_admits_rx() {
+  reset(); SerialBLEInterface interface;
+  assert(interface.begin("MeshCore-", "eviction", 123456));
+  interface.enable();
+  NimBLEConnInfo info;
+  authenticate(interface, info);
+  auto* server = NimBLEDevice::server.get();
+  auto* tx = server->service->characteristics[0].get();
+  auto* rx = server->service->characteristics[1].get();
+  tx->callbacks->onSubscribe(tx, info, 1);
+  mtu = 179;
+  for (uint8_t i = 0; i < 3; ++i) {
+    const uint8_t log[] = {0x88, i};
+    assert(interface.writeFrame(log, sizeof(log)) == sizeof(log));
+  }
+  const uint8_t extra_log[] = {0x88, 9};
+  assert(interface.writeFrame(extra_log, sizeof(extra_log)) == 0);
+  const uint8_t previous_reply[] = {0x0d, 7};
+  assert(interface.writeFrame(previous_reply, sizeof(previous_reply)) == sizeof(previous_reply));
+  const uint8_t command[] = {66, 'B', '8', '|', 'v', 'e', 'r'};
+  rx->setValue(command, sizeof(command));
+  rx->callbacks->onWrite(rx, info);
+  uint8_t received[MAX_FRAME_SIZE] = {};
+  accept_notification = false;
+  // Full is not blocked when a required response can evict a packet log.
+  assert(!interface.isWriteBusy());
+  assert(interface.checkRecvFrame(received) == sizeof(command));
+  assert(memcmp(received, command, sizeof(command)) == 0);
+  const uint8_t cli_reply[] = {29, 'B', '8', '|', 'O', 'K'};
+  assert(interface.writeFrame(cli_reply, sizeof(cli_reply)) == sizeof(cli_reply));
+  assert(transmitted.empty());
+  accept_notification = true;
+  for (unsigned i = 0; i < 6; ++i) {
+    test_millis += 61;
+    assert(interface.checkRecvFrame(received) == 0);
+  }
+  assert(transmitted.size() == 4);
+  assert(transmitted[0] == std::vector<uint8_t>(previous_reply, previous_reply + sizeof(previous_reply)));
+  assert(transmitted[1] == std::vector<uint8_t>(cli_reply, cli_reply + sizeof(cli_reply)));
+  assert((transmitted[2] == std::vector<uint8_t>{0x88, 0}));
+  assert((transmitted[3] == std::vector<uint8_t>{0x88, 1}));
+  assert(!interface.hasPendingIO());
+  NimBLEDevice::deinit(true);
+}
+
+static void test_exact_notify_interval_failure_retention_and_rollover() {
+#if defined(MESH_HIL_BLE_INTERVAL_MS)
+  constexpr uint32_t interval = MESH_HIL_BLE_INTERVAL_MS;
+#else
+  constexpr uint32_t interval = 60;
+#endif
+  reset(); SerialBLEInterface interface;
+  assert(interface.begin("MeshCore-", "interval", 123456));
+  interface.enable();
+  NimBLEConnInfo info; authenticate(interface, info);
+  auto* tx = NimBLEDevice::server->service->characteristics[0].get();
+  uint8_t first[MAX_FRAME_SIZE] = {0x70, 1}, second[MAX_FRAME_SIZE] = {0x70, 2};
+  uint8_t received[MAX_FRAME_SIZE] = {};
+  assert(interface.writeFrame(first, sizeof(first)) == sizeof(first));
+  interface.checkRecvFrame(received);
+  assert(notification_attempts == 0); // CCCD is not ready.
+  tx->callbacks->onSubscribe(tx, info, 1);
+  mtu = 23; interface.checkRecvFrame(received);
+  assert(notification_attempts == 0); // Never truncate to fit a smaller MTU.
+  mtu = 179; accept_notification = false;
+  interface.checkRecvFrame(received);
+  assert(notification_attempts == 1 && transmitted.empty() && interface.hasPendingIO());
+  test_millis += interval - 1;
+  assert(interface.isWriteBusy());
+  interface.checkRecvFrame(received);
+  assert(notification_attempts == 1); // Failed attempts must also be paced.
+  ++test_millis;
+  assert(!interface.isWriteBusy());
+  accept_notification = true; interface.checkRecvFrame(received);
+  assert(notification_attempts == 2 && transmitted.size() == 1);
+  assert(transmitted[0] == std::vector<uint8_t>(first, first + sizeof(first)));
+  assert(!interface.hasPendingIO()); // One accepted copy, no retry/late-status duplicate.
+  assert(interface.writeFrame(second, sizeof(second)) == sizeof(second));
+  test_millis += interval - 1; interface.checkRecvFrame(received);
+  assert(notification_attempts == 2 && transmitted.size() == 1);
+  ++test_millis; interface.checkRecvFrame(received);
+  assert(notification_attempts == 3 && transmitted.size() == 2);
+  assert(transmitted[1] == std::vector<uint8_t>(second, second + sizeof(second)));
+  // Real adapter arithmetic must remain correct when millis wraps at UINT32.
+  test_millis = UINT32_MAX - 7;
+  assert(interface.writeFrame(first, sizeof(first)) == sizeof(first));
+  interface.checkRecvFrame(received);
+  assert(notification_attempts == 4);
+  assert(interface.writeFrame(second, sizeof(second)) == sizeof(second));
+  test_millis += interval - 1;
+  assert(interface.isWriteBusy()); interface.checkRecvFrame(received);
+  assert(notification_attempts == 4);
+  ++test_millis;
+  assert(!interface.isWriteBusy()); interface.checkRecvFrame(received);
+  assert(notification_attempts == 5 && transmitted.size() == 4);
+  interface.checkRecvFrame(received);
+  assert(notification_attempts == 5 && !interface.hasPendingIO());
+  // A faster lab pace must not shorten or remove the existing stall recovery.
+  assert(interface.writeFrame(first, sizeof(first)) == sizeof(first));
+  accept_notification = false; test_millis += interval;
+  interface.checkRecvFrame(received);
+  test_millis += 9999; interface.checkRecvFrame(received);
+  assert(disconnects == 0 && interface.hasPendingIO());
+  test_millis += interval; interface.checkRecvFrame(received);
+  assert(disconnects == 1 && !interface.isConnected());
+  NimBLEDevice::deinit(true);
+}
+
 int main() {
   test_identity_modes();
   test_pairing_wake_for_random_and_saved_pins();
   test_startup_failures();
   test_authentication_and_bond_lifecycle();
   test_real_transport_frames();
+  test_required_reply_backpressure_services_tx_before_rx();
+  test_full_queue_best_effort_eviction_still_admits_rx();
+  test_exact_notify_interval_failure_retention_and_rollover();
   puts("NimBLE adapter: identity, faults, bonding, stealth and full frames PASS");
 }

test/test_ble_visibility_probe.py

@@ -0,0 +1,69 @@
+import importlib.util
+from pathlib import Path
+import unittest
+
+SPEC = importlib.util.spec_from_file_location("ble_visibility_probe", Path(__file__).resolve().parents[1] / "tools/hil/ble_visibility_probe.py")
+probe = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(probe)
+TARGET = "D2:90:05:46:31:01"
+
+
+class VisibilityTests(unittest.TestCase):
+    def test_info_not_available_is_not_a_cached_device(self):
+        for code in (0, 1):
+            self.assertFalse(probe.is_cached_device(f"Device {TARGET} not available\n", TARGET, code))
+        self.assertFalse(probe.is_cached_device(f"Device {TARGET}\n", TARGET, 1))
+        self.assertFalse(probe.is_cached_device(f"Name: Device {TARGET}\n", TARGET, 0))
+
+    def test_successful_info_requires_exact_header(self):
+        for header in (f"Device {TARGET}", f"Device {TARGET} (random)", f"Device {TARGET} (public)"):
+            self.assertTrue(probe.is_cached_device(f"\x1b[0;92m{header}\x1b[0m\n\tName: test\n", TARGET, 0))
+
+    def test_cache_query_errors_are_not_verified_absence(self):
+        self.assertIsNone(probe.device_cache_state("org.bluez failed", TARGET, 1))
+        self.assertIsNone(probe.device_cache_state(f"Device {TARGET}\n", TARGET, 1))
+        self.assertIsNone(probe.device_cache_state(f"Device {TARGET}\nDevice {TARGET} not available\n", TARGET, 0))
+        self.assertIs(probe.device_cache_state(f"Device {TARGET} not available\n", TARGET, 1), False)
+
+    def test_new_cache_record_is_evidence_without_fabricated_events_or_rssi(self):
+        scan = probe.parse_discovery(f"[NEW] Device {TARGET} name\nDiscovery started\n", TARGET)
+        result = probe.with_cache_evidence(scan, False, True)
+        self.assertTrue(result["advertisement_evidence"])
+        self.assertTrue(result["new_target_record_during_scan"])
+        self.assertFalse(result["post_start_event_evidence"])
+        self.assertEqual(result["target_events"], 0)
+        self.assertEqual(result["rssi_dbm"], [])
+        self.assertEqual(result["evidence_sources"], ["absent_before_present_after_successful_discovery"])
+
+    def test_existing_or_unknown_cache_and_failed_discovery_not_new_evidence(self):
+        scan = probe.parse_discovery("Discovery started\n", TARGET)
+        for before, after in ((True, True), (None, True), (False, None), (False, False)):
+            result = probe.with_cache_evidence(scan, before, after)
+            self.assertFalse(result["advertisement_evidence"])
+        failed = probe.parse_discovery("Failed to start discovery\n", TARGET)
+        self.assertFalse(probe.with_cache_evidence(failed, False, True)["advertisement_evidence"])
+
+    def test_fresh_rssi_remains_separate_from_cache_transition(self):
+        scan = probe.parse_discovery(f"Discovery started\n[CHG] Device {TARGET} RSSI: -40\n", TARGET)
+        result = probe.with_cache_evidence(scan, True, True)
+        self.assertTrue(result["post_start_event_evidence"])
+        self.assertFalse(result["new_target_record_during_scan"])
+        self.assertEqual(result["evidence_sources"], ["post_start_rssi_update"])
+
+    def test_cached_enumeration_is_not_actual_discovery(self):
+        result = probe.parse_discovery(f"[NEW] Device {TARGET} cached\nDiscovery started\n", TARGET)
+        self.assertFalse(result["advertisement_evidence"])
+        self.assertEqual(result["target_events"], 0)
+
+    def test_filters_other_devices_and_parses_target_rssi(self):
+        result = probe.parse_discovery(f"Discovery started\n[CHG] Device AA:00:00:00:00:01 RSSI: -12\n\x1b[0;92m[CHG]\x1b[0m Device {TARGET} RSSI: -54\n[CHG] Device {TARGET} RSSI: 0xffffffcb (-53)\n", TARGET)
+        self.assertTrue(result["advertisement_evidence"])
+        self.assertEqual(result["rssi_dbm"], [-54, -53])
+        self.assertEqual(result["target_events"], 2)
+
+    def test_scan_start_failure_does_not_use_stale_events(self):
+        self.assertFalse(probe.parse_discovery(f"Failed to start discovery\n[NEW] Device {TARGET} cached\n", TARGET)["advertisement_evidence"])
+
+
+if __name__ == "__main__":
+    unittest.main()

test/test_companion_ble_bench.py

@@ -0,0 +1,181 @@
+"""Actual lab echo/dispatch boundary; no BLE, serial hardware or radio."""
+from pathlib import Path
+import re
+import shutil
+import subprocess
+import tempfile
+import unittest
+
+from test_replay_reset_integration import extract_braced
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+class CompanionBleBenchTest(unittest.TestCase):
+    def compile(self, source, *flags, expect_failure=False):
+        compiler = shutil.which("c++") or shutil.which("g++")
+        if not compiler:
+            self.skipTest("a native C++ compiler is required")
+        with tempfile.TemporaryDirectory(prefix="mesh-ble-bench-") as folder:
+            cpp, binary = Path(folder) / "test.cpp", Path(folder) / "test"
+            cpp.write_text(source)
+            result = subprocess.run([compiler, "-std=c++11", "-Wall", "-Wextra", "-Werror",
+                "-fsanitize=address,undefined", "-fno-sanitize-recover=all", "-fno-pie", "-no-pie",
+                "-I", str(ROOT / "src"), *flags, str(cpp), "-o", str(binary)],
+                capture_output=True, text=True, timeout=30)
+            if expect_failure:
+                self.assertNotEqual(result.returncode, 0)
+                return result.stderr
+            self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+            run = subprocess.run([str(binary)], capture_output=True, text=True, timeout=10)
+            self.assertEqual(run.returncode, 0, run.stdout + run.stderr)
+
+    def test_production_echo_exact_binary_maximum_and_malformed_bounds(self):
+        self.compile(r'''
+#include <helpers/CompanionBleBench.h>
+#include <cassert>
+#include <array>
+#include <cstring>
+using namespace mesh::companion;
+int main() {
+  static_assert(kBleBenchCommand == 0x70 && kBleBenchHeaderBytes == 15, "wire contract");
+  std::array<uint8_t, MAX_FRAME_SIZE> request{};
+  std::array<uint8_t, MAX_FRAME_SIZE + 2> output{};
+  for (size_t total : {size_t(15), size_t(20), size_t(64), size_t(172), size_t(176)}) {
+    request[0] = kBleBenchCommand; memcpy(request.data() + 1, "BLE1", 4);
+    for (size_t i = 5; i < total; ++i) request[i] = uint8_t(i * 71);
+    const size_t payload = total - kBleBenchHeaderBytes;
+    request[13] = payload; request[14] = payload >> 8;
+    output.fill(0xa5);
+    assert(buildBleBenchEcho(request.data(), total, output.data() + 1, total) == total);
+    assert(!memcmp(request.data(), output.data() + 1, total));
+    assert(output[0] == 0xa5 && output[total + 1] == 0xa5);
+    output.fill(0xa5);
+    assert(buildBleBenchEcho(request.data(), total, output.data(), total - 1) == 0);
+    for (uint8_t value : output) assert(value == 0xa5);
+  }
+  // Full binary byte range, including NUL/FEND/FESC/FF, is neither escaped nor
+  // interpreted. Two sequences cover all possible byte values in payloads.
+  for (unsigned part : {0U, 128U}) {
+    const size_t total = 143;
+    request[13] = 128; request[14] = 0;
+    for (size_t i = 0; i < 128; ++i) request[15 + i] = uint8_t(part + i);
+    assert(buildBleBenchEcho(request.data(), total, output.data(), output.size()) == total);
+    assert(!memcmp(request.data(), output.data(), total));
+  }
+  for (size_t total = 0; total < kBleBenchHeaderBytes; ++total)
+    assert(buildBleBenchEcho(request.data(), total, output.data(), output.size()) == 0);
+  assert(buildBleBenchEcho(nullptr, 15, output.data(), output.size()) == 0);
+  assert(buildBleBenchEcho(request.data(), 15, nullptr, 176) == 0);
+  assert(buildBleBenchEcho(request.data(), 177, output.data(), output.size()) == 0);
+  request[13] = 0; request[14] = 0;
+  for (unsigned fault = 0; fault < 6; ++fault) {
+    auto bad = request;
+    if (fault == 0) bad[0] ^= 1;
+    if (fault == 1) bad[1] ^= 1;
+    if (fault == 2) bad[4] = '2';
+    if (fault == 3) bad[13] = 1;
+    if (fault == 4) bad[14] = 1;
+    if (fault == 5) bad[13] = bad[14] = 255;
+    assert(buildBleBenchEcho(bad.data(), 15, output.data(), output.size()) == 0);
+  }
+}
+''', "-DMESH_HIL_BLE_BENCH=1")
+
+    def test_real_dispatch_returns_once_before_rf_or_pending_operation_service(self):
+        source = (ROOT / "examples/companion_radio/MyMesh.cpp").read_text()
+        dispatch = extract_braced(source,
+            "if (cmd_frame[0] == mesh::companion::kBleBenchCommand)")
+        handler = extract_braced(source, "void MyMesh::handleCmdFrame(")
+        self.assertLess(handler.index(dispatch), handler.index("servicePendingSerialReply();"))
+        self.assertRegex(handler, r"#if defined\(MESH_HIL_BLE_BENCH\) && MESH_HIL_BLE_BENCH\s+"
+                                 r"if \(cmd_frame\[0\] == mesh::companion::kBleBenchCommand\)")
+        for forbidden in ("sendDirect", "sendPacket", "self_id", "savePrefs", "_radio", "new ", "malloc"):
+            self.assertNotIn(forbidden, dispatch)
+        self.compile(r'''
+#include <helpers/CompanionBleBench.h>
+#include <cassert>
+#include <vector>
+static const uint8_t ERR_CODE_ILLEGAL_ARG = 6;
+struct Serial {
+  bool busy = false;
+  unsigned calls = 0;
+  std::vector<uint8_t> frame;
+  size_t writeFrame(const uint8_t* bytes, size_t count) {
+    ++calls;
+    if (busy) return 0; // No retry loop, queue growth or fallback broadcast.
+    frame.assign(bytes, bytes + count); return count;
+  }
+};
+struct Handler {
+  uint8_t cmd_frame[MAX_FRAME_SIZE + 1]{}, out_frame[MAX_FRAME_SIZE + 1]{};
+  Serial serial; Serial* _serial = &serial;
+  unsigned errors = 0, normal_path = 0;
+  void writeErrFrame(uint8_t code) { assert(code == 6); ++errors; }
+  void run(size_t len) {
+    @DISPATCH@
+    ++normal_path;
+  }
+};
+int main() {
+  Handler h; h.cmd_frame[0] = 0x70; memcpy(h.cmd_frame + 1, "BLE1", 4);
+  h.run(15);
+  assert(h.serial.calls == 1 && h.serial.frame.size() == 15 && h.normal_path == 0 && h.errors == 0);
+  h.serial.busy = true; h.run(15);
+  assert(h.serial.calls == 2 && h.normal_path == 0 && h.errors == 0);
+  h.run(177);
+  assert(h.serial.calls == 2 && h.errors == 1 && h.normal_path == 0);
+  h.cmd_frame[13] = 1; h.run(15);
+  assert(h.serial.calls == 2 && h.errors == 2 && h.normal_path == 0);
+  h.cmd_frame[0] = 0x05; h.run(15);
+  assert(h.normal_path == 1 && h.serial.calls == 2 && h.errors == 2);
+}
+'''.replace("@DISPATCH@", dispatch), "-DMESH_HIL_BLE_BENCH=1")
+
+    def test_release_zero_effect_and_invalid_flag_rejected(self):
+        source = "#include <helpers/CompanionBleBench.h>\nint main() {}\n"
+        self.compile(source)
+        self.compile(source, "-DMESH_HIL_BLE_BENCH=0")
+        failure = self.compile(source, "-DMESH_HIL_BLE_BENCH=2", expect_failure=True)
+        self.assertIn("must be explicitly 0 or 1", failure)
+        # No echo symbols/constants exist at all in normal release compilation.
+        failure = self.compile(source.replace("int main() {}", "int main() { return mesh::companion::kBleBenchCommand; }"),
+                               expect_failure=True)
+        self.assertIn("mesh", failure)
+
+    def test_echo_uses_existing_required_response_route_and_queue_admission(self):
+        self.compile(r'''
+#include <helpers/CompanionBleBench.h>
+#include <helpers/CompanionFrameQueue.h>
+#include <cassert>
+struct Frame { uint8_t buf[MAX_FRAME_SIZE]{}; size_t len = 0; };
+int main() {
+  const uint8_t echo[] = {0x70, 'B','L','E','1', 1,0,0,0, 2,0,0,0, 0,0};
+  assert(mesh::companionFrameRequiresDelivery(echo, sizeof(echo)));
+  assert(mesh::companionFrameUsesRequesterRoute(echo, sizeof(echo)));
+  Frame queue[2]; size_t count = 0;
+  assert(mesh::enqueueCompanionFrame(queue, count, 2, echo, sizeof(echo)));
+  assert(mesh::enqueueCompanionFrame(queue, count, 2, echo, sizeof(echo)));
+  assert(count == 2);
+  assert(!mesh::enqueueCompanionFrame(queue, count, 2, echo, sizeof(echo)));
+  assert(count == 2); // Never grows or evicts another required response.
+  const uint8_t log[] = {0x88, 0};
+  count = 0;
+  assert(mesh::enqueueCompanionFrame(queue, count, 2, log, sizeof(log)));
+  assert(mesh::enqueueCompanionFrame(queue, count, 2, echo, sizeof(echo)));
+  assert(mesh::enqueueCompanionFrame(queue, count, 2, echo, sizeof(echo)));
+  assert(count == 2 && queue[0].buf[0] == 0x70 && queue[1].buf[0] == 0x70);
+}
+''', "-DMESH_HIL_BLE_BENCH=1")
+
+    def test_lab_opcode_does_not_collide_with_shipped_protocol_commands(self):
+        source = (ROOT / "examples/companion_radio/MyMesh.cpp").read_text()
+        for token in re.findall(r"^#define CMD_\w+\s+(0[xX][0-9a-fA-F]+|[0-9]+)", source, re.MULTILINE):
+            self.assertNotEqual(int(token, 0), 0x70)
+        for name in ("CompanionHardwareCommandCompat.h", "CompanionMotaControl.h"):
+            source = (ROOT / "src/helpers" / name).read_text()
+            self.assertNotRegex(source, r"=\s*(?:0x70|112)\s*;")
+
+
+if __name__ == "__main__":
+    unittest.main()

test/test_mota_serial_deadline.py

@@ -5,12 +5,81 @@ from pathlib import Path
 import subprocess
 import tempfile
 import unittest
+from test_t096_full_memory import method
 
 
 ROOT = Path(__file__).resolve().parents[1]
 
 
 class MotaSerialDeadlineTests(unittest.TestCase):
+    def test_actual_nrf_ble_capacity_uses_current_ready_connection(self):
+        serial = (ROOT / "src/helpers/nrf52/SerialBLEInterface.cpp").read_text()
+        getter = method(serial, "uint16_t SerialBLEInterface::motaResponseCapacity()")
+        harness = r'''
+#include <cassert>
+#include <cstdint>
+#include "helpers/BleMotaProtocol.h"
+struct BLEConnection {
+  bool online = true;
+  uint16_t mtu = 247;
+  bool connected() const { return online; }
+  uint16_t getMtu() const { return mtu; }
+};
+struct {
+  BLEConnection connection;
+  bool available = true;
+  uint16_t last_handle = 0;
+  BLEConnection* Connection(uint16_t handle) {
+    last_handle = handle;
+    return available ? &connection : nullptr;
+  }
+} Bluefruit;
+struct SerialBLEInterface {
+  bool ready = true;
+  uint16_t _conn_handle = 77;
+  struct { bool active = true; bool isActive() const { return active; } } _mota_stream;
+  bool isMotaChannelReady() const { return ready; }
+  uint16_t motaResponseCapacity();
+};
+@GETTER@
+int main() {
+  SerialBLEInterface port;
+  for (uint16_t mtu : {uint16_t(247), uint16_t(185), uint16_t(179), uint16_t(23)}) {
+    Bluefruit.connection.mtu = mtu;
+    const uint16_t expected = mtu == 247 ? 197 : mtu - 3;
+    assert(port.motaResponseCapacity() == expected);
+    assert(Bluefruit.last_handle == 77);
+  }
+  Bluefruit.connection.mtu = 22;
+  assert(port.motaResponseCapacity() == 0);
+  Bluefruit.connection.mtu = 247;
+  Bluefruit.connection.online = false;
+  assert(port.motaResponseCapacity() == 0);
+  Bluefruit.connection.online = true;
+  Bluefruit.available = false;
+  assert(port.motaResponseCapacity() == 0);
+  Bluefruit.available = true;
+  port.ready = false;
+  assert(port.motaResponseCapacity() == 0);
+  port.ready = true;
+  port._mota_stream.active = false;
+  assert(port.motaResponseCapacity() == 0);
+}
+'''.replace("@GETTER@", getter).replace("#include <cstdint>",
+                                          "#include <cstdint>\n#include <initializer_list>")
+        with tempfile.TemporaryDirectory() as directory:
+            cpp = Path(directory) / "capacity.cpp"
+            binary = Path(directory) / "capacity"
+            cpp.write_text(harness)
+            built = subprocess.run([
+                "g++", "-std=c++17", "-fsanitize=address,undefined",
+                "-fno-sanitize-recover=all", "-fno-pie", "-no-pie",
+                "-I", str(ROOT / "src"), str(cpp), "-o", str(binary)],
+                capture_output=True, text=True)
+            self.assertEqual(built.returncode, 0, built.stdout + built.stderr)
+            result = subprocess.run([str(binary)], capture_output=True, text=True)
+            self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+
     def test_actual_native_serial_cases(self):
         native = (ROOT / "test/test_ota/test_ota_core.cpp").read_text()
         start = native.index("class FakeMotaSeederStream")
@@ -32,6 +101,98 @@ using namespace mesh::ota;
 '''
         source += native[start:end]
         source += r'''
+class MtuMotaSource : public SerialMotaSource {
+public:
+  using SerialMotaSource::SerialMotaSource;
+  uint16_t frame_capacity = 197;
+  mutable unsigned checks = 0;
+  unsigned disconnect_after = 0;
+protected:
+  uint16_t responseFrameCapacity() const override {
+    if (disconnect_after && ++checks >= disconnect_after) return 0;
+    return frame_capacity;
+  }
+};
+
+TEST(MotaSourceMtu, RawRequestsFitOneActualAttPayload) {
+  for (uint16_t mtu : {247, 185, 179, 23}) {
+    resetArduinoMock();
+    FakeMotaSeederStream stream;
+    MtuMotaSource source(stream, MotaStreamWritePolicy::NoFlush);
+    source.frame_capacity = std::min<uint16_t>(197, mtu - 3);
+    std::array<uint8_t, 2048> bytes{};
+    ASSERT_TRUE(source.read(0, 99, bytes.data(), bytes.size()));
+    const uint16_t limit = source.frame_capacity - 5;
+    EXPECT_EQ(stream.lengths.size(), (bytes.size() + limit - 1) / limit);
+    uint32_t offset = 99;
+    for (size_t i = 0; i < stream.lengths.size(); ++i) {
+      EXPECT_EQ(stream.offsets[i], offset);
+      EXPECT_LE(stream.lengths[i] + 5u, source.frame_capacity);
+      offset += stream.lengths[i];
+    }
+    for (size_t i = 0; i < bytes.size(); ++i) EXPECT_EQ(bytes[i], uint8_t(99 + i));
+    if (mtu == 247) EXPECT_EQ(stream.lengths.size(), 11u);
+    if (mtu == 185) EXPECT_EQ(stream.lengths.size(), 12u);
+    if (mtu == 179) EXPECT_EQ(stream.lengths.size(), 12u);
+  }
+}
+
+TEST(MotaSourceMtu, DeflateRequestsIncludeSizeAndFrameOverheads) {
+  for (uint16_t mtu : {247, 185, 179, 23}) {
+    resetArduinoMock();
+    FakeDeflateMotaSeederStream stream;
+    stream.encoded.resize(2048);
+    for (size_t i = 0; i < stream.encoded.size(); ++i) stream.encoded[i] = uint8_t(i * 17);
+    MtuMotaSource source(stream, MotaStreamWritePolicy::NoFlush);
+    source.frame_capacity = std::min<uint16_t>(197, mtu - 3);
+    std::array<uint8_t, 2048> bytes{};
+    uint16_t length = 0;
+    ASSERT_TRUE(source.read_deflated_block(0, 7, bytes.data(), bytes.size(), &length));
+    const uint16_t limit = source.frame_capacity - 7;
+    EXPECT_EQ(stream.lengths.size(), 1u + (bytes.size() + limit - 1) / limit);
+    EXPECT_EQ(stream.lengths[0], 0u); // bounded size query
+    uint16_t offset = 0;
+    for (size_t i = 1; i < stream.lengths.size(); ++i) {
+      EXPECT_EQ(stream.offsets[i], offset);
+      EXPECT_LE(stream.lengths[i] + 7u, source.frame_capacity);
+      offset += stream.lengths[i];
+    }
+    EXPECT_EQ(length, bytes.size());
+    EXPECT_EQ(0, memcmp(bytes.data(), stream.encoded.data(), bytes.size()));
+  }
+}
+
+TEST(MotaSourceMtu, InvalidOrDisconnectedCapacityFailsBeforeNextRequest) {
+  for (uint16_t invalid : {0, 1, 5, 7}) {
+    FakeMotaSeederStream stream;
+    MtuMotaSource source(stream, MotaStreamWritePolicy::NoFlush);
+    source.frame_capacity = invalid;
+    uint8_t byte = 0;
+    EXPECT_FALSE(source.read(0, 0, &byte, 1));
+    EXPECT_TRUE(stream.lengths.empty());
+    uint16_t length = 99;
+    EXPECT_FALSE(source.read_deflated_block(0, 0, &byte, 1, &length));
+    EXPECT_EQ(length, 0u);
+    EXPECT_TRUE(stream.lengths.empty());
+  }
+  FakeMotaSeederStream stream;
+  MtuMotaSource source(stream, MotaStreamWritePolicy::NoFlush);
+  source.disconnect_after = 2;
+  std::array<uint8_t, 1024> bytes{};
+  EXPECT_FALSE(source.read(0, 0, bytes.data(), bytes.size()));
+  EXPECT_EQ(stream.lengths, (std::vector<uint16_t>{192}));
+}
+
+TEST(MotaSourceMtu, OversizedCapacityCannotIncreaseLegacyTransportMaximum) {
+  FakeMotaSeederStream stream;
+  MtuMotaSource source(stream, MotaStreamWritePolicy::NoFlush);
+  source.frame_capacity = UINT16_MAX;
+  std::array<uint8_t, 1024> bytes{};
+  ASSERT_TRUE(source.read(0, 0, bytes.data(), bytes.size()));
+  EXPECT_EQ(stream.lengths, (std::vector<uint16_t>{192, 192, 192, 192, 192, 64}));
+}
+'''
+        source += r'''
 int main(int argc, char** argv) {
   ::testing::InitGoogleTest(&argc, argv);
   return RUN_ALL_TESTS();

test/test_nimble_companion.py

@@ -10,11 +10,37 @@ ROOT = Path(__file__).resolve().parents[1]
 
 class NimbleCompanionTests(unittest.TestCase):
     def test_identity_security_and_data_path_under_sanitizers(self):
+        self.run_adapter()
+
+    def test_lab15_interval_real_adapter_under_sanitizers(self):
+        self.run_adapter("-DMESH_HIL_BLE_BENCH=1", "-DMESH_HIL_BLE_INTERVAL_MS=15")
+
+    def test_fast_interval_is_lab_only_and_bounded(self):
+        invalid = (
+            ("-DMESH_HIL_BLE_INTERVAL_MS=15",),
+            ("-DMESH_HIL_BLE_BENCH=0", "-DMESH_HIL_BLE_INTERVAL_MS=15"),
+            ("-DMESH_HIL_BLE_BENCH=2", "-DMESH_HIL_BLE_INTERVAL_MS=15"),
+            ("-DMESH_HIL_BLE_BENCH=1", "-DMESH_HIL_BLE_INTERVAL_MS=6"),
+            ("-DMESH_HIL_BLE_BENCH=1", "-DMESH_HIL_BLE_INTERVAL_MS=61"),
+        )
+        for flags in invalid:
+            with self.subTest(flags=flags):
+                result = subprocess.run([
+                    "c++", "-std=c++17", "-fsyntax-only", "-DMESH_USE_NIMBLE_ARDUINO=1", *flags,
+                    "-I", str(ROOT / "test/fixtures/nimble_companion/mocks"),
+                    "-I", str(ROOT / "src"),
+                    str(ROOT / "src/helpers/esp32/SerialBLEInterface.cpp"),
+                ], text=True, capture_output=True, timeout=30)
+                self.assertNotEqual(result.returncode, 0)
+                self.assertIn("MESH_HIL_BLE_INTERVAL_MS", result.stderr)
+
+    def run_adapter(self, *flags):
         with tempfile.TemporaryDirectory(prefix="meshcore-nimble-") as temp:
             binary = Path(temp) / "nimble-companion"
             result = subprocess.run([
                 "c++", "-std=c++17", "-g", "-fsanitize=address,undefined",
                 "-DMESH_USE_NIMBLE_ARDUINO=1",
+                *flags,
                 "-I", str(ROOT / "test/fixtures/nimble_companion/mocks"),
                 "-I", str(ROOT / "src"),
                 str(ROOT / "src/helpers/esp32/SerialBLEInterface.cpp"),

test/test_nrf52_ble_backpressure.py

@@ -0,0 +1,199 @@
+#!/usr/bin/env python3
+"""Execute production nRF52 BLE queue/dispatch methods with simulated UART IO.
+
+This covers real admission, RX-dequeue, retry-throttle and recovery ordering,
+not Bluefruit/SoftDevice callbacks or physical ATT delivery.
+"""
+from pathlib import Path
+import os
+import subprocess
+import tempfile
+import unittest
+
+from test_t096_full_memory import method
+
+ROOT = Path(__file__).resolve().parents[1]
+SOURCE = ROOT / "src/helpers/nrf52/SerialBLEInterface.cpp"
+
+HARNESS = r'''
+#include <cassert>
+#include <cstdint>
+#include <cstring>
+#include <vector>
+#include <helpers/CompanionFrameQueue.h>
+#include <helpers/BleTxStallWatchdog.h>
+
+#define MAX_FRAME_SIZE 176
+#define FRAME_QUEUE_SIZE 12
+#define BLE_RETRY_THROTTLE_MS 250
+#define BLE_DEBUG_PRINTLN(...) do {} while (0)
+#define meshBleTrace(...) do {} while (0)
+static uint32_t current_millis = 100;
+uint32_t millis() { return current_millis; }
+
+struct SerialBLEInterface {
+  struct Frame { uint8_t len = 0; uint8_t buf[MAX_FRAME_SIZE] = {}; };
+  uint8_t send_queue_len = 0, recv_queue_len = 0;
+  Frame send_queue[FRAME_QUEUE_SIZE], recv_queue[FRAME_QUEUE_SIZE];
+  unsigned long _last_retry_attempt = 0;
+  mesh::BleTxStallWatchdog _tx_stall_watchdog;
+  mesh::BleDisconnectRecovery _tx_disconnect_recovery;
+  bool connected = true, allow_write = false, partial_write = false;
+  unsigned write_attempts = 0, recovery_services = 0, recoveries = 0;
+  std::vector<std::vector<uint8_t>> transmitted;
+  bool isConnected() const { return connected && !_tx_disconnect_recovery.pending(); }
+  void serviceBondedOnlyTransition() {}
+  void serviceTxRecovery(uint32_t) { ++recovery_services; }
+  void recoverStalledTx(const char*) { ++recoveries; _tx_disconnect_recovery.begin(); }
+  size_t writeBleUartFrame(const Frame& frame) {
+    ++write_attempts;
+    if (partial_write) return 1;
+    if (!allow_write) return 0;
+    transmitted.emplace_back(frame.buf, frame.buf + frame.len);
+    return frame.len;
+  }
+  void receive(const uint8_t* bytes, size_t size) {
+    assert(recv_queue_len < FRAME_QUEUE_SIZE && size <= MAX_FRAME_SIZE);
+    recv_queue[recv_queue_len].len = size;
+    memcpy(recv_queue[recv_queue_len++].buf, bytes, size);
+  }
+  void shiftSendQueueLeft();
+  void shiftRecvQueueLeft();
+  bool hasRequiredReplyCapacity() const;
+  bool isWriteBusy() const;
+  size_t writeFrame(const uint8_t*, size_t);
+  size_t checkRecvFrame(uint8_t*);
+};
+
+@METHODS@
+
+static void full_required_queue() {
+  current_millis = 100;
+  SerialBLEInterface link;
+  for (uint8_t i = 0; i < FRAME_QUEUE_SIZE; ++i) {
+    const uint8_t reply[] = {0x0d, i};
+    assert(link.writeFrame(reply, sizeof(reply)) == sizeof(reply));
+  }
+  const uint8_t command[] = {66, 'A', '7', '|', 'v', 'e', 'r'};
+  link.receive(command, sizeof(command));
+  uint8_t received[MAX_FRAME_SIZE] = {};
+  assert(link.checkRecvFrame(received) == 0);
+  assert(link.recv_queue_len == 1 && link.send_queue_len == FRAME_QUEUE_SIZE);
+  assert(link.write_attempts == 1 && link.transmitted.empty());
+  ++current_millis;
+  assert(link.checkRecvFrame(received) == 0);
+  assert(link.write_attempts == 1); // Existing 250 ms retry throttle is retained.
+  assert(link.recv_queue_len == 1 && link.isWriteBusy());
+
+  current_millis = 351;
+  link.allow_write = true;
+  // A completed TX frees admission on this SAME call, before RX is examined.
+  assert(link.checkRecvFrame(received) == sizeof(command));
+  assert(memcmp(received, command, sizeof(command)) == 0);
+  assert(link.recv_queue_len == 0 && link.send_queue_len == FRAME_QUEUE_SIZE - 1);
+  const uint8_t reply[] = {29, 'A', '7', '|', 'O', 'K'};
+  assert(link.writeFrame(reply, sizeof(reply)) == sizeof(reply));
+  for (unsigned i = 0; i < FRAME_QUEUE_SIZE + 2; ++i) {
+    ++current_millis;
+    assert(link.checkRecvFrame(received) == 0);
+  }
+  assert(link.transmitted.size() == FRAME_QUEUE_SIZE + 1);
+  for (uint8_t i = 0; i < FRAME_QUEUE_SIZE; ++i)
+    assert((link.transmitted[i] == std::vector<uint8_t>{0x0d, i}));
+  assert(link.transmitted.back() == std::vector<uint8_t>(reply, reply + sizeof(reply)));
+  assert(link.send_queue_len == 0 && link.recv_queue_len == 0);
+}
+
+static void best_effort_eviction() {
+  current_millis = 100;
+  SerialBLEInterface link;
+  for (uint8_t i = 0; i < FRAME_QUEUE_SIZE - 1; ++i) {
+    const uint8_t log[] = {0x88, i};
+    assert(link.writeFrame(log, sizeof(log)) == sizeof(log));
+  }
+  const uint8_t log[] = {0x88, 99};
+  assert(link.writeFrame(log, sizeof(log)) == 0); // One required slot stays reserved.
+  const uint8_t previous[] = {0x0d, 99};
+  assert(link.writeFrame(previous, sizeof(previous)) == sizeof(previous));
+  const uint8_t command[] = {66, 'B', '8', '|', 'v', 'e', 'r'};
+  link.receive(command, sizeof(command));
+  uint8_t received[MAX_FRAME_SIZE] = {};
+  assert(link.isWriteBusy()); // Preserve the nRF52 >=8/12 producer high-water mark.
+  assert(link.checkRecvFrame(received) == sizeof(command));
+  assert(memcmp(received, command, sizeof(command)) == 0);
+  const uint8_t reply[] = {29, 'B', '8', '|', 'O', 'K'};
+  assert(link.writeFrame(reply, sizeof(reply)) == sizeof(reply));
+  current_millis = 351;
+  link.allow_write = true;
+  for (unsigned i = 0; i < FRAME_QUEUE_SIZE + 2; ++i) {
+    assert(link.checkRecvFrame(received) == 0);
+    ++current_millis;
+  }
+  assert(link.transmitted.size() == FRAME_QUEUE_SIZE);
+  assert(link.transmitted[0] == std::vector<uint8_t>(previous, previous + sizeof(previous)));
+  assert(link.transmitted[1] == std::vector<uint8_t>(reply, reply + sizeof(reply)));
+  for (uint8_t i = 0; i < FRAME_QUEUE_SIZE - 2; ++i)
+    assert((link.transmitted[i + 2] == std::vector<uint8_t>{0x88, i}));
+}
+
+static void producer_threshold_and_recovery_ordering() {
+  current_millis = 100;
+  SerialBLEInterface link;
+  const uint8_t reply[] = {0x0d, 1};
+  for (unsigned i = 0; i < 7; ++i) assert(link.writeFrame(reply, sizeof(reply)) == sizeof(reply));
+  assert(!link.isWriteBusy());
+  assert(link.writeFrame(reply, sizeof(reply)) == sizeof(reply));
+  assert(link.isWriteBusy());
+  const uint8_t command[] = {66, 'v'};
+  link.receive(command, sizeof(command));
+  link._tx_disconnect_recovery.begin();
+  uint8_t received[MAX_FRAME_SIZE] = {};
+  assert(link.checkRecvFrame(received) == 0);
+  assert(link.recovery_services == 1 && link.write_attempts == 0 && link.recv_queue_len == 1);
+
+  SerialBLEInterface partial;
+  partial.partial_write = true;
+  assert(partial.writeFrame(reply, sizeof(reply)) == sizeof(reply));
+  partial.receive(command, sizeof(command));
+  assert(partial.checkRecvFrame(received) == 0);
+  assert(partial.recoveries == 1 && partial.write_attempts == 1);
+  assert(partial.recv_queue_len == 1); // Recovery is invoked instead of dispatch.
+}
+
+int main() {
+  full_required_queue();
+  best_effort_eviction();
+  producer_threshold_and_recovery_ordering();
+}
+'''
+
+
+class Nrf52BleBackpressureTests(unittest.TestCase):
+    def test_real_queue_dispatch_throttle_and_recovery_under_sanitizers(self):
+        source = SOURCE.read_text()
+        signatures = ["void SerialBLEInterface::shiftSendQueueLeft()",
+                      "void SerialBLEInterface::shiftRecvQueueLeft()",
+                      "size_t SerialBLEInterface::writeFrame(",
+                      "size_t SerialBLEInterface::checkRecvFrame(",
+                      "bool SerialBLEInterface::isWriteBusy() const"]
+        capacity = "bool SerialBLEInterface::hasRequiredReplyCapacity() const"
+        if capacity in source:
+            signatures.append(capacity)
+        extracted = "\n".join(method(source, signature) for signature in signatures)
+        sanitizers = [] if os.name == "nt" else ["-fsanitize=address,undefined", "-fno-sanitize-recover=all", "-fno-pie", "-no-pie"]
+        with tempfile.TemporaryDirectory(prefix="meshcore-nrf-ble-backpressure-") as temp:
+            binary = Path(temp) / "backpressure"
+            built = subprocess.run(["c++", "-std=c++17", "-x", "c++", "-", "-I" + str(ROOT / "src"),
+                                    *sanitizers, "-o", str(binary)], input=HARNESS.replace("@METHODS@", extracted),
+                                   capture_output=True, text=True, timeout=30)
+            self.assertEqual(built.returncode, 0, built.stderr)
+            result = subprocess.run([str(binary)], capture_output=True, text=True, timeout=10)
+            self.assertEqual(result.returncode, 0, result.stderr)
+
+    def test_queue_capacity_remains_twelve(self):
+        header = (ROOT / "src/helpers/nrf52/SerialBLEInterface.h").read_text()
+        self.assertRegex(header, r"#define\s+FRAME_QUEUE_SIZE\s+12\b")
+
+
+if __name__ == "__main__":
+    unittest.main()

tools/hil/ble_visibility_probe.py

@@ -0,0 +1,163 @@
+#!/usr/bin/env python3
+"""Owned, bounded BlueZ discovery; print only one explicitly selected peer.
+
+Does not pair/connect, power adapters, open USB, or stop shared discovery.
+All nearby-device output stays in process memory and is discarded.
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import os
+from pathlib import Path
+import re
+import subprocess
+
+
+ANSI = re.compile(r"\x1b\[[0-?]*[ -/]*[@-~]")
+MAC = re.compile(r"^(?:[0-9A-F]{2}:){5}[0-9A-F]{2}$")
+
+
+def device_cache_state(output: str, target: str, returncode: int) -> bool | None:
+    """True=exact successful info, False=explicit unavailable, None=query error."""
+    header = re.compile(r"^Device\s+" + re.escape(target) + r"(?:\s+\((?:public|random)\))?\s*$", re.I)
+    unavailable = re.compile(r"^Device\s+" + re.escape(target) + r"\s+not available\s*$", re.I)
+    lines = ANSI.sub("", output).splitlines()
+    headers = sum(bool(header.fullmatch(line)) for line in lines)
+    absent = sum(bool(unavailable.fullmatch(line)) for line in lines)
+    if returncode == 0 and headers == 1 and not absent:
+        return True
+    if returncode in (0, 1) and absent == 1 and not headers:
+        return False
+    return None
+
+
+def is_cached_device(output: str, target: str, returncode: int) -> bool:
+    """Only a successful info header proves a cached record, never its error."""
+    return device_cache_state(output, target, returncode) is True
+
+
+def parse_discovery(output: str, target: str) -> dict:
+    started = False
+    events = 0
+    new_events = 0
+    rssi = []
+    for line in ANSI.sub("", output).splitlines():
+        if "Discovery started" in line:
+            started = True
+        if not started or not re.search(r"\bDevice " + re.escape(target) + r"(?:\s|$)", line, re.I):
+            continue
+        if "[NEW]" in line or "[CHG]" in line:
+            events += 1
+            new_events += int("[NEW]" in line)
+            value = re.search(r"RSSI:\s*(?:0x[0-9a-f]+\s*\()?(-\d+)\)?", line, re.I)
+            if value:
+                rssi.append(int(value[1]))
+    return {
+        "discovery_started": started,
+        "target_events": events,
+        "new_target_events": new_events,
+        "rssi_updates": len(rssi),
+        "rssi_dbm": rssi,
+        # Startup cached-device enumeration is ignored until StartDiscovery succeeds.
+        "advertisement_evidence": bool(rssi or new_events),
+    }
+
+
+def with_cache_evidence(scan: dict, before: bool | None, after: bool | None) -> dict:
+    """Do not invent event counts when BlueZ prints its start marker late.
+
+    A verified absent -> present transition during successful discovery is
+    separate evidence of a newly learned target. It does not prove the exact
+    advertising time/RSSI, nor attribution if another discovery client exists.
+    Unknown/failed info queries never constitute verified prior absence.
+    """
+    result = dict(scan)
+    created = bool(scan["discovery_started"] and before is False and after is True)
+    event_evidence = bool(scan["advertisement_evidence"])
+    sources = []
+    if scan["rssi_updates"]:
+        sources.append("post_start_rssi_update")
+    if scan["new_target_events"]:
+        sources.append("post_start_new_target_event")
+    if created:
+        sources.append("absent_before_present_after_successful_discovery")
+    result.update(
+        cached_before=before is True,
+        cached_after=after is True,
+        cache_query_before_verified=before is not None,
+        cache_query_after_verified=after is not None,
+        post_start_event_evidence=event_evidence,
+        new_target_record_during_scan=created,
+        advertisement_evidence=event_evidence or created,
+        evidence_sources=sources,
+    )
+    return result
+
+
+def usb_identity(alias: str, expected_serial: str) -> dict:
+    path = Path(alias)
+    if path.parent != Path("/dev/serial/by-id") or expected_serial not in path.name:
+        raise ValueError("Explicit matching USB by-id identity required")
+    resolved = path.resolve(strict=True)
+    if not re.fullmatch(r"ttyACM\d+", resolved.name):
+        raise ValueError("Expected native USB ACM radio")
+    device = Path("/sys/class/tty") / resolved.name / "device"
+    for parent in (device.resolve(), *device.resolve().parents):
+        if (parent / "serial").is_file():
+            serial = (parent / "serial").read_text().strip()
+            vid = (parent / "idVendor").read_text().strip().lower()
+            pid = (parent / "idProduct").read_text().strip().lower()
+            if (serial, vid, pid) != (expected_serial, "239a", "8029"):
+                raise ValueError("USB identity mismatch")
+            return {"serial": serial, "vid": vid, "pid": pid, "port": str(resolved)}
+    raise ValueError("USB identity unavailable")
+
+
+def bluetooth(*args: str, timeout: int = 8) -> subprocess.CompletedProcess:
+    return subprocess.run(
+        ["bluetoothctl", *args], stdout=subprocess.PIPE,
+        stderr=subprocess.STDOUT, text=True, timeout=timeout,
+        env={**os.environ, "LC_ALL": "C"}, check=False,
+    )
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--target", required=True)
+    parser.add_argument("--controller", required=True)
+    parser.add_argument("--usb-by-id", required=True)
+    parser.add_argument("--usb-serial", required=True)
+    parser.add_argument("--seconds", type=int, default=20, choices=range(1, 21))
+    args = parser.parse_args()
+    target, controller = args.target.upper(), args.controller.upper()
+    result = {"target": target, "duration_seconds": args.seconds}
+    try:
+        if not MAC.fullmatch(target) or not MAC.fullmatch(controller):
+            raise ValueError("Exact peer and controller MAC required")
+        result["usb_before"] = usb_identity(args.usb_by_id, args.usb_serial)
+        adapter = bluetooth("show", controller)
+        if not re.search(r"Powered:\s*yes\b", adapter.stdout):
+            raise ValueError("Expected controller is not already powered")
+        # Verify this is the selected/default controller, rather than changing it.
+        default = bluetooth("show")
+        if not re.search(r"Controller\s+" + re.escape(controller) + r"\b", default.stdout, re.I):
+            raise ValueError("Expected controller is not the default")
+        before = bluetooth("info", target)
+        before_state = device_cache_state(before.stdout, target, before.returncode)
+        scan = bluetooth("--timeout", str(args.seconds), "scan", "on", timeout=args.seconds + 5)
+        result["scan_exit_code"] = scan.returncode
+        after = bluetooth("info", target)
+        after_state = device_cache_state(after.stdout, target, after.returncode)
+        result.update(with_cache_evidence(parse_discovery(scan.stdout, target), before_state, after_state))
+        result["usb_after"] = usb_identity(args.usb_by_id, args.usb_serial)
+        result["success"] = result["discovery_started"]
+    except (ValueError, OSError, subprocess.SubprocessError):
+        result["success"] = False
+        result["error"] = "Identity-bound discovery could not complete; raw output suppressed"
+    print(json.dumps(result, sort_keys=True))
+    return 0 if result["success"] else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())