Qualify Full Meshtext builds and share cold storage on constrained companions

build_legacy.sh

@@ -4541,6 +4541,10 @@ apply_companion_radio_full_profile() {
     heltec_tracker_v2_companion_radio_full_*)
       append_platformio_build_unflags "-DMAX_CONTACTS=350 -DOFFLINE_QUEUE_SIZE=256"
       export PLATFORMIO_BUILD_FLAGS="${PLATFORMIO_BUILD_FLAGS} -DMAX_CONTACTS=100 -DOFFLINE_QUEUE_SIZE=224"
+      # Reuse idle queue storage during explicit OTA; retain all 224 normal slots.
+      if [[ "${env_name,,}" != generic_espnow_companion_radio_full ]]; then
+        export PLATFORMIO_BUILD_FLAGS="${PLATFORMIO_BUILD_FLAGS} -DOTA_SHARED_COMPANION_QUEUE=1 -DONE_KEY_DM_COMPACT_HELD=1"
+      fi
       record_build_reduction \
         "companion.capacity limited to 100 contacts for runtime RAM; 224 queued frames and all Full transports retained"
       ;;
@@ -4678,6 +4682,76 @@ copy_build_output() {
   cp -- "$source_path" "$output_path"
 }
 
+requires_meshtext_companion_code() {
+  local env_name=$1
+  local pio_env_name=$2
+  local decision_status=0
+  is_companion_radio_full_target "$env_name" || return 1
+
+  # Check the effective recipe and overlay, including explicit feature-off
+  # builds and the Indicator LoRa recipe's inherited ESP-NOW undefine.
+  python3 -c '
+import json, os, re, shlex, sys
+rows = {name: dict(options) for name, options in json.load(sys.stdin)}
+options = rows.get("env:" + sys.argv[1], {})
+
+def operations(value):
+    words = iter(shlex.split(value if isinstance(value, str) else "\n".join(map(str, value))))
+    result = []
+    for word in words:
+        if word in ("-D", "-U"):
+            word += next(words, "")
+        match = re.fullmatch(r"-([DU])(\w+)(?:=(.*))?", word)
+        if match:
+            result.append(match.groups())
+    return result
+
+flags = operations(options.get("build_flags", []))
+flags += operations(os.environ.get("PLATFORMIO_BUILD_FLAGS", ""))
+removed = set(operations(options.get("build_unflags", []))
+              + operations(os.environ.get("PLATFORMIO_BUILD_UNFLAGS", "")))
+flags = [flag for flag in flags if flag not in removed
+         and not (flag[0] == "D" and ("D", flag[1], None) in removed)]
+defines = {name: "1" if value is None else value
+           for kind, name, value in flags if kind == "D"}
+# PlatformIO emits CPPDEFINES before CCFLAGS, so every surviving -U wins.
+for kind, name, _value in flags:
+    if kind == "U":
+        defines.pop(name, None)
+
+def enabled(name, default=False, seen=()):
+    if name not in defines:
+        return default
+    if name in seen:
+        return None
+    value = defines[name].strip().strip("() ")
+    if value in ("false", "true"):
+        return value == "true"
+    if re.fullmatch(r"[A-Za-z_]\w*", value):
+        return enabled(value, False, (*seen, name))
+    number = re.fullmatch(r"([+-]?(?:0[xX][0-9a-fA-F]+|[0-9]+))[uUlL]*", value)
+    if number:
+        value = number[1]
+        base = 16 if "x" in value.lower() else 8 if re.fullmatch(r"[+-]?0[0-7]+", value) else 10
+        return int(value, base) != 0
+    return None
+
+# Explicit enablement overrides the normal ESP-NOW exclusion, as in the header.
+# Unknown expressions retain the requirement rather than silently dropping it.
+if "MESH_ENABLE_MESHTEXT_CHAT" in defines:
+    required = enabled("MESH_ENABLE_MESHTEXT_CHAT") is not False
+else:
+    required = (enabled("COMPANION_RADIO_FULL", True) is not False
+                and enabled("MESH_PRIMARY_ESPNOW") is not True)
+sys.exit(0 if required else 3)
+' "$pio_env_name" <<<"$PIO_CONFIG_JSON" || decision_status=$?
+  case "$decision_status" in
+    0) return 0 ;;
+    3) return 1 ;;
+    *) return 2 ;;
+  esac
+}
+
 write_build_capability_manifest() {
   local env_name=$1
   local env_platform=$2
@@ -4705,10 +4779,15 @@ write_build_capability_manifest() {
       || is_xiao_qspi_canonical_repeater_build "$env_name"; then
     checker_args+=(--require-ota)
   fi
-  if [ "$env_name" = "RAK_4631_companion_radio_full" ]; then
-    # This profile is offered for MeshTalk testing. Check executable codec and
-    # Companion transport definitions, not a channel name or CLI help string.
+  local meshtext_status=0
+  requires_meshtext_companion_code "$env_name" "$pio_env_name" || meshtext_status=$?
+  if [ "$meshtext_status" -eq 0 ]; then
+    # Every enabled Full LoRa Companion must retain executable codec and
+    # transport definitions, not merely a channel name or CLI help string.
     checker_args+=(--require-meshtext)
+  elif [ "$meshtext_status" -ne 1 ]; then
+    echo "Cannot determine the Full Companion Meshtext capability contract." >&2
+    return "$meshtext_status"
   fi
   if [ "$env_platform" = "ESP32_PLATFORM" ]; then
     checker_args+=(--firmware-bin "${build_output_dir}/firmware.bin"

docs/radio_profiles.md

@@ -568,6 +568,9 @@ accepts an `mt-` name; a factory reset cannot add code omitted from an image.
 The common default and modem support do not establish physical qualification
 for every board. Each release image must still pass its flash and runtime RAM
 budgets, and hardware interoperability must be checked separately.
+For example, [LLCC68 modulation limits](https://www.semtech.com/products/wireless-rf/lora-connect/llcc68)
+exclude LongFast (SF11 at BW250): enabling the backend cannot add that modem
+combination. Compatible presets such as MediumSlow or LongTurbo remain available.
 
 In an enabled build, an `mt-` name always requests Meshtext, even if radio2 is
 off or setup is blocked. It never falls back to sending MeshCore frames. In a

examples/companion_radio/MyMesh.cpp

@@ -567,9 +567,9 @@ uint16_t MyMesh::drainOfflineQueueForOta(void* owner) {
   MyMesh* mesh = static_cast<MyMesh*>(owner);
   uint8_t discarded[MAX_FRAME_SIZE];
   uint16_t removed = 0;
-  // Explicit OTA work borrows the upper half of this volatile queue. Keep the
-  // newest 128 frames and notify the UI for each oldest frame consumed.
-  while (mesh->offline_queue_len > 128) {
+  // Explicit OTA work borrows the queue's workspace region. Keep the newest
+  // retained frames and notify the UI for each oldest frame consumed.
+  while (mesh->offline_queue_len > static_cast<int>(OTA_RETAINED_OFFLINE_FRAMES)) {
     if (mesh->getFromOfflineQueue(discarded) <= 0) break;
     ++removed;
   }
@@ -1442,8 +1442,14 @@ bool MyMesh::onAddressedTextPacket(mesh::Packet* packet, uint8_t src_hash,
     uint8_t id[ONE_KEY_DM_ID_SIZE];
     makeOneKeyDMId(id, timestamp, text);
     for (uint8_t j = 0; j < held_dm_count; ++j) {
+#if ONE_KEY_DM_COMPACT_HELD
+      uint8_t held_id[ONE_KEY_DM_ID_SIZE];
+      makeOneKeyDMId(held_id, held_dms[j].timestamp, held_dms[j].text);
+#else
+      const uint8_t* held_id = held_dms[j].id;
+#endif
       if (memcmp(held_dms[j].sender_key, pub_key, PUB_KEY_SIZE) == 0 &&
-          memcmp(held_dms[j].id, id, ONE_KEY_DM_ID_SIZE) == 0) return true;
+          memcmp(held_id, id, ONE_KEY_DM_ID_SIZE) == 0) return true;
     }
 
     // Only verified and decryptable messages count toward the FIFO of 15.
@@ -1483,8 +1489,19 @@ bool MyMesh::onAddressedTextPacket(mesh::Packet* packet, uint8_t src_hash,
     }
     HeldOneKeyDM& entry = held_dms[held_dm_count++];
     memcpy(entry.sender_key, pub_key, PUB_KEY_SIZE);
+#if ONE_KEY_DM_COMPACT_HELD
+    entry.timestamp = timestamp;
+    // Preserve the exact ACK of the authenticated original plaintext,
+    // including its attempt/type byte; the delivery ID is derived separately.
+    mesh::Utils::sha256(entry.ack_hash, sizeof(entry.ack_hash), data,
+                        5 + strlen(text), pub_key, PUB_KEY_SIZE);
+    entry.path_len = packet->isRouteFlood() ? packet->path_len : 0xFF;
+    entry.snr = packet->_snr;
+    memcpy(entry.text, text, strlen(text) + 1);
+#else
     memcpy(entry.id, id, ONE_KEY_DM_ID_SIZE);
     entry.packet = *packet;
+#endif
 #endif
     return true;
   }
@@ -1541,6 +1558,19 @@ void MyMesh::releaseHeldOneKeyDMs() {
 #else
     if (offline_queue_len >= getOfflineQueueCapacity()) break;
 
+#if ONE_KEY_DM_COMPACT_HELD
+    uint8_t id[ONE_KEY_DM_ID_SIZE];
+    makeOneKeyDMId(id, entry.timestamp, entry.text);
+    if (!wasDeliveredOneKeyDM(entry.sender_key, id)) {
+      mesh::Packet packet;
+      packet.header = entry.path_len == 0xFF ? ROUTE_TYPE_DIRECT : ROUTE_TYPE_FLOOD;
+      packet.path_len = entry.path_len == 0xFF ? 0 : entry.path_len;
+      packet._snr = entry.snr;
+      onMessageRecv(*contact, &packet, entry.timestamp, entry.text);
+      rememberDeliveredOneKeyDM(entry.sender_key, id);
+    }
+    sendAckTo(*contact, entry.ack_hash, sizeof(entry.ack_hash));
+#else
     uint8_t data[MAX_PACKET_PAYLOAD + 1];
     const int decoded_len = mesh::Utils::MACThenDecrypt(
         contact->getSharedSecret(self_id), data, &entry.packet.payload[2],
@@ -1563,6 +1593,7 @@ void MyMesh::releaseHeldOneKeyDMs() {
         }
       }
     }
+#endif
     for (uint8_t j = i + 1; j < held_dm_count; ++j) held_dms[j - 1] = held_dms[j];
     --held_dm_count;
 #endif
@@ -2062,8 +2093,14 @@ void MyMesh::onMessageRecv(const ContactInfo &from, mesh::Packet *pkt, uint32_t
   bool replaced_held_dm = false;
 #endif
   for (uint8_t i = 0; i < held_dm_count; ++i) {
+#if ONE_KEY_DM_COMPACT_HELD
+    uint8_t held_id[ONE_KEY_DM_ID_SIZE];
+    makeOneKeyDMId(held_id, held_dms[i].timestamp, held_dms[i].text);
+#else
+    const uint8_t* held_id = held_dms[i].id;
+#endif
     if (memcmp(held_dms[i].sender_key, from.id.pub_key, PUB_KEY_SIZE) == 0 &&
-        memcmp(held_dms[i].id, dm_id, ONE_KEY_DM_ID_SIZE) == 0) {
+        memcmp(held_id, dm_id, ONE_KEY_DM_ID_SIZE) == 0) {
 #if ONE_KEY_DM_SHARED_OFFLINE_QUEUE
       // Removing the held frame frees exactly one ordinary slot for this retry.
       removeHeldOneKeyDM(i);

examples/companion_radio/MyMesh.h

@@ -124,6 +124,18 @@ namespace mesh { class BeaconManager; }
 #define ONE_KEY_DM_SHARED_OFFLINE_QUEUE 0
 #endif
 
+// Constrained Full recipes may retain only the authenticated plain DM data.
+// This is independent of the ordinary offline queue and its OTA workspace.
+#ifndef ONE_KEY_DM_COMPACT_HELD
+#define ONE_KEY_DM_COMPACT_HELD 0
+#endif
+#if ONE_KEY_DM_COMPACT_HELD != 0 && ONE_KEY_DM_COMPACT_HELD != 1
+#error "ONE_KEY_DM_COMPACT_HELD must be 0 or 1"
+#endif
+#if ONE_KEY_DM_COMPACT_HELD && ONE_KEY_DM_SHARED_OFFLINE_QUEUE
+#error "Compact held DMs and shared held-DM queue storage are separate policies"
+#endif
+
 #ifndef OFFLINE_QUEUE_SIZE
 #if defined(ESP32_PLATFORM) && defined(BOARD_HAS_PSRAM)
 #define OFFLINE_QUEUE_SIZE 512
@@ -955,8 +967,13 @@ private:
   // Local onboarding only: RAM-backed like the receive queue, not RF traffic.
   uint8_t pending_emergency_notices[(MAX_GROUP_CHANNELS + 7) / 8] = {};
 #if defined(OTA_SHARED_COMPANION_QUEUE)
-  static_assert(OFFLINE_QUEUE_SIZE == 256, "Shared mOTA queue requires 256 normal slots");
-  mesh::BorrowableFrameBuffer<Frame, OFFLINE_QUEUE_SIZE, 128, mesh::ota::OtaContext> offline_queue;
+  static constexpr unsigned OTA_RETAINED_OFFLINE_FRAMES = 128;
+  static_assert(OFFLINE_QUEUE_SIZE > OTA_RETAINED_OFFLINE_FRAMES,
+                "Shared mOTA queue must retain 128 frames and provide workspace storage");
+  // The buffer also asserts that the complete context fits in the remaining
+  // frames. Constrained ESP32 recipes keep 224 normal slots; nRF52 and Paper keep 256.
+  mesh::BorrowableFrameBuffer<Frame, OFFLINE_QUEUE_SIZE,
+                             OTA_RETAINED_OFFLINE_FRAMES, mesh::ota::OtaContext> offline_queue;
   static mesh::ota::OtaContext* acquireOfflineQueueForOta(void* owner);
   static uint16_t drainOfflineQueueForOta(void* owner);
   static void releaseOfflineQueueFromOta(void* owner);
@@ -1021,11 +1038,23 @@ private:
   uint8_t verified_pending_count = 0;
   struct HeldOneKeyDM {
     uint8_t sender_key[PUB_KEY_SIZE];
+#if ONE_KEY_DM_COMPACT_HELD
+    uint32_t timestamp;
+    uint8_t ack_hash[4];
+    uint8_t path_len;
+    int8_t snr;
+    char text[MAX_TEXT_LEN + 1];
+#else
     uint8_t id[ONE_KEY_DM_ID_SIZE];
 #if !ONE_KEY_DM_SHARED_OFFLINE_QUEUE
     mesh::Packet packet;
+#endif
 #endif
   };
+#if ONE_KEY_DM_COMPACT_HELD
+  static_assert(sizeof(HeldOneKeyDM) <= PUB_KEY_SIZE + MAX_TEXT_LEN + 12,
+                "Compact held DMs must not retain a complete packet or duplicate ID");
+#endif
   HeldOneKeyDM held_dms[MAX_HELD_ONE_KEY_DMS];
   uint8_t held_dm_count = 0;
   struct DeliveredOneKeyDM {

src/helpers/ESPNowRawFragmentation.h

@@ -152,10 +152,10 @@ class ESPNowRawReassemblerT {
  public:
   static_assert(SlotCount > 0, "ESP-NOW reassembly needs at least one slot");
 
-  explicit ESPNowRawReassemblerT(uint32_t timeout_ms = 2000)
-      : _timeout_ms(timeout_ms) {
-    reset();
-  }
+  // Constant initialization also lets the linker discard the primary-radio
+  // workspace when this shared source is compiled into a LoRa-only image.
+  explicit constexpr ESPNowRawReassemblerT(uint32_t timeout_ms = 2000)
+      : _timeout_ms(timeout_ms), _slots{} {}
 
   void reset() {
     memset(_slots, 0, sizeof(_slots));

src/helpers/ota/OtaMemoryPolicy.h

@@ -24,6 +24,9 @@
 
 // Qualified Full Companions stream host-provided firmware and need no permanent
 // mOTA workspace. Apply the same policy to build.sh and direct PlatformIO builds.
+// Other qualified recipes, including the constrained ESP32 224-frame queue, opt in through
+// OTA_SHARED_COMPANION_QUEUE. The queue checks the complete context size against
+// its actual upper region and retains the same 128 frames during explicit OTA.
 #if defined(OTA_SEEDER_ONLY) && defined(COMPANION_RADIO_FULL) && \
     (defined(NRF52_PLATFORM) || \
      (defined(ESP32_PLATFORM) && defined(HELTEC_WIRELESS_PAPER)))

test/test_companion_shared_ota_capacity.py

@@ -0,0 +1,176 @@
+#!/usr/bin/env python3
+"""Exercise C3's full queue while its explicit OTA work borrows cold storage."""
+
+from pathlib import Path
+import re
+import subprocess
+import tempfile
+import unittest
+
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+def method(text, signature):
+    start = text.index(signature)
+    end = text.index("{", start) + 1
+    depth = 1
+    while depth:
+        depth += (text[end] == "{") - (text[end] == "}")
+        end += 1
+    return text[start:end]
+
+
+class SharedCompanionOtaCapacityTest(unittest.TestCase):
+    def compile(self, text, *, run=False):
+        with tempfile.TemporaryDirectory(prefix="meshcore-shared-c3-") as temp:
+            binary = Path(temp) / "queue"
+            command = ["c++", "-std=c++17", "-Wall", "-Wextra", "-Werror",
+                       "-I", str(ROOT / "src"), "-x", "c++", "-",
+                       "-o", str(binary)]
+            if run:
+                command[1:1] = ["-fsanitize=address,undefined", "-g"]
+            result = subprocess.run(command, input=text, text=True,
+                                    capture_output=True)
+            if run:
+                self.assertEqual(result.returncode, 0, result.stderr)
+                result = subprocess.run([str(binary)], text=True,
+                                        capture_output=True)
+                self.assertEqual(result.returncode, 0, result.stderr)
+            return result
+
+    def test_224_normal_slots_and_128_retained_slots_preserve_fifo_and_ownership(self):
+        source = (ROOT / "examples/companion_radio/MyMesh.cpp").read_text()
+        declaration = (ROOT / "examples/companion_radio/MyMesh.h").read_text()
+        retained = re.search(
+            r"static constexpr unsigned OTA_RETAINED_OFFLINE_FRAMES = (\d+);",
+            declaration)[1]
+        methods = "\n".join(method(source, signature) for signature in (
+            "mesh::ota::OtaContext* MyMesh::acquireOfflineQueueForOta(void* owner)",
+            "uint16_t MyMesh::drainOfflineQueueForOta(void* owner)",
+            "void MyMesh::releaseOfflineQueueFromOta(void* owner)",
+        ))
+        harness = r'''
+#include <helpers/BorrowableFrameBuffer.h>
+#include <cassert>
+#include <cstdint>
+#include <cstring>
+#include <deque>
+constexpr unsigned MAX_FRAME_SIZE = 176;
+namespace mesh { namespace ota {
+struct OtaContext {
+  static unsigned live, constructions, destructions;
+  uint8_t bytes[11284]; // source-only ESP32 linked context, not a RAM budget
+  OtaContext() {
+    ++live; ++constructions;
+    memset(bytes, 0xA5, sizeof bytes);
+  }
+  ~OtaContext() {
+    --live; ++destructions;
+    memset(bytes, 0x5A, sizeof bytes);
+  }
+};
+unsigned OtaContext::live = 0, OtaContext::constructions = 0,
+         OtaContext::destructions = 0;
+} }
+struct MyMesh {
+  struct Frame { uint8_t len; uint8_t buf[MAX_FRAME_SIZE]; };
+  static constexpr unsigned OTA_RETAINED_OFFLINE_FRAMES = @RETAINED@;
+  mesh::BorrowableFrameBuffer<Frame, 224, OTA_RETAINED_OFFLINE_FRAMES,
+                             mesh::ota::OtaContext> offline_queue;
+  int offline_queue_len = 0, offline_queue_head = 0;
+  unsigned next = 0, notifications = 0;
+  std::deque<unsigned> expected;
+  static mesh::ota::OtaContext* acquireOfflineQueueForOta(void* owner);
+  static uint16_t drainOfflineQueueForOta(void* owner);
+  static void releaseOfflineQueueFromOta(void* owner);
+  void add() {
+    assert(offline_queue_len < int(offline_queue.capacity()));
+    Frame& frame = offline_queue.at(
+        (offline_queue_head + offline_queue_len) % offline_queue.capacity());
+    frame.len = sizeof frame.buf;
+    memset(frame.buf, next % 251, sizeof frame.buf);
+    memcpy(frame.buf, &next, sizeof next);
+    expected.push_back(next++);
+    ++offline_queue_len;
+  }
+  int getFromOfflineQueue(uint8_t* bytes) {
+    if (!offline_queue_len) return 0;
+    Frame& frame = offline_queue.at(offline_queue_head);
+    unsigned sequence;
+    memcpy(&sequence, frame.buf, sizeof sequence);
+    assert(frame.len == sizeof frame.buf && sequence == expected.front());
+    for (unsigned i = sizeof sequence; i < sizeof frame.buf; ++i)
+      assert(frame.buf[i] == sequence % 251);
+    memcpy(bytes, frame.buf, frame.len);
+    expected.pop_front();
+    --offline_queue_len;
+    offline_queue_head = offline_queue_len
+        ? (offline_queue_head + 1) % offline_queue.capacity() : 0;
+    ++notifications;
+    return frame.len;
+  }
+  void pop() { uint8_t bytes[MAX_FRAME_SIZE]; assert(getFromOfflineQueue(bytes)); }
+};
+@METHODS@
+int main() {
+  using mesh::ota::OtaContext;
+  MyMesh mesh;
+  for (unsigned head = 0; head < 224; ++head) {
+    while (mesh.offline_queue_len < 224) mesh.add();
+    // Visit every wrapped head with a full 224-frame normal queue.
+    for (unsigned i = 0; i < head; ++i) { mesh.pop(); mesh.add(); }
+    int original_head = mesh.offline_queue_head;
+    unsigned constructed = OtaContext::constructions;
+    assert(!MyMesh::acquireOfflineQueueForOta(&mesh));
+    assert(mesh.offline_queue_len == 224);
+    assert(mesh.offline_queue_head == original_head);
+    assert(OtaContext::constructions == constructed && OtaContext::live == 0);
+    // Only the explicit drain consumes frames, preserving the newest 128.
+    unsigned notified = mesh.notifications;
+    assert(MyMesh::drainOfflineQueueForOta(&mesh) == 96);
+    assert(mesh.notifications == notified + 96 && mesh.offline_queue_len == 128);
+    assert(MyMesh::drainOfflineQueueForOta(&mesh) == 0);
+    OtaContext* owner = MyMesh::acquireOfflineQueueForOta(&mesh);
+    assert(owner && OtaContext::live == 1 && mesh.offline_queue.capacity() == 128);
+    assert(OtaContext::constructions == constructed + 1);
+    assert(MyMesh::acquireOfflineQueueForOta(&mesh) == owner);
+    assert(OtaContext::constructions == constructed + 1);
+    memset(owner->bytes, 0xCC, sizeof owner->bytes);
+    for (unsigned i = 0; i < head % 128; ++i) { mesh.pop(); mesh.add(); }
+    MyMesh::releaseOfflineQueueFromOta(&mesh);
+    assert(OtaContext::live == 0 && mesh.offline_queue.capacity() == 224);
+    assert(OtaContext::constructions == OtaContext::destructions);
+    MyMesh::releaseOfflineQueueFromOta(&mesh); // idle release cannot destroy twice
+    while (mesh.offline_queue_len < 224) mesh.add();
+    while (mesh.offline_queue_len) mesh.pop();
+    assert(mesh.expected.empty());
+  }
+}
+'''.replace("@RETAINED@", retained).replace("@METHODS@", methods)
+        self.compile(harness, run=True)
+
+    def test_insufficient_workspace_or_no_retained_region_cannot_compile(self):
+        harness = r'''
+#include <helpers/BorrowableFrameBuffer.h>
+#include <cstdint>
+struct Frame { uint8_t len; uint8_t bytes[176]; };
+struct Workspace { uint8_t bytes[@SIZE@]; };
+mesh::BorrowableFrameBuffer<Frame, @CAPACITY@, 128, Workspace> queue;
+int main() { return queue.capacity() == @CAPACITY@ ? 0 : 1; }
+'''
+        # Exact region bounds and the existing 256/128 geometry remain valid.
+        for capacity, size, accepted in ((224, 96 * 177, True),
+                                         (224, 96 * 177 + 1, False),
+                                         (256, 19624, True),
+                                         (128, 1, False)):
+            with self.subTest(capacity=capacity, size=size):
+                result = self.compile(harness.replace("@CAPACITY@", str(capacity))
+                                      .replace("@SIZE@", str(size)))
+                self.assertEqual(result.returncode == 0, accepted, result.stderr)
+                if not accepted:
+                    self.assertIn("static assertion failed", result.stderr)
+
+
+if __name__ == "__main__":
+    unittest.main()

test/test_espnow_raw_initialization.py

@@ -0,0 +1,101 @@
+#!/usr/bin/env python3
+"""Check cold ESP-NOW storage initialization and unused linker reclamation."""
+
+from pathlib import Path
+import subprocess
+import tempfile
+import unittest
+
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+class EspNowRawInitializationTest(unittest.TestCase):
+    def test_constexpr_initialization_does_not_root_an_unused_workspace(self):
+        with tempfile.TemporaryDirectory(prefix="espnow-unused-init-") as temp:
+            workspace = Path(temp) / "workspace.cpp"
+            workspace.write_text(r'''
+#include <helpers/ESPNowRawFragmentation.h>
+constexpr mesh::espnow::ESPNowRawReassembler constant_initialized(1234);
+mesh::espnow::ESPNowRawReassembler unused_reassembler;
+''')
+            main = Path(temp) / "main.cpp"
+            main.write_text("int main() { return 0; }\n")
+            binary = Path(temp) / "without-espnow"
+            # Deliberately disable optimization/LTO: only the linker's real
+            # section reclamation may drop this externally visible global.
+            result = subprocess.run([
+                "c++", "-std=c++11", "-O0", "-fno-lto",
+                "-ffunction-sections", "-fdata-sections", "-Wl,--gc-sections",
+                "-I", str(ROOT / "src"), str(workspace), str(main),
+                "-o", str(binary),
+            ], text=True, capture_output=True)
+            self.assertEqual(result.returncode, 0, result.stderr)
+            symbols = subprocess.run(["nm", "--defined-only", str(binary)],
+                                     text=True, capture_output=True, check=True).stdout
+            self.assertNotIn("unused_reassembler", symbols)
+            self.assertNotIn("_GLOBAL__sub_I_unused_reassembler", symbols)
+            subprocess.run([str(binary)], check=True)
+
+    def test_default_and_custom_timeouts_with_fragmentation_reset_and_rollover(self):
+        harness = r'''
+#include <helpers/ESPNowRawFragmentation.h>
+#include <cassert>
+#include <cstdint>
+#include <cstring>
+#include <new>
+using namespace mesh::espnow;
+ESPNowRawReassembler default_initialized;
+int main() {
+  uint8_t packet[255], output[255], mac[6] = {2, 1, 2, 3, 4, 5};
+  for (unsigned i = 0; i < sizeof packet; ++i) packet[i] = i;
+  ESPNowRawFrames frames;
+  assert(encodeEspNowRawFrames(packet, sizeof packet, frames));
+  size_t size = 99;
+  assert(default_initialized.expire(0) == 0);
+  assert(default_initialized.acceptFrame(mac, frames.data[0], frames.lengths[0],
+      1, output, sizeof output, size) == ESPNowRawReassemblyResult::FRAGMENT_STORED);
+  assert(default_initialized.expire(2000) == 0);
+  assert(default_initialized.expire(2001) == 1); // unchanged 2000 ms default
+  // A placement-new object must clear arbitrary prior bytes, not depend on
+  // global .bss zeroing. Custom initialization preserves its exact timeout.
+  alignas(ESPNowRawReassemblerT<1>) uint8_t arena[sizeof(ESPNowRawReassemblerT<1>)];
+  memset(arena, 0xCC, sizeof arena);
+  auto* reassembler = new (arena) ESPNowRawReassemblerT<1>(20);
+  assert(reassembler->expire(1000) == 0);
+  assert(reassembler->acceptFrame(mac, frames.data[0], frames.lengths[0],
+      0xFFFFFFF0UL, output, sizeof output, size)
+      == ESPNowRawReassemblyResult::FRAGMENT_STORED);
+  assert(reassembler->expire(3) == 0);
+  assert(reassembler->expire(4) == 1);
+  assert(reassembler->acceptFrame(mac, frames.data[1], frames.lengths[1],
+      5, output, sizeof output, size) == ESPNowRawReassemblyResult::FRAGMENT_STORED);
+  assert(reassembler->acceptFrame(mac, frames.data[0], frames.lengths[0],
+      6, output, sizeof output, size) == ESPNowRawReassemblyResult::PACKET_COMPLETE);
+  assert(size == sizeof packet && !memcmp(output, packet, sizeof packet));
+  assert(reassembler->acceptFrame(mac, frames.data[0], frames.lengths[0],
+      7, output, sizeof output, size) == ESPNowRawReassemblyResult::DUPLICATE);
+  reassembler->reset();
+  assert(reassembler->expire(1000) == 0);
+  assert(reassembler->acceptFrame(mac, frames.data[0], frames.lengths[0],
+      8, output, sizeof output, size) == ESPNowRawReassemblyResult::FRAGMENT_STORED);
+  assert(reassembler->acceptFrame(mac, frames.data[1], frames.lengths[1],
+      9, output, sizeof output, size) == ESPNowRawReassemblyResult::PACKET_COMPLETE);
+  assert(size == sizeof packet && !memcmp(output, packet, sizeof packet));
+  reassembler->~ESPNowRawReassemblerT<1>();
+}
+'''
+        with tempfile.TemporaryDirectory(prefix="espnow-live-init-") as temp:
+            binary = Path(temp) / "reassembler"
+            result = subprocess.run([
+                "c++", "-std=c++11", "-Wall", "-Wextra", "-Werror",
+                "-fsanitize=address,undefined", "-g", "-I", str(ROOT / "src"),
+                "-x", "c++", "-", "-o", str(binary),
+            ], input=harness, text=True, capture_output=True)
+            self.assertEqual(result.returncode, 0, result.stderr)
+            result = subprocess.run([str(binary)], text=True, capture_output=True)
+            self.assertEqual(result.returncode, 0, result.stderr)
+
+
+if __name__ == "__main__":
+    unittest.main()

test/test_meshtext_license.py

@@ -1,6 +1,7 @@
 """Qualify linked chat code and ship its MIT notice without touching firmware."""
 import hashlib
 import json
+import os
 from pathlib import Path
 import shutil
 import struct
@@ -42,6 +43,107 @@ def elf32(path, symbols=SYMBOLS, flags=6, undefined=(), zero=()):
 
 
 class MeshtextLicenseTest(unittest.TestCase):
+    def builder_contract(self, target, platform, *, flags=(), overlay="", unflags="",
+                         config_unflags=(), symbols=("unrelated_defined_function",), invalid_config=False):
+        with tempfile.TemporaryDirectory(prefix="meshtext-build-contract-") as folder:
+            root = Path(folder)
+            build = root / "build/base"
+            build.mkdir(parents=True)
+            elf32(build / "firmware.elf", symbols=symbols)
+            config = [["env:base", [["build_flags", ["-DCOMPANION_RADIO_FULL=1", *flags]],
+                                    ["build_unflags", list(config_unflags)]]]]
+            script = r'''
+set -euo pipefail
+pio() { printf 'unexpected PlatformIO invocation\n' >&2; return 99; }
+source build.sh
+PIO_ENV_PLATFORM_BY_NAME[$1]=$2
+PIO_CONFIG_JSON=$MESH_TEST_CONFIG
+OUTPUT_DIR=$MESH_TEST_DIRECTORY/out
+PIO_BUILD_DIR_OVERRIDE=$MESH_TEST_DIRECTORY/build
+BUILD_PROFILE_FOR_TARGET=full
+BUILD_RECIPE_SHA256=test
+BUILD_CAPABILITIES=(); BUILD_EXPECTATIONS=()
+BUILD_APPLICATION_EXPECTATIONS=(); BUILD_REDUCTIONS=()
+requires_wireless_self_update() { return 1; }
+is_xiao_qspi_canonical_repeater_build() { return 1; }
+is_nrf52_sensor_ota_pair_target() { return 1; }
+embedded_bootloader_artifact_target() { printf '%s\n' "$1"; }
+python3() {
+  # Only the unrelated recipe attachment is substituted. The real builder
+  # decision and ELF capability checker must reject omitted chat code.
+  if [ "$1" = scripts/firmware_build_recipe.py ]; then return 0; fi
+  command "$MESH_TEST_PYTHON" "$@"
+}
+write_build_capability_manifest "$1" "$2" base fixture
+'''
+            result = subprocess.run(["bash", "-c", script, "meshtext-contract", target, platform],
+                cwd=ROOT, text=True, capture_output=True, env={**os.environ,
+                    "MESH_TEST_CONFIG": "invalid JSON" if invalid_config else json.dumps(config),
+                    "MESH_TEST_DIRECTORY": str(root), "MESH_TEST_PYTHON": sys.executable,
+                    "PLATFORMIO_BUILD_FLAGS": overlay, "PLATFORMIO_BUILD_UNFLAGS": unflags})
+            manifest = root / "out/fixture.capabilities.json"
+            return result, json.loads(manifest.read_text()) if manifest.exists() else None
+
+    def test_builder_requires_real_chat_on_every_full_companion_family(self):
+        for target, platform in (
+            ("RAK_4631_companion_radio_full", "NRF52_PLATFORM"),
+            ("RAK_3401_companion_radio_full", "NRF52_PLATFORM"),
+            ("t1000e_companion_radio_full", "NRF52_PLATFORM"),
+            ("Xiao_nrf52_companion_radio_full", "NRF52_PLATFORM"),
+            ("Heltec_v2_companion_radio_full", "ESP32_PLATFORM"),
+            ("heltec_v4_companion_radio_full_femon", "ESP32_PLATFORM"),
+            ("Xiao_C3_companion_radio_full", "ESP32_PLATFORM"),
+            ("Xiao_C6_companion_radio_full_", "ESP32_PLATFORM"),
+        ):
+            with self.subTest(target=target):
+                result, manifest = self.builder_contract(target, platform)
+                self.assertEqual(result.returncode, 1, result.stdout + result.stderr)
+                self.assertFalse(manifest["verified"])
+                self.assertIn("promised", result.stderr)
+        result, manifest = self.builder_contract("t1000e_companion_radio_full", "NRF52_PLATFORM",
+                                                 symbols=SYMBOLS)
+        self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+        self.assertTrue(manifest["verified"])
+        self.assertIn(license.CAPABILITY, manifest["capabilities"])
+
+    def test_builder_preserves_effective_feature_overrides_and_espnow_exclusion(self):
+        cases = (
+            (dict(flags=["-DMESH_ENABLE_MESHTEXT_CHAT=0"]), False),
+            (dict(overlay="-D MESH_ENABLE_MESHTEXT_CHAT=0"), False),
+            (dict(flags=["-DMESH_ENABLE_MESHTEXT_CHAT=0"], overlay="-DMESH_ENABLE_MESHTEXT_CHAT=1"), True),
+            (dict(flags=["-DMESH_ENABLE_MESHTEXT_CHAT=1"], overlay="-DMESH_ENABLE_MESHTEXT_CHAT=0"), False),
+            (dict(flags=["-DMESH_PRIMARY_ESPNOW=1"]), False),
+            (dict(flags=["-DMESH_PRIMARY_ESPNOW=1", "-UMESH_PRIMARY_ESPNOW"]), True),
+            (dict(flags=["-DMESH_PRIMARY_ESPNOW=1"], overlay="-DMESH_PRIMARY_ESPNOW=0"), True),
+            (dict(flags=["-DMESH_PRIMARY_ESPNOW=1", "-DMESH_ENABLE_MESHTEXT_CHAT=1"]), True),
+            (dict(flags=["-DMESH_ENABLE_MESHTEXT_CHAT=0"], overlay="-UMESH_ENABLE_MESHTEXT_CHAT"), True),
+            (dict(flags=["-DMESH_ENABLE_MESHTEXT_CHAT=0"], unflags="-D MESH_ENABLE_MESHTEXT_CHAT=0"), True),
+            (dict(flags=["-DMESH_ENABLE_MESHTEXT_CHAT=0"], config_unflags=["-DMESH_ENABLE_MESHTEXT_CHAT"]), True),
+            (dict(flags=["-DMESH_PRIMARY_ESPNOW=1", "-UMESH_PRIMARY_ESPNOW"],
+                  config_unflags=["-UMESH_PRIMARY_ESPNOW"]), False),
+            (dict(flags=["-DCHAT_POLICY=0", "-DMESH_ENABLE_MESHTEXT_CHAT=CHAT_POLICY"]), False),
+            (dict(flags=["-DMESH_ENABLE_MESHTEXT_CHAT=(0UL)"]), False),
+            (dict(flags=["-DCOMPANION_RADIO_FULL=0"]), False),
+        )
+        for options, required in cases:
+            with self.subTest(options=options):
+                result, manifest = self.builder_contract("Heltec_v2_companion_radio_full",
+                                                         "ESP32_PLATFORM", **options)
+                self.assertEqual(result.returncode, int(required), result.stdout + result.stderr)
+                self.assertEqual(manifest["verified"], not required)
+        for target, platform in (("t1000e_companion_radio_usb", "NRF52_PLATFORM"),
+                                 ("PicoW_companion_radio_full", "RP2040_PLATFORM"),
+                                 ("RAK_3x72_companion_radio_full", "STM32_PLATFORM")):
+            with self.subTest(target=target):
+                result, manifest = self.builder_contract(target, platform)
+                self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+                self.assertTrue(manifest["verified"])
+        result, manifest = self.builder_contract("Heltec_v2_companion_radio_full", "ESP32_PLATFORM",
+                                                 invalid_config=True)
+        self.assertEqual(result.returncode, 2, result.stdout + result.stderr)
+        self.assertIsNone(manifest)
+        self.assertIn("Cannot determine", result.stderr)
+
     def qualify(self, build, stem):
         output = stem.parent / (stem.name + ".capabilities.json")
         result = subprocess.run([sys.executable, str(ROOT / "scripts/check_firmware_capabilities.py"),

test/test_one_key_dm_compact_held.py

@@ -0,0 +1,316 @@
+#!/usr/bin/env python3
+"""Run the C3 compact-held-DM policy through the real Companion methods."""
+
+from pathlib import Path
+import subprocess
+import tempfile
+import unittest
+
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+def block(text, signature):
+    start = text.index(signature)
+    end = text.index("{", start) + 1
+    depth = 1
+    while depth:
+        depth += (text[end] == "{") - (text[end] == "}")
+        end += 1
+    return text[start:end]
+
+
+class CompactHeldDmTest(unittest.TestCase):
+    def test_independent_capacity_fifo_promotion_metadata_ack_and_retry_dedup(self):
+        source = (ROOT / "examples/companion_radio/MyMesh.cpp").read_text()
+        header = (ROOT / "examples/companion_radio/MyMesh.h").read_text()
+        methods = "\n".join(block(source, signature) for signature in (
+            "bool MyMesh::Frame::isChannelMsg() const",
+            "int MyMesh::getOfflineQueueCapacity() const",
+            "MyMesh::Frame& MyMesh::offlineQueueFrameAt(int logical_index)",
+            "bool MyMesh::addToOfflineQueue(const uint8_t frame[], int len)",
+            "int MyMesh::peekOfflineQueue(uint8_t frame[])",
+            "void MyMesh::popOfflineQueue()",
+            "int MyMesh::getFromOfflineQueue(uint8_t frame[])",
+            "void MyMesh::forgetVerifiedPendingSender(const uint8_t* pub_key)",
+            "void MyMesh::makeOneKeyDMId(uint8_t id[ONE_KEY_DM_ID_SIZE], uint32_t timestamp,",
+            "bool MyMesh::wasDeliveredOneKeyDM(const uint8_t* pub_key,",
+            "void MyMesh::rememberDeliveredOneKeyDM(const uint8_t* pub_key,",
+            "bool MyMesh::onAddressedTextPacket(mesh::Packet* packet, uint8_t src_hash,",
+            "void MyMesh::releaseHeldOneKeyDMs()",
+            "void MyMesh::onMessageRecv(const ContactInfo &from, mesh::Packet *pkt, uint32_t sender_timestamp,",
+        ))
+        harness = r'''
+#include <helpers/BorrowableFrameBuffer.h>
+#include <openssl/sha.h>
+#include <array>
+#include <cassert>
+#include <cstdint>
+#include <cstring>
+#include <string>
+#include <vector>
+#define ONE_KEY_DM_COMPACT_HELD 1
+#define ONE_KEY_DM_SHARED_OFFLINE_QUEUE 0
+#define OTA_SHARED_COMPANION_QUEUE 1
+#define OFFLINE_QUEUE_SIZE 224
+#define MAX_FRAME_SIZE 176
+#define MAX_PACKET_PAYLOAD 184
+#define MAX_TEXT_LEN 160
+#define PUB_KEY_SIZE 32
+#define TXT_TYPE_PLAIN 0
+#define ROUTE_TYPE_DIRECT 2
+#define ROUTE_TYPE_FLOOD 1
+#define MESH_ENABLE_ONE_KEY_DM 1
+#define MESH_ENABLE_LOST_REPLY 0
+#define RESP_CODE_CHANNEL_MSG_RECV 0x8c
+#define RESP_CODE_CHANNEL_MSG_RECV_V3 0x90
+#define RESP_CODE_CHANNEL_DATA_RECV 0x8d
+#define MESH_DEBUG_PRINTLN(...) ((void)0)
+namespace mesh {
+struct Packet {
+  uint8_t header = ROUTE_TYPE_FLOOD;
+  uint8_t path_len = 0;
+  int8_t _snr = 0;
+  bool isRouteFlood() const { return header == ROUTE_TYPE_FLOOD; }
+};
+struct Utils {
+  static bool authentic;
+  static int MACThenDecrypt(const uint8_t*, uint8_t* out,
+                            const uint8_t* in, size_t len) {
+    if (!authentic) return 0;
+    memcpy(out, in, len);
+    return static_cast<int>(len);
+  }
+  static void sha256(uint8_t* out, size_t size, const uint8_t* a, size_t a_size,
+                     const uint8_t* b, size_t b_size) {
+    std::vector<uint8_t> input(a, a + a_size);
+    input.insert(input.end(), b, b + b_size);
+    uint8_t digest[SHA256_DIGEST_LENGTH];
+    SHA256(input.data(), input.size(), digest);
+    memcpy(out, digest, size);
+  }
+};
+bool Utils::authentic = true;
+namespace ota { struct OtaContext { uint8_t bytes[11284]; }; }
+}
+struct ContactInfo { struct { uint8_t pub_key[PUB_KEY_SIZE] = {}; } id; };
+struct SelfIdentity {
+  void calcSharedSecret(uint8_t* secret, const uint8_t*) { memset(secret, 0, PUB_KEY_SIZE); }
+};
+struct Delivery {
+  uint32_t timestamp;
+  std::string text;
+  uint8_t route, path;
+  int8_t snr;
+};
+class MyMesh {
+public:
+  static constexpr uint8_t MAX_HELD_ONE_KEY_DMS = 15;
+  static constexpr uint8_t ONE_KEY_DM_ID_SIZE = 8;
+  struct Frame {
+    uint8_t len = 0, buf[MAX_FRAME_SIZE] = {};
+    bool isChannelMsg() const;
+  };
+  @HELD@;
+  @DELIVERED@;
+  static_assert(sizeof(HeldOneKeyDM) == 204, "15 slots must save 1740 bytes");
+  mesh::BorrowableFrameBuffer<Frame, 224, 128, mesh::ota::OtaContext> offline_queue;
+  int offline_queue_len = 0, offline_queue_head = 0;
+  HeldOneKeyDM held_dms[MAX_HELD_ONE_KEY_DMS] = {};
+  uint8_t held_dm_count = 0;
+  uint8_t verified_pending_keys[MAX_HELD_ONE_KEY_DMS][PUB_KEY_SIZE] = {};
+  uint8_t verified_pending_count = 1;
+  DeliveredOneKeyDM delivered_dms[MAX_HELD_ONE_KEY_DMS] = {};
+  uint8_t delivered_dm_count = 0, delivered_dm_next = 0;
+  std::vector<Delivery> deliveries;
+  std::vector<std::array<uint8_t, 4>> acks;
+  bool accepted = false;
+  unsigned writes = 0, active = 0;
+  ContactInfo contact;
+  SelfIdentity self_id;
+  MyMesh() {
+    for (unsigned i = 0; i < PUB_KEY_SIZE; ++i)
+      contact.id.pub_key[i] = verified_pending_keys[0][i] = 0x42 + i;
+  }
+  int getOfflineQueueCapacity() const;
+  Frame& offlineQueueFrameAt(int index);
+  bool addToOfflineQueue(const uint8_t frame[], int len);
+  int peekOfflineQueue(uint8_t frame[]);
+  void popOfflineQueue();
+  int getFromOfflineQueue(uint8_t frame[]);
+  bool onAddressedTextPacket(mesh::Packet*, uint8_t, const uint8_t*, size_t);
+  void releaseHeldOneKeyDMs();
+  void forgetVerifiedPendingSender(const uint8_t*);
+  static void makeOneKeyDMId(uint8_t[ONE_KEY_DM_ID_SIZE], uint32_t, const char*);
+  bool wasDeliveredOneKeyDM(const uint8_t*, const uint8_t[ONE_KEY_DM_ID_SIZE]) const;
+  void rememberDeliveredOneKeyDM(const uint8_t*, const uint8_t[ONE_KEY_DM_ID_SIZE]);
+  void onMessageRecv(const ContactInfo&, mesh::Packet*, uint32_t, const char*);
+  ContactInfo* lookupPersistentContactByPubKey(const uint8_t* key, size_t) {
+    return accepted && !memcmp(contact.id.pub_key, key, PUB_KEY_SIZE) ? &contact : nullptr;
+  }
+  void markConnectionActive(const ContactInfo&) { ++active; }
+  void scheduleContactWrite(const ContactInfo&) { ++writes; }
+  bool queueMessage(const ContactInfo&, uint8_t, mesh::Packet* packet,
+                     uint32_t timestamp, const uint8_t*, int, const char* text) {
+    uint8_t frame[MAX_FRAME_SIZE] = {1}; // private, never evicted as a channel
+    memcpy(frame + 1, &timestamp, 4);
+    memcpy(frame + 5, text, strlen(text));
+    if (!addToOfflineQueue(frame, 5 + strlen(text))) return false;
+    deliveries.push_back({timestamp, text, packet->header, packet->path_len, packet->_snr});
+    return true;
+  }
+  void sendAckTo(const ContactInfo&, const uint8_t* hash, uint8_t length) {
+    assert(length == 4);
+    std::array<uint8_t, 4> ack;
+    memcpy(ack.data(), hash, 4);
+    acks.push_back(ack);
+  }
+  void fill() {
+    const uint8_t frame = 1;
+    while (offline_queue_len < getOfflineQueueCapacity()) assert(addToOfflineQueue(&frame, 1));
+  }
+  void drain() { uint8_t frame[MAX_FRAME_SIZE]; while (getFromOfflineQueue(frame)) {} }
+};
+@METHODS@
+struct Message {
+  uint32_t timestamp;
+  std::string text;
+  mesh::Packet packet;
+  std::vector<uint8_t> plain;
+  std::array<uint8_t, 4> ack;
+  Message(unsigned id, size_t length = 0) : timestamp(id) {
+    text = length ? std::string(length, 'x') : "message-" + std::to_string(id);
+    if (length) { text[length - 2] = char(0xC3); text[length - 1] = char(0xA9); }
+    plain.resize(5 + text.size());
+    memcpy(plain.data(), &timestamp, 4);
+    plain[4] = id % 4; // real attempt bits must survive in the exact ACK
+    memcpy(plain.data() + 5, text.data(), text.size());
+    packet.header = id % 2 ? ROUTE_TYPE_DIRECT : ROUTE_TYPE_FLOOD;
+    packet.path_len = id % 2 ? 21 : 0x82;
+    packet._snr = id % 2 ? -91 : 51;
+    // Independently concatenate and hash exactly the original plaintext/key.
+    std::vector<uint8_t> input = plain;
+    for (unsigned i = 0; i < PUB_KEY_SIZE; ++i) input.push_back(0x42 + i);
+    uint8_t digest[SHA256_DIGEST_LENGTH];
+    SHA256(input.data(), input.size(), digest);
+    memcpy(ack.data(), digest, 4);
+  }
+  bool hold(MyMesh& receiver) {
+    return receiver.onAddressedTextPacket(&packet, 0x42, plain.data(), plain.size());
+  }
+};
+int main() {
+  {
+    MyMesh receiver;
+    std::vector<Message> messages;
+    for (unsigned i = 1; i <= 16; ++i) {
+      messages.emplace_back(i, i == 16 ? MAX_TEXT_LEN : 0);
+      assert(messages.back().hold(receiver));
+    }
+    assert(receiver.held_dm_count == 15 && receiver.held_dms[0].timestamp == 2);
+    assert(receiver.held_dms[14].text[MAX_TEXT_LEN] == 0);
+    auto original_ack = receiver.held_dms[14].ack_hash[0];
+    messages.back().plain[4] ^= 1; // same text/timestamp, different attempt
+    assert(messages.back().hold(receiver));
+    assert(receiver.held_dm_count == 15 && receiver.held_dms[14].ack_hash[0] == original_ack);
+    // Pending storage stays independent even with every normal slot occupied.
+    receiver.fill();
+    assert(receiver.offline_queue_len == 224 && receiver.held_dm_count == 15);
+    receiver.accepted = true;
+    receiver.releaseHeldOneKeyDMs();
+    assert(receiver.verified_pending_count == 0 && receiver.held_dm_count == 15);
+    assert(receiver.deliveries.empty() && receiver.acks.empty());
+    receiver.popOfflineQueue();
+    receiver.releaseHeldOneKeyDMs();
+    assert(receiver.held_dm_count == 14 && receiver.offline_queue_len == 224);
+    assert(receiver.deliveries.size() == 1 && receiver.deliveries[0].timestamp == 2);
+    receiver.drain();
+    receiver.releaseHeldOneKeyDMs();
+    assert(receiver.held_dm_count == 0 && receiver.deliveries.size() == 15);
+    assert(receiver.acks.size() == 15 && receiver.offline_queue_len == 14);
+    for (unsigned i = 0; i < 15; ++i) {
+      const Delivery& delivery = receiver.deliveries[i];
+      const Message& expected = messages[i + 1];
+      assert(delivery.timestamp == expected.timestamp && delivery.text == expected.text);
+      assert(delivery.route == expected.packet.header && delivery.snr == expected.packet._snr);
+      assert(delivery.path == (expected.packet.isRouteFlood() ? expected.packet.path_len : 0));
+      assert(receiver.acks[i] == expected.ack);
+    }
+  }
+  {
+    MyMesh receiver;
+    Message message(7, MAX_TEXT_LEN);
+    assert(message.hold(receiver));
+    receiver.fill();
+    receiver.accepted = true;
+    receiver.onMessageRecv(receiver.contact, &message.packet, message.timestamp, message.text.c_str());
+    assert(receiver.deliveries.empty() && receiver.held_dm_count == 1);
+    receiver.popOfflineQueue();
+    receiver.onMessageRecv(receiver.contact, &message.packet, message.timestamp, message.text.c_str());
+    assert(receiver.deliveries.size() == 1 && receiver.offline_queue_len == 224);
+    receiver.drain();
+    receiver.releaseHeldOneKeyDMs();
+    assert(receiver.held_dm_count == 0 && receiver.deliveries.size() == 1);
+    assert(receiver.acks.size() == 1 && receiver.acks[0] == message.ack);
+    receiver.onMessageRecv(receiver.contact, &message.packet, message.timestamp, message.text.c_str());
+    assert(receiver.deliveries.size() == 1); // already delivered retry stays deduplicated
+  }
+  {
+    MyMesh receiver;
+    Message message(33, MAX_TEXT_LEN);
+    mesh::Utils::authentic = false;
+    assert(!message.hold(receiver) && receiver.held_dm_count == 0);
+    mesh::Utils::authentic = true;
+    Message oversized(34, MAX_TEXT_LEN + 1);
+    assert(!oversized.hold(receiver) && receiver.held_dm_count == 0);
+    assert(!receiver.onAddressedTextPacket(&message.packet, 0x99,
+        message.plain.data(), message.plain.size()));
+    message.plain[4] = 4; // non-plain text must not enter the compact FIFO
+    assert(!message.hold(receiver) && receiver.held_dm_count == 0);
+    message.plain[4] = 1;
+    assert(message.hold(receiver));
+    receiver.fill();
+    for (unsigned i = 0; i < 96; ++i) receiver.popOfflineQueue();
+    auto* workspace = receiver.offline_queue.acquire(receiver.offline_queue_len,
+                                                     receiver.offline_queue_head);
+    assert(workspace && receiver.getOfflineQueueCapacity() == 128);
+    memset(workspace->bytes, 0xCC, sizeof workspace->bytes);
+    assert(!strcmp(receiver.held_dms[0].text, message.text.c_str()));
+    receiver.offline_queue.release(receiver.offline_queue_head);
+    assert(receiver.getOfflineQueueCapacity() == 224);
+    receiver.drain();
+    receiver.accepted = true;
+    receiver.releaseHeldOneKeyDMs();
+    assert(receiver.deliveries.size() == 1 && receiver.deliveries[0].text == message.text);
+    assert(receiver.acks.size() == 1 && receiver.acks[0] == message.ack);
+  }
+  {
+    MyMesh receiver;
+    Message first(50), second(50, 24);
+    assert(first.hold(receiver) && second.hold(receiver));
+    assert(receiver.held_dm_count == 2); // equal timestamps do not merge different text
+    assert(first.hold(receiver) && receiver.held_dm_count == 2);
+    receiver.accepted = true;
+    receiver.releaseHeldOneKeyDMs();
+    assert(receiver.deliveries.size() == 2 && receiver.acks.size() == 2);
+    assert(receiver.deliveries[0].text == first.text && receiver.deliveries[1].text == second.text);
+  }
+}
+'''
+        harness = harness.replace("@HELD@", block(header, "struct HeldOneKeyDM {"))
+        harness = harness.replace("@DELIVERED@", block(header, "struct DeliveredOneKeyDM {"))
+        harness = harness.replace("@METHODS@", methods)
+        with tempfile.TemporaryDirectory(prefix="meshcore-compact-dm-") as temp:
+            binary = Path(temp) / "compact"
+            built = subprocess.run([
+                "c++", "-std=c++17", "-Wall", "-Wextra", "-Werror",
+                "-fsanitize=address,undefined", "-g", "-I", str(ROOT / "src"),
+                "-x", "c++", "-", "-lcrypto", "-o", str(binary),
+            ], input=harness, text=True, capture_output=True)
+            self.assertEqual(built.returncode, 0, built.stderr)
+            result = subprocess.run([str(binary)], text=True, capture_output=True)
+            self.assertEqual(result.returncode, 0, result.stderr)
+
+
+if __name__ == "__main__":
+    unittest.main()