Fit V3 Full with NimBLE and bounded OTA source proofs

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

@@ -885,6 +885,7 @@ jobs:
           python3 -B test/test_sh1106_memory.py -v
           python3 -B test/test_recent_repeater_storage.py -v
           python3 -B test/test_ota_cold_buffers.py -v
+          python3 -B test/test_ota_source_proofs.py -v
           python3 -B test/test_neighbor_scope_storage.py
           python3 -B test/test_ed25519_compact.py
           python3 -B test/test_cascade_release_package.py

scripts/check_firmware_ram.py

@@ -300,11 +300,24 @@ def requirements(platform, defines, target):
     if "ENABLE_OTA" in defines:
         # These are lazy on every platform. Freeing static arrays increases
         # linked heap capacity but cannot erase their active-operation cost.
-        # Leaves are sized to the selected source, up to the unchanged limit;
-        # output holds only one block. They can overlap self-serving/fetching.
+        # Legacy sources cache every leaf. Qualified sources instead retain a
+        # bounded grouped-proof workspace (including getter fragments), with
+        # sizeof enforced by the production SourceMerkleProof assertion.
+        # Both paths can overlap self-serving/fetching and one output block.
         leaf_limit = integer(defines, "OTA_PROOFGEN_SCRATCH",
                              16384 if platform == "ESP32_PLATFORM" else
                              8192 if "OTA_SD_STORE" in defines else 4096)
+        bounded_default = (platform == "ESP32_PLATFORM"
+                           and "COMPANION_RADIO_FULL" in defines
+                           and "BOARD_HAS_PSRAM" not in defines
+                           and 4 <= leaf_limit <= 16384)
+        bounded = integer(defines, "OTA_BOUNDED_SOURCE_PROOFS", int(bounded_default))
+        if bounded not in (0, 1):
+            raise ValueError("OTA_BOUNDED_SOURCE_PROOFS must be 0 or 1")
+        if bounded:
+            if not 4 <= leaf_limit <= 16384:
+                raise ValueError("bounded source proofs require a 4..16384-byte leaf limit")
+            leaf_limit = 1536
         block_size = integer(defines, "OTA_MAX_BLOCK", 2048)
         parts["ota_generic_source_buffers"] = leaf_limit + block_size + 32
     # Match OtaDeflateConfig.h's compile eligibility and build-role override.

src/helpers/ota/MotaSourceSerial.cpp

@@ -209,6 +209,16 @@ uint16_t SerialMotaSource::responsePayloadCapacity() const {
   return (uint16_t)(frame - 5u);
 }
 
+#if OTA_BOUNDED_SOURCE_PROOFS
+bool SerialMotaSource::read_metadata(uint8_t idx, uint32_t off, uint8_t* buf, uint32_t len) {
+  if (!buf || !len || len > responsePayloadCapacity() || len > UINT32_MAX - off) return false;
+  uint8_t args[7];
+  args[0] = idx; wr_u32le(args + 1, off);
+  args[5] = (uint8_t)len; args[6] = (uint8_t)(len >> 8);
+  return txn(MS_OP_READ, args, sizeof(args), buf, len);
+}
+#endif
+
 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;

src/helpers/ota/MotaSourceSerial.h

@@ -24,6 +24,10 @@ public:
   uint8_t count() override;
   bool    describe(uint8_t idx, MotaDesc& out) override;
   bool    read(uint8_t idx, uint32_t off, uint8_t* buf, uint32_t len) override;
+#if OTA_BOUNDED_SOURCE_PROOFS
+  uint16_t read_quantum() const override { return responsePayloadCapacity(); }
+  bool read_metadata(uint8_t idx, uint32_t off, uint8_t* buf, uint32_t len) override;
+#endif
   bool    read_deflated_block(uint8_t idx, uint16_t block, uint8_t* buf,
                               uint16_t cap, uint16_t* len) override;
 

src/helpers/ota/OtaManager.cpp

@@ -6,6 +6,9 @@
 #include "OtaDebug.h"
 #include <string.h>
 #include <stdlib.h>   // malloc/free for the (transient) leaf-diff buffer
