Reuse outbound block storage for host-compressed OTA data

src/helpers/ota/OtaManager.cpp

@@ -742,22 +742,22 @@ bool OtaManager::loadActiveServeBlock() {
   _serve_wire_deflated = false;
 
   // A host-backed MotaSource can provide a pre-compressed block without putting an encoder in firmware.
-  // Other sources may opt into a caller-supplied encoder. Encoded DATA remains
-  // pinned until every fragment has left. Streaming proofs need no output buffer.
-  if (job.wire_v2 && job.allow_deflate && (v->read_deflated || _deflate_encode) && !v->scratch) {
+  // Only device encoders need separate raw input and compressed output. Host
+  // output can use the existing outbound cache, which remains pinned until
+  // every DATA fragment has left. Streaming proofs need no output buffer.
+  if (job.wire_v2 && job.allow_deflate && !v->read_deflated && _deflate_encode && !v->scratch) {
     v->scratch = ensureScratch();
     v->scratch_sz = v->scratch ? OTA_MAX_BLOCK : 0;
   }
   _source_buffer_used_at = _now_ms;
   // A host-precompressed representation is enough to serve DATA: proofs use
   // authenticated leaves. Do not pull the same raw block over BLE/USB first.
-  if (job.wire_v2 && job.allow_deflate && v->read_deflated &&
-      v->scratch && v->scratch_sz >= blen) {
+  if (job.wire_v2 && job.allow_deflate && v->read_deflated) {
     uint16_t encoded_len = 0;
-    const bool encoded = v->read_deflated(v->deflate_ctx, job.block, v->scratch,
+    const bool encoded = v->read_deflated(v->deflate_ctx, job.block, _serve_block,
                                          (uint16_t)blen, &encoded_len);
     if (encoded && encoded_len > 0 && encoded_len < blen) {
-      _serve_wire_block = v->scratch;
+      _serve_wire_block = _serve_block;
       _serve_wire_len = encoded_len;
       _serve_wire_deflated = true;
     }

src/helpers/ota/OtaManager.h

@@ -744,15 +744,17 @@ private:
   uint8_t     _n_src_obj = 0;
   uint8_t     _src_manifest[OTA_SRC_MANIFEST_MAX];   // manifest-minus-leaves of the loaded source mota
   // Cold source leaves are sized to the selected image, within the unchanged
-  // capacity limit. Encoded DATA needs at most one logical block. These buffers
-  // remain pinned until queued egress drains, then return to the heap at idle.
+  // 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.
   uint8_t*    _src_leaves = nullptr;
   uint32_t    _src_leaves_capacity = 0;
   uint8_t*    _scratch = nullptr;
   uint32_t    _source_buffer_used_at = 0;
 
-  // Server-side response descriptors are tiny. The active logical block has its own buffer so proof generation
-  // or a simultaneous fetch cannot overwrite DATA retained behind radio-queue backpressure.
+  // Server-side response descriptors are tiny. Active DATA stays cached behind
+  // radio-queue backpressure. Proofs do not touch that cache; inbound decoding
+  // may reuse it but invalidates it before the next outbound fragment.
   struct ServeJob {
     OtaReplyRoute route;
     uint8_t mid[4] = {0};
@@ -790,7 +792,8 @@ private:
   uint8_t    _serve_block[OTA_MAX_BLOCK];
   uint16_t   _serve_block_len = 0;
   bool       _serve_block_loaded = false;
-  const uint8_t* _serve_wire_block = nullptr;       // raw _serve_block or DEFLATE bytes in ServeView::scratch
+  // Raw/host-encoded _serve_block, or device-encoder output in ServeView::scratch.
+  const uint8_t* _serve_wire_block = nullptr;
   uint16_t   _serve_wire_len = 0;
   bool       _serve_wire_deflated = false;
   uint8_t    _serve_wire_id[4] = {0};              // SHA-256:4 of this exact encoded representation

test/fixtures/ota_cold_buffers/test.cpp

@@ -91,23 +91,31 @@ static void proofs() {
 
 class Source : public MotaSource {
 public:
+  enum class HostEncoding { Off, Good, Refused, PartialFailure, Empty, NonShrinking, Oversized };
   std::vector<uint8_t> mota;
   MotaManifest manifest;
   bool fail_read = false;
-  Source(uint32_t blocks, uint8_t value) {
-    const uint32_t total = 8 + MOTA_MFL + blocks * 4 + blocks * OTA_MAX_BLOCK + 5;
+  bool fail_payload = false;
+  HostEncoding host_encoding = HostEncoding::Off;
+  uint16_t host_encoded_size = 8;
+  unsigned host_calls = 0, raw_reads = 0;
+  Source(uint32_t blocks, uint8_t value, uint16_t final_block_size = OTA_MAX_BLOCK) {
+    assert(blocks && final_block_size && final_block_size <= OTA_MAX_BLOCK);
+    const uint32_t payload_size = (blocks - 1) * OTA_MAX_BLOCK + final_block_size;
+    const uint32_t total = 8 + MOTA_MFL + blocks * 4 + payload_size + 5;
     mota.resize(total);
     memcpy(mota.data(), MOTA_MAGIC, 4); wr_u32le(mota.data() + 4, total);
     uint8_t* mf = mota.data() + 8;
     mf[0] = MOTA_FORMAT_VER; mf[1] = MFLAG_FULL; mf[2] = HASH_ALGO_SHA256;
     wr_u32le(mf + 3, 123); wr_u32le(mf + 7, 1);
-    wr_u32le(mf + 11, blocks * OTA_MAX_BLOCK); wr_u32le(mf + 15, blocks * OTA_MAX_BLOCK);
+    wr_u32le(mf + 11, payload_size); wr_u32le(mf + 15, payload_size);
     mf[19] = 11; mf[56] = CODEC_FULL;
     uint8_t* leaves = mf + MOTA_MFL;
     uint8_t* payload = leaves + blocks * 4;
     for (uint32_t i = 0; i < blocks; ++i) {
-      memset(payload + i * OTA_MAX_BLOCK, uint8_t(value + i), OTA_MAX_BLOCK);
-      merkle_leaf(leaves + i * 4, payload + i * OTA_MAX_BLOCK, OTA_MAX_BLOCK);
+      const uint16_t length = i == blocks - 1 ? final_block_size : OTA_MAX_BLOCK;
+      memset(payload + i * OTA_MAX_BLOCK, uint8_t(value + i), length);
+      merkle_leaf(leaves + i * 4, payload + i * OTA_MAX_BLOCK, length);
     }
     merkle_root(mf + 20, leaves, blocks);
     memcpy(mota.data() + total - 5, MOTA_TRAILER, 5);
@@ -121,13 +129,46 @@ public:
     d.total_size = mota.size(); d.leaves_off = 8 + MOTA_MFL;
     d.block_count = manifest.block_count;
     d.payload_off = manifest.payload - mota.data(); d.payload_size = manifest.payload_size;
+    d.source_caps = host_encoding == HostEncoding::Off ? 0 : MOTA_SOURCE_CAP_DEFLATE_BLOCK;
     return true;
   }
   bool read(uint8_t index, uint32_t off, uint8_t* out, uint32_t len) override {
     if (index || (uint64_t)off + len > mota.size()) return false;
+    const uint32_t payload_off = manifest.payload - mota.data();
+    if (off >= payload_off && off < payload_off + manifest.payload_size) {
+      ++raw_reads;
+      if (fail_payload) { if (len) out[0] = 0xEE; return false; }
+    }
     if (fail_read) { if (len) out[0] = 0xEE; return false; }
     memcpy(out, mota.data() + off, len); return true;
   }
+  bool read_deflated_block(uint8_t index, uint16_t block, uint8_t* out,
+                           uint16_t cap, uint16_t* len) override {
+    ++host_calls;
+    assert(index == 0 && block < manifest.block_count && len);
+    assert(cap == std::min<uint32_t>(OTA_MAX_BLOCK, manifest.payload_size - block * OTA_MAX_BLOCK));
+    // Deliberately opaque fixture representations: manager framing, hashes,
+    // proofs and cache ownership are real; this is not a DEFLATE codec test.
+    if (host_encoding == HostEncoding::Good) {
+      assert(host_encoded_size > 0 && host_encoded_size < cap);
+      memset(out, manifest.payload[block * OTA_MAX_BLOCK], host_encoded_size);
+      *len = host_encoded_size;
+      return true;
+    }
+    if (host_encoding == HostEncoding::PartialFailure) {
+      memset(out, 0xEE, 17); *len = 17; return false;
+    }
+    if (host_encoding == HostEncoding::Empty) { out[0] = 0xEE; *len = 0; return true; }
+    if (host_encoding == HostEncoding::NonShrinking) {
+      memset(out, 0xEE, cap); *len = cap; return true;
+    }
+    if (host_encoding == HostEncoding::Oversized) {
+      // Report a bad length without violating the callback's output capacity.
+      out[0] = 0xEE; *len = cap + 1; return true;
+    }
+    *len = 0;
+    return false;
+  }
 };
 struct Sent {
   bool accept_data = true, accept_proof = true, accept_manifest = true;
@@ -142,7 +183,7 @@ static bool send(void* opaque, const uint8_t* data, uint16_t len, bool) {
 }
 static bool encode(void*, const uint8_t* src, uint16_t, uint8_t* dst,
                    uint16_t cap, uint16_t* len) {
-  assert(cap == OTA_MAX_BLOCK); memset(dst, src[0], 8); *len = 8; return true;
+  assert(cap == OTA_MAX_BLOCK && src != dst); memset(dst, src[0], 8); *len = 8; return true;
 }
 static bool encode_long(void*, const uint8_t* src, uint16_t, uint8_t* dst,
                         uint16_t cap, uint16_t* len) {
@@ -159,6 +200,12 @@ static bool request(OtaManager& manager, const Source& source, uint16_t block =
   uint8_t wire[MAX_PACKET_PAYLOAD];
   return manager.on_message(wire, encode_req(wire, sizeof(wire), req));
 }
+static bool request_v2(OtaManager& manager, const Source& source, uint16_t block, bool allow_deflate) {
+  ReqMsg req{}; memcpy(req.manifest_id, source.manifest.merkle_root, 4);
+  req.block_idx = block; req.want_mask = ota_req_make_v2(0xFFF, allow_deflate, true);
+  uint8_t wire[MAX_PACKET_PAYLOAD];
+  return manager.on_message(wire, encode_req(wire, sizeof(wire), req));
+}
 static bool manifest_request(OtaManager& manager, const Source& source) {
   GetManifestMsg req{}; memcpy(req.manifest_id, source.manifest.merkle_root, 4); req.want_mask = 0xFFFF;
   uint8_t wire[MAX_PACKET_PAYLOAD];
@@ -193,6 +240,132 @@ static void verify(const Sent& sent, const Source& source, bool compressed) {
   assert(!memcmp(bytes.data(), source.manifest.payload, OTA_MAX_BLOCK));
 }
 
+static void verify_v2(const Sent& sent, const Source& source, uint16_t block,
+                      bool compressed, uint16_t encoded_size = 8) {
+  std::vector<uint8_t> bytes;
+  uint8_t identity[4]{};
+  bool proof_seen = false;
+  const uint16_t raw_size = std::min<uint32_t>(OTA_MAX_BLOCK, source.manifest.payload_size - block * OTA_MAX_BLOCK);
+  const uint16_t expected_size = compressed ? encoded_size : raw_size;
+  for (const auto& packet : sent.packets) {
+    DataMsg data{}; ProofMsg proof{};
+    if (decode_data(packet.data(), packet.size(), data)
+        && !memcmp(data.manifest_id, source.manifest.merkle_root, 4)) {
+      assert(data.block_idx == block);
+      uint8_t fragment; uint16_t length; bool deflated;
+      assert(ota_data_v2_unpack_extended(data.frag_off, raw_size, fragment, length, deflated));
+      assert(deflated == compressed && length == expected_size);
+      assert(fragment * OTA_FRAG_DATA_V2 == bytes.size());
+      if (bytes.empty()) memcpy(identity, data.data, 4);
+      else assert(!memcmp(identity, data.data, 4));
+      bytes.insert(bytes.end(), data.data + 4, data.data + data.data_len);
+    } else if (decode_proof(packet.data(), packet.size(), proof)
+               && !memcmp(proof.manifest_id, source.manifest.merkle_root, 4)) {
+      assert(proof.block_idx == block);
+      assert(merkle_verify(source.manifest.payload + block * OTA_MAX_BLOCK, raw_size, block,
+                           proof.proof, proof.n_proof, source.manifest.merkle_root,
+                           source.manifest.block_count));
+      proof_seen = true;
+    }
+  }
+  assert(proof_seen && bytes.size() == expected_size);
+  if (compressed) assert(bytes == std::vector<uint8_t>(expected_size, source.manifest.payload[block * OTA_MAX_BLOCK]));
+  else assert(!memcmp(bytes.data(), source.manifest.payload + block * OTA_MAX_BLOCK, raw_size));
+  uint8_t expected[4]; mh4(expected, bytes.data(), bytes.size());
+  assert(!memcmp(identity, expected, 4));
+}
+
+static void host_allocation_and_fallbacks() {
+  for (bool fail_raw : {false, true}) {
+    Source source(9, 0x42); source.host_encoding = Source::HostEncoding::Good;
+    source.host_encoded_size = 400; source.fail_payload = fail_raw;
+    Sent sent; OtaManager manager;
+    manager.begin(123, send, &sent); manager.set_transport_deflate_encoder(encode);
+    assert(manager.add_source(&source) && request(manager, source));
+    assert(allocated == 36);
+    fail_next = true;
+    manager.serviceEgress();
+    assert(fail_next && allocated == 36 && source.host_calls == 1 && source.raw_reads == 0);
+    drain(manager); verify_v2(sent, source, 0, true, 400);
+    assert(fail_next && allocated == 36 && source.host_calls == 1);
+    fail_next = false; manager.clear_sources(); assert(allocated == 0);
+  }
+  for (auto mode : {Source::HostEncoding::Refused, Source::HostEncoding::PartialFailure,
+                    Source::HostEncoding::Empty, Source::HostEncoding::NonShrinking,
+                    Source::HostEncoding::Oversized}) {
+    Source source(3, 0x51); source.host_encoding = mode;
+    Sent sent; OtaManager manager;
+    manager.begin(123, send, &sent); manager.set_transport_deflate_encoder(encode);
+    assert(manager.add_source(&source) && request(manager, source));
+    fail_next = true; drain(manager);
+    assert(fail_next && allocated == 12 && source.host_calls == 1 && source.raw_reads == 1);
+    verify_v2(sent, source, 0, false);
+    fail_next = false; manager.clear_sources(); assert(allocated == 0);
+  }
+  Source source(3, 0x61); source.host_encoding = Source::HostEncoding::PartialFailure;
+  source.fail_payload = true;
+  Sent sent; OtaManager manager; manager.begin(123, send, &sent);
+  assert(manager.add_source(&source) && request(manager, source));
+  fail_next = true; manager.serviceEgress();
+  assert(fail_next && sent.packets.empty() && !manager.pendingServeJobs());
+  assert(source.host_calls == 1 && source.raw_reads == 1 && allocated == 12);
+  fail_next = false; source.fail_payload = false; source.host_encoding = Source::HostEncoding::Good;
+  assert(request(manager, source)); drain(manager); verify_v2(sent, source, 0, true);
+  manager.clear_sources(); assert(allocated == 0);
+  puts("PASS: host compression needs leaves only; allocation denial and invalid-output fallback");
+}
+
+static void host_tail_and_maximum_capacity() {
+  for (bool maximum : {false, true}) {
+    const uint32_t blocks = maximum ? OTA_PROOFGEN_SCRATCH / 4 : 3;
+    Source source(blocks, 0x65, maximum ? OTA_MAX_BLOCK : 37);
+    source.host_encoding = Source::HostEncoding::Good;
+    source.fail_payload = true;
+    const uint16_t block = blocks - 1;
+    Sent sent; OtaManager manager; manager.begin(123, send, &sent);
+    assert(manager.add_source(&source) && request(manager, source, block));
+    assert(allocated == blocks * 4);
+    if (maximum) assert(allocated == OTA_PROOFGEN_SCRATCH);
+    fail_next = true; drain(manager); verify_v2(sent, source, block, true);
+    assert(fail_next && allocated == blocks * 4 && source.host_calls == 1 && source.raw_reads == 0);
+    fail_next = false; manager.clear_sources(); assert(allocated == 0);
+  }
+  puts("PASS: host short tail and maximum leaf capacity retain original source bounds");
+}
+
+static void host_backpressure_and_raw_switch() {
+  Source source(5, 0x43); source.host_encoding = Source::HostEncoding::Good;
+  source.host_encoded_size = 400;
+  Sent sent; OtaManager manager; manager.begin(123, send, &sent);
+  assert(manager.add_source(&source) && request(manager, source));
+  sent.accept_data = false; fail_next = true;
+  manager.serviceEgress(); manager.set_clock(100000); manager.serviceEgress();
+  assert(fail_next && allocated == 20 && source.host_calls == 1 && sent.packets.empty());
+  sent.accept_data = true; manager.serviceEgress(); // Admit exactly one of three fragments.
+  assert(sent.packets.size() == 1 && manager.pendingServeJobs());
+  source.host_encoding = Source::HostEncoding::PartialFailure; // A pinned representation must not be re-read.
+  sent.accept_data = false; manager.set_clock(200000); manager.loop();
+  assert(source.host_calls == 1 && allocated == 20);
+  sent.accept_data = true; sent.accept_proof = false;
+  for (unsigned turn = 0; turn < 10; ++turn) {
+    manager.set_clock(300000 + turn * 10000); manager.serviceEgress();
+  }
+  assert(manager.pendingServeJobs() && fail_next && source.host_calls == 1 && allocated == 20);
+  sent.accept_proof = true; drain(manager, 500000); verify_v2(sent, source, 0, true, 400);
+  assert(fail_next); fail_next = false;
+  sent.packets.clear();
+  assert(request_v2(manager, source, 1, false)); drain(manager, 600000);
+  assert(source.host_calls == 1 && source.raw_reads == 1 && allocated == 20);
+  verify_v2(sent, source, 1, false);
+  source.host_encoding = Source::HostEncoding::Good; source.host_encoded_size = 8;
+  sent.packets.clear();
+  assert(request(manager, source, 2)); drain(manager, 800000);
+  assert(source.host_calls == 2 && source.raw_reads == 1 && allocated == 20);
+  verify_v2(sent, source, 2, true);
+  manager.clear_sources(); assert(allocated == 0);
+  puts("PASS: host bytes stay pinned through DATA/PROOF backpressure and raw/compressed switches");
+}
+
 static void lifetimes() {
   Source source(9, 0x42);
   Sent sent; OtaManager manager;
@@ -261,14 +434,7 @@ static void faults_and_capacity() {
   fail_next = true; assert(request(manager, small, 0, false)); drain(manager);
   assert(fail_next && allocated == 0); fail_next = false; // raw proof serving allocates nothing
 }
-static void concurrent_receive() {
-  Source outbound(5, 0x40), incoming(3, 0x70);
-  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(&outbound) && request(manager, outbound));
-  sent.accept_data = false; manager.serviceEgress();
-  assert(allocated == 20 + OTA_MAX_BLOCK);
+static void begin_receive(OtaManager& manager, const Source& incoming) {
   assert(manager.pull(incoming.manifest.merkle_root, 123) == OtaManager::PULL_STARTED);
   for (unsigned f = 0; f < 2; ++f) {
     ManifestMsg mf{}; memcpy(mf.manifest_id, incoming.manifest.merkle_root, 4);
@@ -277,6 +443,23 @@ static void concurrent_receive() {
     uint8_t wire[MAX_PACKET_PAYLOAD]; manager.on_message(wire, encode_manifest(wire, sizeof(wire), mf));
   }
   assert(manager.fetchState() == OtaManager::FETCHING);
+}
+static void concurrent_receive(bool host = false) {
+  Source outbound(5, 0x40), incoming(3, 0x70);
+  if (host) {
+    outbound.host_encoding = Source::HostEncoding::Good;
+    outbound.host_encoded_size = 400;
+  }
+  const size_t expected_allocation = 20 + (host ? 0 : OTA_MAX_BLOCK);
+  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(&outbound) && request(manager, outbound));
+  sent.accept_data = host; manager.serviceEgress();
+  if (host) assert(sent.packets.size() == 1 && manager.pendingServeJobs());
+  sent.accept_data = false;
+  assert(allocated == expected_allocation);
+  begin_receive(manager, incoming);
   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);
@@ -287,30 +470,69 @@ static void concurrent_receive() {
     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 == 20 + OTA_MAX_BLOCK && manager.pendingServeJobs());
+    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); verify(sent, outbound, true);
+  sent.accept_data = true; drain(manager, 200000);
+  if (host) verify_v2(sent, outbound, 0, true, 400);
+  else verify(sent, outbound, true);
+  if (host) assert(outbound.host_calls > 1 && outbound.raw_reads == 0);
   manager.clear_sources(); assert(allocated == 0);
+  if (host) puts("PASS: host DATA reloads after successful interleaved inbound decoding");
 }
-static void interleaved_manifest() {
+static bool partial_decode_failure(void*, const uint8_t*, uint16_t, uint8_t* out,
+                                    uint16_t cap, uint16_t* produced) {
+  assert(cap == OTA_MAX_BLOCK); memset(out, 0xEE, cap / 2); *produced = cap / 2;
+  return false;
+}
+static void host_receive_failure() {
+  Source outbound(5, 0x44), incoming(3, 0x71);
+  outbound.host_encoding = Source::HostEncoding::Good; outbound.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_decoder(partial_decode_failure);
+  assert(manager.add_source(&outbound) && request(manager, outbound));
+  manager.serviceEgress(); // Already-admitted bytes must keep the same representation after reload.
+  assert(sent.packets.size() == 1 && manager.pendingServeJobs());
+  sent.accept_data = false;
+  assert(outbound.host_calls == 1 && allocated == 20);
+  begin_receive(manager, incoming);
+  uint8_t representation[12]; memset(representation + 4, 0x71, 8);
+  mh4(representation, representation + 4, 8);
+  DataMsg data{}; memcpy(data.manifest_id, incoming.manifest.merkle_root, 4);
+  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)));
+  assert(manager.fetchState() == OtaManager::FETCHING);
+  sent.accept_data = true; drain(manager, 100000); verify_v2(sent, outbound, 0, true, 400);
+  assert(outbound.host_calls == 2 && outbound.raw_reads == 0 && allocated == 20);
+  manager.clear_sources(); assert(allocated == 0);
+  puts("PASS: partial failed inbound decoder cannot leak into pinned host DATA");
+}
+static void interleaved_manifest(bool host = false) {
   Source data_source(9, 0x35), manifest_source(17, 0x75);
+  if (host) {
+    data_source.host_encoding = Source::HostEncoding::Good;
+    data_source.host_encoded_size = 400;
+  }
+  const size_t output_allocation = host ? 0 : OTA_MAX_BLOCK;
   Sent sent; OtaManager manager;
   manager.begin(123, send, &sent); manager.set_transport_deflate_encoder(encode_long);
   assert(manager.add_source(&data_source) && manager.add_source(&manifest_source));
   assert(request(manager, data_source));
   manager.serviceEgress(); // first compressed fragment is admitted; later DATA stays pinned
-  assert(manager.pendingServeJobs() && allocated == 36 + OTA_MAX_BLOCK);
+  assert(manager.pendingServeJobs() && allocated == 36 + output_allocation);
   peak = allocated;
   assert(manifest_request(manager, manifest_source));
-  assert(allocated == 68 + OTA_MAX_BLOCK && peak == allocated); // old leaves were freed before growth
+  assert(allocated == 68 + output_allocation && peak == allocated); // old leaves were freed before growth
   sent.accept_data = false;
   manager.set_clock(100000); manager.serviceEgress();
   manager.set_clock(200000); manager.serviceEgress();
   assert(!manager.pendingManifestJobs() && manager.pendingServeJobs());
-  manager.loop(); assert(allocated == 68 + OTA_MAX_BLOCK);
+  manager.loop(); assert(allocated == 68 + output_allocation);
   sent.accept_data = true; drain(manager, 300000);
   std::vector<uint8_t> encoded;
   uint8_t identity[4]{}; bool proof_seen = false;
@@ -336,7 +558,7 @@ static void interleaved_manifest() {
   manager.clearPendingEgress(); assert(allocated == 0 && manager.servedCount() == 2);
   sent.packets.clear(); assert(request(manager, data_source)); manager.serviceEgress();
   fail_next = true; assert(!manifest_request(manager, manifest_source));
-  assert(allocated == OTA_MAX_BLOCK); // failed growth released old leaves, but pinned DATA survives
+  assert(allocated == output_allocation); // failed growth released old leaves, but pinned DATA survives
   drain(manager, 900000);
   encoded.clear();
   for (const auto& packet : sent.packets) {
@@ -348,6 +570,38 @@ static void interleaved_manifest() {
   manager.clearPendingEgress(); assert(allocated == 0);
   manager.set_clock(0xFFFFFF00u); assert(request(manager, data_source)); drain(manager, 0xFFFFFF00u);
   manager.set_clock(0xFFFFFF00u + 100000u); manager.loop(); assert(allocated == 0);
+  if (host) puts("PASS: host DATA survives interleaved source growth, failure and rollover");
+}
+
+struct SelfEncoder {
+  uint8_t* expected_output;
+  unsigned calls = 0;
+};
+static bool self_encode(void* context, const uint8_t* input, uint16_t length,
+                         uint8_t* output, uint16_t capacity, uint16_t* produced) {
+  auto& encoder = *static_cast<SelfEncoder*>(context);
+  assert(output == encoder.expected_output && input != output);
+  ++encoder.calls;
+  return encode(nullptr, input, length, output, capacity, produced);
+}
+static bool self_read(void* context, uint32_t offset, uint8_t* output, uint32_t length) {
+  auto& source = *static_cast<Source*>(context);
+  return source.read(0, uint32_t(source.manifest.payload - source.mota.data()) + offset, output, length);
+}
+static void caller_owned_scratch() {
+  Source source(3, 0x63);
+  std::array<uint8_t, OTA_MAX_BLOCK> scratch{};
+  SelfEncoder encoder{scratch.data()};
+  Sent sent; OtaManager manager; manager.begin(123, send, &sent);
+  manager.set_transport_deflate_encoder(self_encode, &encoder);
+  assert(manager.serve_self(source.manifest.manifest_start, MOTA_MFL, source.manifest.leaves,
+                            source.manifest.block_count, scratch.data(), scratch.size(), self_read, &source));
+  fail_next = true; assert(request(manager, source)); drain(manager);
+  assert(fail_next && allocated == 0 && encoder.calls == 1);
+  verify_v2(sent, source, 0, true);
+  fail_next = false; manager.clearPendingEgress();
+  assert(scratch[0] == 0x63 && allocated == 0);
+  puts("PASS: caller-owned self-source encoder output remains separate and unallocated");
 }
 int main() {
   proofs(); assert(allocated == 0);
@@ -355,5 +609,12 @@ int main() {
   faults_and_capacity(); assert(allocated == 0);
   concurrent_receive(); assert(allocated == 0);
   interleaved_manifest(); assert(allocated == 0);
+  host_allocation_and_fallbacks(); assert(allocated == 0);
+  host_tail_and_maximum_capacity(); assert(allocated == 0);
+  host_backpressure_and_raw_switch(); assert(allocated == 0);
+  concurrent_receive(true); assert(allocated == 0);
+  host_receive_failure(); assert(allocated == 0);
+  interleaved_manifest(true); assert(allocated == 0);
+  caller_owned_scratch(); assert(allocated == 0);
   puts("PASS: streaming proofs and OTA cold-buffer lifetimes");
 }

test/test_ota_cold_buffers.py

@@ -1,5 +1,10 @@
 #!/usr/bin/env python3
-"""Verify streaming proofs and real OTA cold-buffer lifetimes without PlatformIO."""
+"""Verify real OTA cold allocations, host DATA cache ownership and streaming proofs.
+
+The fixture uses controlled host representations, encoder/decoder callbacks and
+radio admission. It validates manager bytes/hashes/proofs under sanitizers, not
+the DEFLATE codec or physical USB/BLE/radio transport.
+"""
 
 from pathlib import Path
 import os
@@ -12,6 +17,15 @@ from test_serial_wifi_sessions import SANITIZER_FLAGS
 
 ROOT = Path(__file__).resolve().parents[1]
 FIXTURE = ROOT / "test/fixtures/ota_cold_buffers/test.cpp"
+HOST_MARKERS = (
+    "host compression needs leaves only; allocation denial and invalid-output fallback",
+    "host short tail and maximum leaf capacity retain original source bounds",
+    "host bytes stay pinned through DATA/PROOF backpressure and raw/compressed switches",
+    "host DATA reloads after successful interleaved inbound decoding",
+    "partial failed inbound decoder cannot leak into pinned host DATA",
+    "host DATA survives interleaved source growth, failure and rollover",
+    "caller-owned self-source encoder output remains separate and unallocated",
+)
 
 
 class OtaColdBufferTests(unittest.TestCase):
@@ -36,7 +50,10 @@ class OtaColdBufferTests(unittest.TestCase):
                 self.assertEqual(built.returncode, 0, built.stdout + built.stderr)
                 checked = subprocess.run([str(binary)], capture_output=True, text=True, timeout=60)
                 self.assertEqual(checked.returncode, 0, checked.stdout + checked.stderr)
+                for marker in HOST_MARKERS:
+                    self.assertEqual(checked.stdout.count("PASS: " + marker), 1, checked.stdout)
                 self.assertIn("PASS: streaming proofs and OTA cold-buffer lifetimes", checked.stdout)
+                print(f"\n[OTA cold buffers: {platform or 'portable'}]\n{checked.stdout}", flush=True)
 
 
 if __name__ == "__main__":