+#if OTA_BOUNDED_SOURCE_PROOFS
+#include <new>
+#endif
 
 namespace mesh {
 namespace ota {
@@ -28,7 +31,12 @@ void OtaManager::invalidateDirectTransport() {
 OtaManager::~OtaManager() {
   free(_catalog_heap);
   free(_leaves_buf);
+#if OTA_BOUNDED_SOURCE_PROOFS
+  if (_source_proofs) _source_proofs->~BoundedSourceProof();
+  free(_source_proofs);
+#else
   free(_src_leaves);
+#endif
   free(_scratch);
 }
 
@@ -89,6 +97,15 @@ void OtaManager::releaseSmallCatalog() {
   _catalog_heap = nullptr;
 }
 
+#if OTA_BOUNDED_SOURCE_PROOFS
+OtaManager::BoundedSourceProof* OtaManager::ensureSourceProofs() {
+  if (!_source_proofs) {
+    void* memory = malloc(sizeof(BoundedSourceProof));
+    if (memory) _source_proofs = new (memory) BoundedSourceProof;
+  }
+  return _source_proofs;
+}
+#else
 uint8_t* OtaManager::ensureSourceLeaves(uint32_t bytes) {
   if (!bytes || bytes > OTA_PROOFGEN_SCRATCH) return nullptr;
   if (_src_leaves_capacity < bytes) {
@@ -102,6 +119,7 @@ uint8_t* OtaManager::ensureSourceLeaves(uint32_t bytes) {
   }
   return _src_leaves;
 }
+#endif
 
 uint8_t* OtaManager::ensureScratch() {
   if (!_scratch) {
@@ -125,15 +143,27 @@ void OtaManager::releaseColdBuffers() {
   _serve_block_loaded = false;
   _serve_wire_block = nullptr;
   _serve_wire_len = 0;
+#if OTA_BOUNDED_SOURCE_PROOFS
+  if (_source_proofs) _source_proofs->~BoundedSourceProof();
+  free(_source_proofs);
+  _source_proofs = nullptr;
+#else
   free(_src_leaves);
   _src_leaves = nullptr;
   _src_leaves_capacity = 0;
+#endif
   free(_scratch);
   _scratch = nullptr;
 }
 
 void OtaManager::releaseIdleColdBuffers() {
-  if ((_src_leaves || _scratch) &&
+  if ((
+#if OTA_BOUNDED_SOURCE_PROOFS
+       _source_proofs ||
+#else
+       _src_leaves ||
+#endif
+       _scratch) &&
       (uint32_t)(_now_ms - _source_buffer_used_at) >= OTA_SOURCE_BUFFER_IDLE_MS) {
     releaseColdBuffers();
   }
@@ -359,10 +389,9 @@ OtaManager::ServeView* OtaManager::resolve(const uint8_t* mid) {
   return loadSource(_serve[i]) ? &_srcv : nullptr;
 }
 
-// Load an external mota into the on-demand _srcv: read its manifest-minus-leaves + leaves[] from the
-// source into RAM, parse, and wire a payload reader that streams blocks from the source on REQ. (The
-// payload itself is NOT held in RAM - only the small head and the leaves, bounded by the configured
-// proof-generation scratch size.)
+// Load an external source's fixed manifest and wire its payload reader. Legacy
+// builds also cache leaves here; bounded builds acquire proof metadata through
+// cooperative egress steps, without delaying request admission on a full table.
 bool OtaManager::loadSource(const ServeEntry& e) {
   const MotaDesc& d = e.desc;
   if (d.leaves_off < 8) return false;
@@ -373,18 +402,25 @@ bool OtaManager::loadSource(const ServeEntry& e) {
   // Both images share these buffers. Even a failed read can overwrite part of
   // the old manifest/leaves, so its cached MID must stop resolving immediately.
   _srcv.valid = false;
+#if !OTA_BOUNDED_SOURCE_PROOFS
   uint8_t* src_leaves = ensureSourceLeaves(d.block_count * 4u);
   if (!src_leaves) return false;
+#endif
   _source_buffer_used_at = _now_ms;
   bool ok = e.src->read(e.src_idx, 8, _src_manifest, mfl);
   if (!ok || !mota_parse_manifest(_src_manifest, mfl, _srcv.m)) return false;
   if (_srcv.m.block_size() == 0 || _srcv.m.block_size() > OTA_MAX_BLOCK) return false;
   if (memcmp(_srcv.m.merkle_root, d.mid, 4) != 0) return false;       // descriptor/bytes disagree
   if (_srcv.m.block_count != d.block_count) return false;
+  _srcv.m.manifest_start = _src_manifest;
+#if OTA_BOUNDED_SOURCE_PROOFS
+  _srcv.m.leaves = nullptr;
+  _src_leaves_off = d.leaves_off;
+#else
   ok = e.src->read(e.src_idx, d.leaves_off, src_leaves, d.block_count * 4);
   if (!ok) return false;
-  _srcv.m.manifest_start = _src_manifest;
   _srcv.m.leaves   = src_leaves;
+#endif
   _srcv.m.payload  = nullptr;
   _srcv.mfl = mfl;
   _srcv_rdctx.src = e.src; _srcv_rdctx.idx = e.src_idx;
@@ -529,7 +565,11 @@ bool OtaManager::handleGetLeaves(const uint8_t* m, uint16_t n) {
   GetLeavesMsg gl;
   if (!decode_get_leaves(m, n, gl) || gl.want_mask == 0) return false;
   ServeView* v = resolve(gl.manifest_id);
-  if (!v || !v->m.leaves || v->m.block_count == 0 ||
+  if (!v || (!v->m.leaves
+#if OTA_BOUNDED_SOURCE_PROOFS
+             && v != &_srcv
+#endif
+            ) || v->m.block_count == 0 ||
       v->m.block_count > OTA_DIFF_MAX_BLOCKS) return false;
   const uint32_t total = (v->m.block_count * 4 + OTA_LEAVES_FRAG - 1) / OTA_LEAVES_FRAG;
   if (total == 0 || total > OTA_LEAVES_MAXFRAG) return false;
@@ -660,7 +700,11 @@ bool OtaManager::serviceManifestEgress(OtaReplyRouteValid route_valid, void* rou
   const uint32_t mfl = job.leaves ? v->m.block_count * 4u : v->mfl;
   const uint16_t fragment_size = job.leaves ? OTA_LEAVES_FRAG : OTA_MF_FRAG;
   const uint32_t total = (mfl + fragment_size - 1u) / fragment_size;
-  if ((job.leaves && (!v->m.leaves || total > OTA_LEAVES_MAXFRAG)) || total > 16u) {
+  if ((job.leaves && ((!v->m.leaves
+#if OTA_BOUNDED_SOURCE_PROOFS
+                       && v != &_srcv
+#endif
+                      ) || total > OTA_LEAVES_MAXFRAG)) || total > 16u) {
     popManifestJob();
     return true;
   }
@@ -689,7 +733,19 @@ bool OtaManager::serviceManifestEgress(OtaReplyRouteValid route_valid, void* rou
     LeavesMsg message;
     memcpy(message.manifest_id, v->m.merkle_root, 4);
     message.frag_idx = fragment; message.frag_total = ftotal;
-    message.bytes = v->m.leaves + off; message.len = (uint16_t)len;
+    message.bytes = v->m.leaves ? v->m.leaves + off : nullptr;
+#if OTA_BOUNDED_SOURCE_PROOFS
+    if (!message.bytes) {
+      BoundedSourceProof* workspace = ensureSourceProofs();
+      if (!workspace) { popManifestJob(); return true; }
+      _source_buffer_used_at = _now_ms;
+      const auto step = workspace->fragment(_srcv_rdctx.src, _srcv_rdctx.idx,
+          _src_leaves_off + off, (uint16_t)len, message.bytes);
+      if (step == BoundedSourceProof::Step::Pending) return true;
+      if (step == BoundedSourceProof::Step::Failed) { popManifestJob(); return true; }
+    }
+#endif
+    message.len = (uint16_t)len;
     wire_len = encode_leaves(wire, sizeof(wire), message);
   } else {
     ManifestMsg message;
@@ -699,6 +755,9 @@ bool OtaManager::serviceManifestEgress(OtaReplyRouteValid route_valid, void* rou
     wire_len = encode_manifest(wire, sizeof(wire), message);
   }
   if (emit(wire, wire_len, false, &job.route)) {
+#if OTA_BOUNDED_SOURCE_PROOFS
+    if (job.leaves && _source_proofs) _source_proofs->releaseFragment();
+#endif
     OTA_DBG("OTA: %s tx frag=%u/%u len=%u\n", job.leaves ? "LEAVES" : "MANIFEST",
             (unsigned)fragment, (unsigned)ftotal, (unsigned)len);
     const uint16_t bit = (uint16_t)(1u << fragment);
@@ -908,10 +967,25 @@ void OtaManager::serviceEgress(OtaReplyRouteValid route_valid, void* route_ctx)
   }
 
   if (job.pending_mask != 0) {
+    const bool was_loaded = _serve_block_loaded;
     if (!_serve_block_loaded && !loadActiveServeBlock()) {
       popServeJob();
       return;
     }
+#if OTA_BOUNDED_SOURCE_PROOFS
+    // Warm proof metadata while DATA waits behind radio backpressure. Never
+    // combine this new read with the existing payload load in the same pass.
+    if (was_loaded && v == &_srcv) {
+      BoundedSourceProof* workspace = ensureSourceProofs();
+      if (!workspace || !workspace->bind(_srcv_rdctx.src, _srcv_rdctx.idx,
+          _src_leaves_off, v->m.block_count, v->m.merkle_root) ||
+          workspace->prepare() == BoundedSourceProof::Step::Failed) {
+        popServeJob(); return;
+      }
+    }
+#else
+    (void)was_loaded;
+#endif
     const uint16_t fragment_data = job.route.direct ? OTA_DIRECT_FRAG_DATA :
         (job.wire_v2 ? OTA_FRAG_DATA_V2 : OTA_FRAG_DATA);
     const uint16_t valid_mask = frag_full_mask(
@@ -974,7 +1048,21 @@ void OtaManager::serviceEgress(OtaReplyRouteValid route_valid, void* route_ctx)
       && (int32_t)(_now_ms - job.proof_ready_at) < 0) return;
 
   uint8_t proof[32 * 4];
-  uint8_t np = merkle_gen_proof(v->m.leaves, v->m.block_count, job.block, v->scratch, proof);
+  uint8_t np = 0;
+#if OTA_BOUNDED_SOURCE_PROOFS
+  if (v == &_srcv) {
+    BoundedSourceProof* workspace = ensureSourceProofs();
+    if (!workspace || !workspace->bind(_srcv_rdctx.src, _srcv_rdctx.idx,
+        _src_leaves_off, v->m.block_count, v->m.merkle_root)) { popServeJob(); return; }
+    _source_buffer_used_at = _now_ms;
+    const auto step = workspace->proof(job.block, proof, np);
+    if (step == BoundedSourceProof::Step::Pending) return;
+    if (step == BoundedSourceProof::Step::Failed) { popServeJob(); return; }
+  } else
+#endif
+  {
+    np = merkle_gen_proof(v->m.leaves, v->m.block_count, job.block, v->scratch, proof);
+  }
   ProofMsg pm;
   memcpy(pm.manifest_id, job.mid, 4);
   pm.block_idx = job.block;

src/helpers/ota/OtaManager.h

@@ -12,6 +12,9 @@
 #include "MotaContainer.h"
 #include "OtaSource.h"
 #include "MerkleTree.h"
+#if OTA_BOUNDED_SOURCE_PROOFS
+#include "SourceMerkleProof.h"
+#endif
 
 // Transport-agnostic OTA session engine (docs/ota_protocol.md Section 5/Section 8). It SERVES a complete `.mota`
 // (answering GET_MANIFEST / REQ) and/or FETCHES one into an OtaStore (verifying every block against
@@ -751,8 +754,17 @@ private:
   // capacity limit. Device-encoder output needs at most one logical block;
   // host-encoded output uses _serve_block. Cold buffers stay pinned until
   // queued egress drains, then return to the heap at idle.
+#if OTA_BOUNDED_SOURCE_PROOFS
+  using BoundedSourceProof = SourceMerkleProof<OTA_PROOFGEN_SCRATCH / 4, OTA_LEAVES_FRAG>;
+  static_assert(OTA_PROOFGEN_SCRATCH >= 4 && OTA_PROOFGEN_SCRATCH <= 16384 &&
+                sizeof(BoundedSourceProof) <= 1536,
+                "Update the bounded source workspace RAM budget in check_firmware_ram.py");
+  BoundedSourceProof* _source_proofs = nullptr;
+  uint32_t _src_leaves_off = 0;
+#else
   uint8_t*    _src_leaves = nullptr;
   uint32_t    _src_leaves_capacity = 0;
+#endif
   uint8_t*    _scratch = nullptr;
   uint32_t    _source_buffer_used_at = 0;
 
@@ -900,7 +912,11 @@ private:
   bool       _diffing = false;                 // leaves are in; diffing the seed a batch per tick
   uint32_t   _diff_idx = 0;                    // next block index to diff against the seed
 
+#if OTA_BOUNDED_SOURCE_PROOFS
+  BoundedSourceProof* ensureSourceProofs();
+#else
   uint8_t* ensureSourceLeaves(uint32_t bytes);
+#endif
   uint8_t* ensureScratch();
   void releaseColdBuffers();
   void releaseIdleColdBuffers();

src/helpers/ota/OtaMemoryPolicy.h

@@ -1,5 +1,25 @@
 #pragma once
 
+// Non-PSRAM Full Companions retain the complete source geometry without
+// retaining every Merkle leaf in internal RAM. An explicit 0 keeps the legacy
+// cache for matched qualification builds.
+#ifndef OTA_BOUNDED_SOURCE_PROOFS
+#if defined(ESP32_PLATFORM) && defined(COMPANION_RADIO_FULL) && !defined(BOARD_HAS_PSRAM) && \
+    (!defined(OTA_PROOFGEN_SCRATCH) || \
+     (OTA_PROOFGEN_SCRATCH >= 4 && OTA_PROOFGEN_SCRATCH <= 16384))
+#define OTA_BOUNDED_SOURCE_PROOFS 1
+#else
+#define OTA_BOUNDED_SOURCE_PROOFS 0
+#endif
+#endif
+#if OTA_BOUNDED_SOURCE_PROOFS != 0 && OTA_BOUNDED_SOURCE_PROOFS != 1
+#error "OTA_BOUNDED_SOURCE_PROOFS must be 0 or 1"
+#endif
+#if OTA_BOUNDED_SOURCE_PROOFS && defined(OTA_PROOFGEN_SCRATCH) && \
+    (OTA_PROOFGEN_SCRATCH < 4 || OTA_PROOFGEN_SCRATCH > 16384)
+#error "Bounded source proofs require 4..16384 leaf bytes; use legacy mode for another capacity"
+#endif
+
 // Qualified Full Companions stream host-provided firmware and need no permanent
 // mOTA workspace. Apply the same policy to build.sh and direct PlatformIO builds.
 #if defined(OTA_SEEDER_ONLY) && defined(COMPANION_RADIO_FULL) && \

src/helpers/ota/OtaSource.h

@@ -3,6 +3,7 @@
 #include <stdint.h>
 #include <stddef.h>
 #include "OtaFormat.h"
+#include "OtaMemoryPolicy.h"
 
 // Transport-agnostic "folder of firmware" abstraction (docs/ota_protocol.md Section 9). A node can RELAY
 // `.mota` images it does not hold in flash - a user drops several `.mota` (different architectures) into
@@ -56,6 +57,17 @@ public:
   // exactly `len` bytes were produced. May block on the transport (serial round-trip); implementations
   // should keep active-transfer reads prompt because requested OTA traffic is primary traffic.
   virtual bool    read(uint8_t idx, uint32_t off, uint8_t* buf, uint32_t len) = 0;
+#if OTA_BOUNDED_SOURCE_PROOFS
+  // Maximum bytes for one cooperative metadata step. Framed transports expose
+  // their negotiated single-response capacity so proof work never batches
+  // several timeout-bearing transactions into one mesh-loop pass.
+  virtual uint16_t read_quantum() const { return 128; }
+  // A framed source must reject a chunk that no longer fits after a negotiated
+  // capacity change, rather than silently splitting it into several requests.
+  virtual bool read_metadata(uint8_t idx, uint32_t off, uint8_t* buf, uint32_t len) {
+    return read(idx, off, buf, len);
+  }
+#endif
   // Optional transport-only raw-DEFLATE representation of one logical payload block. Implementations which
   // can generate/cache it off-device return a stream strictly smaller than the raw block. The default keeps
   // every existing source source-compatible and makes OtaManager fall back to raw 171-byte DATA.

src/helpers/ota/SourceMerkleProof.h

@@ -0,0 +1,154 @@
+#pragma once
+
+#include "MerkleTree.h"
+#include "OtaSource.h"
+#include <string.h>
+
+namespace mesh { namespace ota {
+
+// A power-of-two group preserves the original tree after five reductions,
+// including promotion of the final partial group. Retain group roots instead
+// of the complete leaf table; a block's wire proof is its local proof followed
+// by the group-root proof. No source/transport pointer owns storage here.
+template <uint32_t MaximumLeaves, uint16_t FragmentCapacity>
+class SourceMerkleProof {
+public:
+  enum class Step : uint8_t { Pending, Ready, Failed };
+  static constexpr uint32_t GroupLeaves = 32;
+  static constexpr uint32_t GroupBytes = GroupLeaves * 4;
+  static constexpr uint32_t MaximumGroups = (MaximumLeaves + GroupLeaves - 1) / GroupLeaves;
+  static_assert(MaximumLeaves > 0 && MaximumLeaves <= UINT32_MAX / 4,
+                "invalid source proof capacity");
+
+  void reset() {
+    _source = nullptr;
+    _roots_ready = false;
+    _fragment_source = nullptr;
+    _fragment_used = 0;
+  }
+
+  bool bind(MotaSource* source, uint8_t index, uint32_t offset,
+            uint32_t count, const uint8_t root[4]) {
+    if (!source || !root || !count || count > MaximumLeaves ||
+        uint64_t(offset) + uint64_t(count) * 4 > UINT32_MAX) return false;
+    if (_source == source && _index == index && _offset == offset &&
+        _count == count && !memcmp(_root, root, 4)) return true;
+    _source = source; _index = index; _offset = offset; _count = count;
+    memcpy(_root, root, 4);
+    _groups = (count + GroupLeaves - 1) / GroupLeaves;
+    _warm_group = 0; _group_used = 0; _selected_group = UINT32_MAX;
+    _roots_ready = false;
+    return true;
+  }
+
+  // At most one source read, at most 32 leaves reduced after that read. The
+  // final root check is bounded by MaximumGroups (128 on ESP32).
+  Step prepare() {
+    if (!_source) return Step::Failed;
+    if (_roots_ready) return Step::Ready;
+    const uint32_t bytes = groupBytes(_warm_group);
+    if (!readPart(_source, _index, _offset + _warm_group * GroupBytes,
+                  _group, bytes, _group_used)) return failProof();
+    if (_group_used != bytes) return Step::Pending;
+    if (!groupRoot(bytes, _roots + _warm_group * 4)) return failProof();
+    ++_warm_group;
+    _group_used = 0;
+    if (_warm_group == _groups) {
+      _accumulator.reset();
+      for (uint32_t i = 0; i < _groups; ++i)
+        if (!_accumulator.add(_roots + i * 4)) return failProof();
+      uint8_t root[4];
+      if (!_accumulator.finish(root) || memcmp(root, _root, 4)) return failProof();
+      _roots_ready = true;
+    }
+    // Even the final warm read yields before reading the requested group.
+    return Step::Pending;
+  }
+
+  Step proof(uint32_t block, uint8_t output[128], uint8_t& length) {
+    length = 0;
+    if (!output || !_source || block >= _count) return Step::Failed;
+    const Step prepared = prepare();
+    if (prepared != Step::Ready) return prepared;
+    const uint32_t group = block / GroupLeaves;
+    if (_selected_group != group) {
+      _selected_group = group;
+      _group_used = 0;
+    }
+    const uint32_t bytes = groupBytes(group);
+    if (_group_used != bytes) {
+      if (!readPart(_source, _index, _offset + group * GroupBytes,
+                    _group, bytes, _group_used)) return failProof();
+      if (_group_used != bytes) return Step::Pending;
+      uint8_t root[4];
+      if (!groupRoot(bytes, root) || memcmp(root, _roots + group * 4, 4))
+        return failProof();
+    }
+    length = merkle_gen_proof(_group, bytes / 4, block % GroupLeaves, nullptr, output);
+    length += merkle_gen_proof(_roots, _groups, group, nullptr, output + length * 4);
+    return Step::Ready;
+  }
+
+  // Getter fragments have an independent binding and buffer. A request for
+  // another MID cannot reset a partially warmed proof cache, and a rejected
+  // radio emit retains the exact bytes without another source transaction.
+  Step fragment(MotaSource* source, uint8_t index, uint32_t offset,
+                uint16_t bytes, const uint8_t*& output) {
+    output = nullptr;
+    if (!source || !bytes || bytes > FragmentCapacity ||
+        uint64_t(offset) + bytes > UINT32_MAX) return Step::Failed;
+    if (_fragment_source != source || _fragment_index != index ||
+        _fragment_offset != offset || _fragment_bytes != bytes) {
+      _fragment_source = source; _fragment_index = index;
+      _fragment_offset = offset; _fragment_bytes = bytes; _fragment_used = 0;
+    }
+    if (_fragment_used != bytes &&
+        !readPart(source, index, offset, _fragment, bytes, _fragment_used)) {
+      _fragment_source = nullptr;
+      return Step::Failed;
+    }
+    if (_fragment_used != bytes) return Step::Pending;
+    output = _fragment;
+    return Step::Ready;
+  }
+  void releaseFragment() { _fragment_source = nullptr; _fragment_used = 0; }
+
+private:
+  static bool readPart(MotaSource* source, uint8_t index, uint32_t offset,
+                       uint8_t* output, uint32_t bytes, uint32_t& used) {
+    uint32_t chunk = source->read_quantum();
+    if (!chunk) return false;
+    if (chunk > GroupBytes) chunk = GroupBytes;
+    if (chunk > bytes - used) chunk = bytes - used;
+    if (!source->read_metadata(index, offset + used, output + used, chunk)) return false;
+    used += chunk;
+    return true;
+  }
+  uint32_t groupBytes(uint32_t group) const {
+    const uint32_t remaining = _count - group * GroupLeaves;
+    return (remaining < GroupLeaves ? remaining : GroupLeaves) * 4;
+  }
+  bool groupRoot(uint32_t bytes, uint8_t root[4]) {
+    _accumulator.reset();
+    for (uint32_t i = 0; i < bytes; i += 4)
+      if (!_accumulator.add(_group + i)) return false;
+    return _accumulator.finish(root);
+  }
+  Step failProof() { _source = nullptr; _roots_ready = false; return Step::Failed; }
+
+  uint8_t _roots[MaximumGroups * 4];
+  uint8_t _group[GroupBytes];
+  uint8_t _fragment[FragmentCapacity];
+  MerkleAccumulator _accumulator;
+  MotaSource* _source = nullptr;
+  MotaSource* _fragment_source = nullptr;
+  uint32_t _offset = 0, _count = 0, _groups = 0;
+  uint32_t _warm_group = 0, _group_used = 0, _selected_group = UINT32_MAX;
+  uint32_t _fragment_offset = 0, _fragment_used = 0;
+  uint16_t _fragment_bytes = 0;
+  uint8_t _root[4] = {};
+  uint8_t _index = 0, _fragment_index = 0;
+  bool _roots_ready = false;
+};
+
+}} // namespace mesh::ota

test/fixtures/ota_source_proofs/test.cpp

@@ -0,0 +1,288 @@
+
+using Workspace = SourceMerkleProof<4096, OTA_LEAVES_FRAG>;
+static_assert(sizeof(Workspace) <= 1536, "bounded allocation includes getter scratch");
+
+class LeafSource : public MotaSource {
+public:
+  std::vector<uint8_t> bytes;
+  uint16_t quantum = 128;
+  unsigned calls = 0, fail_at = 0;
+  uint8_t root[4];
+  explicit LeafSource(uint32_t count) : bytes(17 + count * 4) {
+    for (uint32_t i = 0; i < count; ++i) {
+      uint8_t value[4]; wr_u32le(value, i * 8191u + count);
+      merkle_leaf(bytes.data() + 17 + i * 4, value, 4);
+    }
+    merkle_root(root, bytes.data() + 17, count);
+  }
+  uint8_t count() override { return 1; }
+  bool describe(uint8_t, MotaDesc&) override { return false; }
+  uint16_t read_quantum() const override { return quantum; }
+  bool read(uint8_t index, uint32_t offset, uint8_t* output, uint32_t length) override {
+    ++calls;
+    assert(length <= quantum && length <= 128);
+    if (fail_at == calls) { if (length) output[0] = 0xEF; return false; }
+    if (index || uint64_t(offset) + length > bytes.size()) return false;
+    memcpy(output, bytes.data() + offset, length); return true;
+  }
+};
+
+static std::vector<std::vector<uint8_t>> reference_levels(const uint8_t* leaves, uint32_t count) {
+  std::vector<std::vector<uint8_t>> levels{{leaves, leaves + count * 4}};
+  while (count > 1) {
+    std::vector<uint8_t> next(((count + 1) / 2) * 4);
+    const auto& previous = levels.back();
+    for (uint32_t i = 0; i < count; i += 2) {
+      if (i + 1 < count) merkle_combine(next.data() + (i / 2) * 4,
+          previous.data() + i * 4, previous.data() + (i + 1) * 4);
+      else memcpy(next.data() + (i / 2) * 4, previous.data() + i * 4, 4);
+    }
+    levels.push_back(std::move(next)); count = (count + 1) / 2;
+  }
+  return levels;
+}
+
+static void grouped_equivalence() {
+  std::vector<uint32_t> counts;
+  for (uint32_t count = 1; count <= 65; ++count) counts.push_back(count);
+  for (uint32_t count : {127,128,129,255,256,257,511,512,513,1023,1024,1025,
+                         2047,2048,2049,4095,4096}) counts.push_back(count);
+  size_t checked = 0;
+  for (uint32_t count : counts) {
+    LeafSource source(count);
+    source.quantum = count <= 65 ? uint16_t(std::array<int, 5>{1,3,17,127,128}[count % 5]) : 127;
+    Workspace workspace;
+    assert(workspace.bind(&source, 0, 17, count, source.root));
+    const auto expected = reference_levels(source.bytes.data() + 17, count);
+    for (uint32_t block = 0; block < count; ++block) {
+      uint8_t output[128]; uint8_t length = 99;
+      Workspace::Step step = Workspace::Step::Pending;
+      for (unsigned turn = 0; step == Workspace::Step::Pending && turn < 20000; ++turn) {
+        const auto before = source.calls;
+        step = workspace.proof(block, output, length);
+        assert(source.calls - before <= 1);
+      }
+      assert(step == Workspace::Step::Ready);
+      std::vector<uint8_t> proof;
+      uint32_t index = block;
+      for (const auto& row : expected) {
+        const uint32_t sibling = index ^ 1u;
+        if (sibling < row.size() / 4)
+          proof.insert(proof.end(), row.begin() + sibling * 4, row.begin() + sibling * 4 + 4);
+        index /= 2;
+      }
+      assert(proof.size() == size_t(length) * 4);
+      assert(proof.empty() || !memcmp(output, proof.data(), proof.size()));
+      assert(merkle_verify_from_leaf(source.bytes.data() + 17 + block * 4, block,
+          output, length, source.root, count));
+      const auto before = source.calls;
+      assert(workspace.proof(block, output, length) == Workspace::Step::Ready);
+      assert(source.calls == before); // exact retry never re-reads a pinned group
+      ++checked;
+    }
+  }
+  printf("PASS: byte-identical grouped proofs (%zu blocks, max4096; workspace=%zu)\n", checked, sizeof(Workspace));
+}
+
+static void helper_faults() {
+  LeafSource source(65), other(65); source.quantum = 3; other.quantum = 17;
+  Workspace workspace;
+  assert(!workspace.bind(nullptr, 0, 17, 65, source.root));
+  assert(!workspace.bind(&source, 0, 17, 0, source.root));
+  assert(!workspace.bind(&source, 0, 17, 4097, source.root));
+  assert(!workspace.bind(&source, 0, UINT32_MAX, 65, source.root));
+  assert(workspace.bind(&source, 0, 17, 65, source.root));
+  assert(workspace.prepare() == Workspace::Step::Pending);
+  const auto warmed = source.calls;
+  const uint8_t* fragment = nullptr;
+  while (workspace.fragment(&other, 0, 17, 101, fragment) == Workspace::Step::Pending) {}
+  assert(fragment && !memcmp(fragment, other.bytes.data() + 17, 101));
+  const auto read = other.calls;
+  assert(workspace.fragment(&other, 0, 17, 101, fragment) == Workspace::Step::Ready);
+  assert(other.calls == read && source.calls == warmed);
+  workspace.releaseFragment();
+  source.fail_at = source.calls + 1;
+  assert(workspace.prepare() == Workspace::Step::Failed); // partial writes never authenticated
+  source.fail_at = 0;
+  assert(workspace.bind(&source, 0, 17, 65, source.root));
+  while (workspace.prepare() == Workspace::Step::Pending) {}
+  source.bytes[17] ^= 1; // source changed after the authenticated root cache was built
+  uint8_t proof[128], length = 0;
+  Workspace::Step step;
+  do { step = workspace.proof(0, proof, length); } while (step == Workspace::Step::Pending);
+  assert(step == Workspace::Step::Failed && length == 0);
+  source.bytes[17] ^= 1;
+  uint8_t wrong_root[4]; memcpy(wrong_root, source.root, 4); wrong_root[0] ^= 1;
+  assert(workspace.bind(&source, 0, 17, 65, wrong_root));
+  do { step = workspace.prepare(); } while (step == Workspace::Step::Pending);
+  assert(step == Workspace::Step::Failed);
+  source.quantum = 0;
+  assert(workspace.bind(&source, 0, 17, 65, source.root));
+  assert(workspace.prepare() == Workspace::Step::Failed);
+  workspace.reset();
+  assert(workspace.proof(0, proof, length) == Workspace::Step::Failed);
+  puts("PASS: partial reads and source changes");
+}
+
+class TrackedSource : public Source {
+public:
+  using Source::Source;
+  uint16_t quantum = 17;
+  uint32_t leaf_bytes = 0, leaf_calls = 0;
+  bool fail_leaves = false;
+  uint16_t read_quantum() const override { return quantum; }
+  bool read(uint8_t index, uint32_t offset, uint8_t* output, uint32_t length) override {
+    const uint32_t first = 8 + MOTA_MFL, last = first + manifest.block_count * 4;
+    if (offset >= first && offset < last) {
+      assert(length <= quantum && length <= 128 && offset + length <= last);
+      ++leaf_calls; leaf_bytes += length;
+      if (fail_leaves) { if (length) output[0] = 0xEE; return false; }
+    }
+    return Source::read(index, offset, output, length);
+  }
+};
+
+static void manager_lifetimes() {
+  TrackedSource a(4096, 0x73), b(65, 0x56);
+  a.host_encoding = b.host_encoding = Source::HostEncoding::Good;
+  a.fail_payload = true;
+  Sent sent;
+  {
+    OtaManager manager; manager.begin(123, send, &sent);
+    assert(manager.add_source(&a) && manager.add_source(&b));
+    assert(request(manager, a, 4095));
+    assert(allocated == 0); // receiving a request never allocates a full leaf table
+    sent.accept_data = sent.accept_proof = false;
+    manager.serviceEgress(); // cache host-generated DATA
+    assert(a.host_calls == 1);
+    for (unsigned i = 0; i < 20; ++i) {
+      const auto before = a.leaf_calls;
+      manager.serviceEgress(); assert(a.leaf_calls - before <= 1);
+    }
+    assert(allocated == sizeof(Workspace) && peak <= 1536);
+    // A different MID's manifest and getter must not restart A's warming.
+    assert(manifest_request(manager, b));
+    for (unsigned i = 0; i < 20 && manager.pendingManifestJobs(); ++i) {
+      manager.set_clock(100000 + i * 10000); manager.serviceEgress();
+    }
+    assert(!manager.pendingManifestJobs());
+    GetLeavesMsg get{}; memcpy(get.manifest_id, b.manifest.merkle_root, 4); get.want_mask = 1;
+    uint8_t wire[MAX_PACKET_PAYLOAD];
+    assert(manager.on_message(wire, encode_get_leaves(wire, sizeof(wire), get)));
+    sent.accept_manifest = false;
+    for (unsigned i = 0; i < 30 && manager.pendingManifestJobs(); ++i) {
+      const auto before = b.leaf_calls;
+      manager.set_clock(400000 + i * 10000); manager.serviceEgress();
+      assert(b.leaf_calls - before <= 1);
+    }
+    assert(manager.pendingManifestJobs());
+    const auto getter_reads = b.leaf_calls;
+    for (unsigned i = 0; i < 10; ++i) manager.serviceEgress();
+    assert(b.leaf_calls == getter_reads); // a rejected getter emit pins its exact fragment
+    sent.accept_manifest = true;
+    manager.serviceEgress();
+    assert(!manager.pendingManifestJobs());
+    bool getter_seen = false;
+    for (const auto& packet : sent.packets) {
+      LeavesMsg leaves{};
+      if (decode_leaves(packet.data(), packet.size(), leaves)) {
+        assert(leaves.len == OTA_LEAVES_FRAG &&
+               !memcmp(leaves.bytes, b.manifest.leaves, leaves.len));
+        getter_seen = true;
+      }
+    }
+    assert(getter_seen);
+    sent.accept_data = true;
+    for (unsigned i = 0; i < 2000; ++i) {
+      const auto before = a.leaf_calls;
+      manager.set_clock(800000 + i * 1000); manager.serviceEgress();
+      assert(a.leaf_calls - before <= 1);
+    }
+    assert(manager.pendingServeJobs() && a.host_calls == 1);
+    const auto before = a.leaf_calls;
+    for (unsigned i = 0; i < 10; ++i) manager.serviceEgress();
+    assert(a.leaf_calls == before); // ready proof held under emit backpressure
+    sent.accept_proof = true;
+    manager.serviceEgress();
+    assert(!manager.pendingServeJobs());
+    verify_v2(sent, a, 4095, true);
+    manager.set_clock(5000000); manager.loop();
+    assert(allocated == 0); // idle timer releases the only workspace without an encoder scratch
+    sent.accept_data = false;
+    assert(request(manager, a, 4095));
+    fail_next = true;
+    manager.serviceEgress(); manager.serviceEgress();
+    assert(!manager.pendingServeJobs() && !fail_next && allocated == 0);
+    a.fail_leaves = true;
+    assert(request(manager, a, 4095));
+    manager.serviceEgress(); manager.serviceEgress();
+    assert(!manager.pendingServeJobs());
+    manager.clear_sources(); assert(allocated == 0);
+    a.fail_leaves = false;
+    assert(manager.add_source(&a) && request(manager, a, 4095));
+    manager.serviceEgress(); manager.serviceEgress();
+    assert(allocated == sizeof(Workspace));
+    assert(manager.remove_source(&a)); assert(allocated == 0 && !manager.pendingServeJobs());
+  }
+  assert(allocated == 0);
+  puts("PASS: bounded manager ownership and backpressure");
+}
+
+static void simultaneous_capture() {
+  for (bool host : {false, true}) {
+    TrackedSource outgoing(5, 0x40);
+    Source incoming(3, 0x70);
+    if (host) {
+      outgoing.host_encoding = Source::HostEncoding::Good;
+      outgoing.host_encoded_size = 400;
+    }
+    Sent sent; OtaManager manager; OtaStoreRam<16384> store;
+    manager.begin(123, send, &sent); manager.set_fetch_store(&store);
+    manager.set_transport_deflate_encoder(encode); manager.set_transport_deflate_decoder(decode);
+    assert(manager.add_source(&outgoing) && request(manager, outgoing));
+    sent.accept_data = false;
+    manager.serviceEgress(); manager.serviceEgress();
+    const size_t expected_allocation = sizeof(Workspace) + (host ? 0 : OTA_MAX_BLOCK);
+    assert(allocated == expected_allocation);
+    assert(manager.pull_archive(incoming.manifest.merkle_root, 123, false) == OtaManager::PULL_STARTED);
+    for (unsigned fragment = 0; fragment < 2; ++fragment) {
+      ManifestMsg mf{}; memcpy(mf.manifest_id, incoming.manifest.merkle_root, 4);
+      mf.frag_idx = fragment; mf.frag_total = 2;
+      mf.bytes = incoming.manifest.manifest_start + fragment * OTA_MF_FRAG;
+      mf.len = fragment == 0 ? OTA_MF_FRAG : MOTA_MFL - OTA_MF_FRAG;
+      uint8_t wire[MAX_PACKET_PAYLOAD];
+      manager.on_message(wire, encode_manifest(wire, sizeof(wire), mf));
+    }
+    assert(manager.fetchState() == OtaManager::FETCHING);
+    for (uint16_t block = 0; block < 3; ++block) {
+      uint8_t representation[12]; memset(representation + 4, uint8_t(0x70 + block), 8);
+      mh4(representation, representation + 4, 8);
+      DataMsg data{}; memcpy(data.manifest_id, incoming.manifest.merkle_root, 4); data.block_idx = block;
+      assert(ota_data_v2_pack_extended(0, 8, data.frag_off));
+      data.data = representation; data.data_len = sizeof(representation);
+      uint8_t wire[MAX_PACKET_PAYLOAD];
+      assert(manager.on_message(wire, encode_data(wire, sizeof(wire), data)));
+      uint8_t siblings[128]; ProofMsg proof{};
+      memcpy(proof.manifest_id, incoming.manifest.merkle_root, 4); proof.block_idx = block;
+      proof.n_proof = merkle_gen_proof(incoming.manifest.leaves, 3, block, nullptr, siblings);
+      proof.proof = siblings;
+      assert(manager.on_message(wire, encode_proof(wire, sizeof(wire), proof)));
+      manager.set_clock(100000 + block * 10000); manager.loop();
+      assert(allocated == expected_allocation && manager.pendingServeJobs());
+    }
+    assert(manager.fetchState() == OtaManager::COMPLETE);
+    MotaManifest staged; assert(mota_parse(store.data(), store.staged_size(), staged));
+    assert(!memcmp(staged.payload, incoming.manifest.payload, incoming.manifest.payload_size));
+    sent.accept_data = true; drain(manager, 200000);
+    if (host) verify_v2(sent, outgoing, 0, true, 400);
+    else verify(sent, outgoing, true);
+    manager.clear_sources(); assert(allocated == 0);
+  }
+  puts("PASS: simultaneous capture and serving");
+}
+
+int main() {
+  grouped_equivalence(); helper_faults(); manager_lifetimes(); simultaneous_capture();
+  assert(allocated == 0);
+  return 0;
+}

test/test_firmware_ram.py

@@ -628,6 +628,44 @@ int main() {
                 }, "RAK_4631_sensor")
                 self.assertEqual(policy["components"]["ethernet_or_external_ota_workspaces"], expected)
 
+    def test_bounded_source_proofs_reserve_real_workspace_without_cutting_other_allowances(self):
+        for platform in ("ESP32_PLATFORM", "NRF52_PLATFORM", "RP2040_PLATFORM", "STM32_PLATFORM"):
+            for full in (False, True):
+                for psram in (False, True):
+                    flags = {"ENABLE_OTA": 1, "WIFI_SSID": "test", "BLE_PIN_CODE": 123456,
+                             "WITH_MQTT_BRIDGE": 1}
+                    if full:
+                        flags["COMPANION_RADIO_FULL"] = 1
+                    if psram:
+                        flags["BOARD_HAS_PSRAM"] = 1
+                    with self.subTest(platform=platform, full=full, psram=psram):
+                        legacy = ram.requirements(platform, {**flags, "OTA_BOUNDED_SOURCE_PROOFS": 0},
+                                                  "board_companion_radio_full")
+                        selected = ram.requirements(platform, flags, "board_companion_radio_full")
+                        expected = 1536 if platform == "ESP32_PLATFORM" and full and not psram else (
+                            16384 if platform == "ESP32_PLATFORM" else 4096)
+                        self.assertEqual(selected["components"]["ota_generic_source_buffers"],
+                                         expected + 2048 + 32)
+                        for name, amount in legacy["components"].items():
+                            if name != "ota_generic_source_buffers":
+                                self.assertEqual(selected["components"][name], amount)
+        for custom in (0, 3, 32768):
+            flags = {"ENABLE_OTA": 1, "COMPANION_RADIO_FULL": 1, "OTA_PROOFGEN_SCRATCH": custom}
+            extended = ram.requirements("ESP32_PLATFORM", flags, "board_companion_radio_full")
+            self.assertEqual(extended["components"]["ota_generic_source_buffers"], custom + 2048 + 32)
+            with self.assertRaises(ValueError):
+                ram.requirements("ESP32_PLATFORM", {**flags, "OTA_BOUNDED_SOURCE_PROOFS": 1},
+                                 "board_companion_radio_full")
+        odd_limit = ram.requirements("ESP32_PLATFORM", {
+            "ENABLE_OTA": 1, "COMPANION_RADIO_FULL": 1, "OTA_PROOFGEN_SCRATCH": 16383,
+        }, "board_companion_radio_full")
+        self.assertEqual(odd_limit["components"]["ota_generic_source_buffers"], 1536 + 2048 + 32)
+        for invalid in (-1, 2, "invalid"):
+            with self.subTest(invalid=invalid), self.assertRaises(ValueError):
+                ram.requirements("ESP32_PLATFORM", {
+                    "ENABLE_OTA": 1, "OTA_BOUNDED_SOURCE_PROOFS": invalid,
+                }, "board_companion_radio_full")
+
     def test_compact_tables_keep_capacity_and_reserve_active_ota_allocations(self):
         for platform, leaf_limit in (("NRF52_PLATFORM", 4096), ("ESP32_PLATFORM", 16384),
                                      ("RP2040_PLATFORM", 4096), ("STM32_PLATFORM", 4096)):

test/test_nimble_companion.py

@@ -10,7 +10,11 @@ ROOT = Path(__file__).resolve().parents[1]
 
 class NimbleCompanionTests(unittest.TestCase):
     def test_identity_security_and_data_path_under_sanitizers(self):
-        self.run_adapter()
+        self.run_adapter("-DCONFIG_BT_NIMBLE_MAX_CONNECTIONS=1",
+                         "-DCONFIG_BT_NIMBLE_MAX_BONDS=15",
+                         "-DCONFIG_BT_NIMBLE_MAX_CCCDS=30",
+                         "-DCONFIG_BT_NIMBLE_ROLE_CENTRAL_DISABLED=1",
+                         "-DCONFIG_BT_NIMBLE_ROLE_OBSERVER_DISABLED=1")
 
     def test_lab15_interval_real_adapter_under_sanitizers(self):
         self.run_adapter("-DMESH_HIL_BLE_BENCH=1", "-DMESH_HIL_BLE_INTERVAL_MS=15")

test/test_ota_source_proofs.py

@@ -0,0 +1,129 @@
+#!/usr/bin/env python3
+"""Execute bounded source proofs and real manager ownership under sanitizers."""
+from pathlib import Path
+import os
+import subprocess
+import tempfile
+import unittest
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+class OtaSourceProofTests(unittest.TestCase):
+    def test_actual_policy_preserves_custom_source_capacities(self):
+        source = """
+#include <helpers/ota/OtaManager.h>
+static_assert(OTA_BOUNDED_SOURCE_PROOFS == EXPECT_BOUNDED, "wrong source policy");
+#if OTA_BOUNDED_SOURCE_PROOFS
+static_assert(sizeof(mesh::ota::SourceMerkleProof<OTA_PROOFGEN_SCRATCH / 4, OTA_LEAVES_FRAG>) <= 1536,
+              "bounded source storage exceeds the actual allocation budget");
+#endif
+#if OTA_PROOFGEN_SCRATCH == 16383
+static_assert(OTA_PROOFGEN_SCRATCH / 4 == 4095, "odd byte limit lost usable leaf capacity");
+#endif
+"""
+        base = [os.environ.get("CXX", "g++"), "-std=c++17", "-fsyntax-only", "-x", "c++", "-",
+                "-I", str(ROOT / "src"), "-DESP32_PLATFORM=1", "-DCOMPANION_RADIO_FULL=1"]
+        for scratch in (None, 0, 1, 3, 4, 5, 16383, 16384, 16385, 32768):
+            bounded = scratch is None or 4 <= scratch <= 16384
+            flags = [] if scratch is None else [f"-DOTA_PROOFGEN_SCRATCH={scratch}"]
+            with self.subTest(scratch=scratch):
+                result = subprocess.run([*base, *flags, f"-DEXPECT_BOUNDED={int(bounded)}"],
+                                        input=source, text=True, capture_output=True, timeout=30)
+                self.assertEqual(result.returncode, 0, result.stderr)
+        for scratch in (0, 3, 16385):
+            with self.subTest(forced_invalid=scratch):
+                result = subprocess.run([*base, f"-DOTA_PROOFGEN_SCRATCH={scratch}",
+                                         "-DOTA_BOUNDED_SOURCE_PROOFS=1", "-DEXPECT_BOUNDED=1"],
+                                        input=source, text=True, capture_output=True, timeout=30)
+                self.assertNotEqual(result.returncode, 0)
+                self.assertIn("Bounded source proofs require 4..16384 leaf bytes", result.stderr)
+
+    def test_serial_metadata_stays_one_transaction_across_capacity_changes(self):
+        native = (ROOT / "test/test_ota/test_ota_core.cpp").read_text()
+        fake = native[native.index("class FakeMotaSeederStream"):
+                      native.index("class FakeDeflateMotaSeederStream")]
+        code = r'''
+#include <cassert>
+#include <vector>
+#include <helpers/ota/MotaSourceSerial.h>
+#include <helpers/ota/MotaSeederProto.h>
+#include <helpers/ota/OtaByteIO.h>
+using namespace mesh::ota;
+@FAKE@
+class MtuSource : public SerialMotaSource {
+public:
+  using SerialMotaSource::SerialMotaSource;
+  uint16_t frame = 197;
+protected:
+  uint16_t responseFrameCapacity() const override { return frame; }
+};
+int main() {
+  FakeMotaSeederStream stream;
+  MtuSource source(stream, MotaStreamWritePolicy::NoFlush);
+  uint8_t bytes[192];
+  for (uint16_t frame : {197,182,176,20}) {
+    source.frame = frame;
+    const auto before = stream.lengths.size();
+    const auto quantum = source.read_quantum();
+    assert(quantum == frame - 5);
+    assert(source.read_metadata(0, 99, bytes, quantum));
+    assert(stream.lengths.size() == before + 1 && stream.lengths.back() == quantum);
+    for (unsigned i = 0; i < quantum; ++i) assert(bytes[i] == uint8_t(99 + i));
+  }
+  source.frame = 197;
+  const auto old_quantum = source.read_quantum();
+  source.frame = 20;
+  const auto before = stream.lengths.size();
+  assert(!source.read_metadata(0, 0, bytes, old_quantum));
+  source.frame = 0;
+  assert(!source.read_metadata(0, 0, bytes, 1));
+  assert(stream.lengths.size() == before);
+}
+'''.replace("@FAKE@", fake)
+        with tempfile.TemporaryDirectory(prefix="ota-source-quantum-") as folder:
+            source, binary = Path(folder) / "test.cpp", Path(folder) / "test"
+            source.write_text(code)
+            built = subprocess.run([
+                os.environ.get("CXX", "g++"), "-std=c++17",
+                "-fsanitize=address,undefined", "-fno-sanitize-recover=all", "-fno-pie", "-no-pie",
+                "-DESP32_PLATFORM=1", "-DCOMPANION_RADIO_FULL=1",
+                "-I", str(ROOT / "src"), "-I", str(ROOT / "test/mocks"), str(source),
+                str(ROOT / "src/helpers/ota/MotaSourceSerial.cpp"), "-o", str(binary),
+            ], capture_output=True, text=True, timeout=60)
+            self.assertEqual(built.returncode, 0, built.stdout + built.stderr)
+            ran = subprocess.run([str(binary)], capture_output=True, text=True, timeout=20)
+            self.assertEqual(ran.returncode, 0, ran.stdout + ran.stderr)
+
+    def test_grouped_equivalence_faults_and_manager_lifetimes(self):
+        # Reuse the existing source/wire/allocator harness, retaining its legacy
+        # tests unchanged in their own suite. Only its helpers precede this point.
+        prefix = (ROOT / "test/fixtures/ota_cold_buffers/test.cpp").read_text().split(
+            "static void host_allocation_and_fallbacks()", 1)[0]
+        prefix = prefix.replace("(data[0] == OTA_MANIFEST && !sent.accept_manifest)",
+                                "((data[0] == OTA_MANIFEST || data[0] == OTA_LEAVES) && !sent.accept_manifest)")
+        code = prefix + (ROOT / "test/fixtures/ota_source_proofs/test.cpp").read_text()
+        with tempfile.TemporaryDirectory(prefix="ota-source-proofs-") as folder:
+            source, binary = Path(folder) / "test.cpp", Path(folder) / "test"
+            source.write_text(code)
+            files = ("OtaManager.cpp", "OtaProtocol.cpp", "MotaContainer.cpp", "MerkleTree.cpp")
+            result = subprocess.run([
+                os.environ.get("CXX", "g++"), "-std=c++17", "-O1", "-Wall", "-Wextra", "-Werror",
+                "-Wno-unused-parameter", "-Wno-unused-function", "-Wno-sign-compare",
+                "-fsanitize=address,undefined", "-fno-sanitize-recover=all", "-fno-pie", "-no-pie",
+                "-DESP32_PLATFORM=1", "-DCOMPANION_RADIO_FULL=1",
+                "-I", str(ROOT / "src"), "-I", str(ROOT / "test/mocks"), str(source),
+                *[str(ROOT / "src/helpers/ota" / name) for name in files], str(ROOT / "src/Utils.cpp"),
+                "-Wl,--wrap=malloc", "-Wl,--wrap=free", "-o", str(binary),
+            ], capture_output=True, text=True, timeout=90)
+            self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+            ran = subprocess.run([str(binary)], capture_output=True, text=True, timeout=120)
+            self.assertEqual(ran.returncode, 0, ran.stdout + ran.stderr)
+            for marker in ("byte-identical grouped proofs", "partial reads and source changes",
+                           "bounded manager ownership and backpressure", "simultaneous capture and serving"):
+                self.assertIn("PASS: " + marker, ran.stdout)
+            print(ran.stdout, flush=True)
+
+
+if __name__ == "__main__":
+    unittest.main()

variants/heltec_v3/platformio.ini

@@ -396,6 +396,19 @@ board_build.flash_mode = ${esp32_s3_full.board_build.flash_mode}
 build_flags =
   ${env:Heltec_v3_companion_radio_wifi_mqtt.build_flags}
   -D MESHCORE_REQUIRES_COMPANION_RADIO_FULL=1
+  ; The Companion is a single-session GATT peripheral. Keep its advertising,
+  ; scan responses and authenticated NUS service with the smaller BLE host.
+  -D MESH_USE_NIMBLE_ARDUINO=1
+  -D CONFIG_BT_NIMBLE_MAX_CONNECTIONS=1
+  ; Preserve the SDK's 15 saved phones and both NUS/Service Changed CCCDs.
+  -D CONFIG_BT_NIMBLE_MAX_BONDS=15
+  -D CONFIG_BT_NIMBLE_MAX_CCCDS=30
+  -D CONFIG_BT_NIMBLE_ROLE_CENTRAL_DISABLED=1
+  -D CONFIG_BT_NIMBLE_ROLE_OBSERVER_DISABLED=1
+lib_deps =
+  ${env:Heltec_v3_companion_radio_wifi_mqtt.lib_deps}
+  h2zero/NimBLE-Arduino @ 2.5.1
+lib_ignore = ESP32 BLE Arduino
 
 [env:Heltec_v3_sensor]
 extends = Heltec_lora32_v3