Preserve folder OTA staging when resume reads fail

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

@@ -243,6 +243,7 @@ jobs:
         run: |
           python3 -B test/test_mota_serial_deadline.py -v
           python3 -B test/test_mota_folder_deadline.py -v
+          python3 -B test/test_ota_folder_resume.py -v
           python3 -B test/test_ota_direct_manager.py -v
           python3 -B test/test_mota_kiss_direct.py -v
           python3 -B test/test_ota_kiss_direct.py -v

src/helpers/ota/FolderMotaStore.cpp

@@ -114,8 +114,12 @@ bool FolderMotaStore::finalize() {
 }
 
 bool FolderMotaStore::reopen() {
+  _reopen_io_error = false;
   uint8_t pl[5];
-  if (!txn(MS_OP_STAT, _mid, 4, pl, 5)) return false;
+  if (!txn(MS_OP_STAT, _mid, 4, pl, 5)) {
+    _reopen_io_error = true;
+    return false;
+  }
   if (pl[0] == 0) return false;                     // host has no partial/complete file -> start fresh
   uint32_t total = rd_u32le(pl + 1);
   if (total < 13) return false;                     // implausible container

src/helpers/ota/FolderMotaStore.h

@@ -40,6 +40,7 @@ public:
   void clear() override { _total = 0; }
   bool finalize() override;
   bool reopen() override;                                       // OP_STAT: adopt an existing file for resume
+  bool reopenHadIoError() const override { return _reopen_io_error; }
   bool canReconnect() const override { return true; }
 
   // A host-backed store is fully random-access, so it needs no pinned-meta RAM page and no layout planning.
@@ -58,6 +59,7 @@ private:
   MotaStreamWritePolicy _write_policy;
   uint8_t  _mid[4] = {0, 0, 0, 0};
   uint32_t _total = 0;
+  bool _reopen_io_error = false;
 };
 
 } // namespace ota

src/helpers/ota/OtaCli.cpp

@@ -740,7 +740,9 @@ static __attribute__((noinline)) bool handle_control_command(
       return true;
     }
     snprintf(reply, 160, "OK %s mid=%s -> %s (primary traffic)",
-             result == OtaManager::PULL_RESUMED ? "resuming" : "pulling", midhx, dname);
+             c.manager.fetchState() == OtaManager::PAUSED
+                 ? "paused (folder I/O error; reconnect to resume)"
+                 : result == OtaManager::PULL_RESUMED ? "resuming" : "pulling", midhx, dname);
 
   // ---- discard the current session (e.g. a stalled old fetch) to free the slot ----
   } else if (is_cmd(a, "drop|cancel|stop", &rest)) {

src/helpers/ota/OtaManager.cpp

@@ -1572,10 +1572,21 @@ OtaManager::PullResult OtaManager::startFetch(const uint8_t* mid, uint32_t targe
   // A validate pull is a FRESH seed capture, not a resume: the store already holds the seed's payload (not a
   // real partial), so never adopt it via resumeStaged - always re-begin and run the manifest->leaves->diff.
   const bool automatic = !_have_desired_mid && _desired_target == 0 && !_archive_fetch;
-  if (!validate && resumeStagedFor(mid, automatic)) return PULL_RESUMED; // adopt a partial container left in the store
   memcpy(_fid, mid, 4);
   _have = 0; _fbc = 0; _ftotal = 0; _fflags = 0;
   _observed_path_transmissions = 0;              // learn the actual source/relay path from this fetch's replies
+  if (!validate) {
+    if (resumeStagedFor(mid, automatic)) return PULL_RESUMED;
+    if (_fetch_error == FETCH_ERROR_STORAGE) {
+      // A failed host read cannot authorize BEGIN to erase a valid partial.
+      // IDLE records that adoption itself must be retried on reconnect; no
+      // fresh-container initialization has been authorized yet.
+      _fstate = PAUSED;
+      _paused_from = IDLE;
+      clearReassembly();
+      return PULL_STARTED;
+    }
+  }
   _fstate = WANT_MANIFEST;
   _mf_total = 0; _mf_mask = 0; _mf_len = 0; _mf_retries = 0; _loop_last_mfmask = 0;   // fresh manifest reassembly
   memset(_mf_buf, 0, sizeof(_mf_buf));
@@ -1616,7 +1627,14 @@ bool OtaManager::pauseFetchForDisconnect() {
 
 bool OtaManager::resumeFetchAfterReconnect() {
   if (_fstate != PAUSED || !_fetch) return false;
-  if (_paused_from == WANT_MANIFEST || _paused_from == WANT_LEAVES) {
+  if (_paused_from == IDLE) {
+    // The first resume probe lost its link before it could establish whether
+    // staging was usable. Retry it before allowing a fresh initialization.
+    uint8_t mid[4]; memcpy(mid, _fid, sizeof(mid));
+    _fstate = IDLE;
+    PullResult result = startFetch(mid, _fexpected_target, _validate);
+    if ((result != PULL_STARTED && result != PULL_RESUMED) || _fstate == PAUSED) return false;
+  } else if (_paused_from == WANT_MANIFEST || _paused_from == WANT_LEAVES) {
     // begin/header/manifest/trailer may only have partly reached the host.
     // Re-request the manifest and run the full initialization transaction.
     // Leaf-diff also restarts initialization: the reconnected host may have
@@ -1859,15 +1877,23 @@ bool OtaManager::resumeStagedFor(const uint8_t* want_mid, bool automatic_resume)
   const uint32_t expected_target = automatic_resume
       ? (_migration_target ? _migration_target : _target)
       : _fexpected_target;
-  if (!_fetch->reopenFor(want_mid, expected_target)) return false;
+  auto readFailure = [&]() {
+    if (_fetch->canReconnect()) _fetch_error = FETCH_ERROR_STORAGE;
+    return false;
+  };
+  if (!_fetch->reopenFor(want_mid, expected_target)) {
+    return _fetch->reopenHadIoError() ? readFailure() : false;
+  }
   uint32_t total = _fetch->staged_size();
   uint8_t hdr[8];
-  if (total < 8u + MOTA_MFL + 5u || !_fetch->read(0, hdr, sizeof(hdr))
-      || memcmp(hdr, MOTA_MAGIC, 4) != 0 || rd_u32le(hdr + 4) != total) return false;
+  if (total < 8u + MOTA_MFL + 5u) return false;
+  if (!_fetch->read(0, hdr, sizeof(hdr))) return readFailure();
+  if (memcmp(hdr, MOTA_MAGIC, 4) != 0 || rd_u32le(hdr + 4) != total) return false;
   // read + parse the stored manifest (everything before leaves[]) to recompute the geometry
   uint8_t mbuf[MOTA_MFL];
   MotaManifest m;
-  if (!_fetch->read(8, mbuf, sizeof(mbuf)) || !mota_parse_manifest(mbuf, sizeof(mbuf), m)) return false;
+  if (!_fetch->read(8, mbuf, sizeof(mbuf))) return readFailure();
+  if (!mota_parse_manifest(mbuf, sizeof(mbuf), m)) return false;
   if (want_mid && memcmp(m.merkle_root, want_mid, 4) != 0) return false;   // a different fw is staged
   // A boot-time resume is an automatic fetch decision and therefore belongs
   // only to this node. An explicit pull retains target=0 as an intentional
@@ -1896,8 +1922,8 @@ bool OtaManager::resumeStagedFor(const uint8_t* want_mid, bool automatic_resume)
   uint32_t payload_off = leaves_off + bc * 4;
   if ((uint64_t)payload_off + m.payload_size + 5 != total) return false;   // geometry must match the header
   uint8_t trailer[5];
-  if (!_fetch->read(total - sizeof(trailer), trailer, sizeof(trailer))
-      || memcmp(trailer, MOTA_TRAILER, sizeof(trailer)) != 0) return false;
+  if (!_fetch->read(total - sizeof(trailer), trailer, sizeof(trailer))) return readFailure();
+  if (memcmp(trailer, MOTA_TRAILER, sizeof(trailer)) != 0) return false;
 
   _wire_v2_session = _wire_v2_enabled;
   _wire_v2_confirmed = false;

src/helpers/ota/OtaStore.h

@@ -70,6 +70,12 @@ public:
     (void)want_mid; (void)expected_target; return reopen();
   }
 
+  // A failed reopen normally means no usable container, so a new pull may
+  // begin by erasing it. Reconnectable stores must distinguish an I/O failure:
+  // an unavailable host has not established that its staged bytes are invalid.
+  // Query only after reopen()/reopenFor() returned false.
+  virtual bool reopenHadIoError() const { return false; }
+
   // External host stores can disappear temporarily while retaining their file.
   // Preserve the selected fetch on an I/O failure until that link reconnects.
   virtual bool canReconnect() const { return false; }

test/fixtures/ota_folder_resume.cpp

@@ -0,0 +1,338 @@
+#include <helpers/ota/FolderMotaStore.h>
+#include <helpers/ota/OtaByteIO.h>
+#include <helpers/ota/OtaManager.h>
+#include <helpers/ota/MotaSeederProto.h>
+#include <Utils.h>
+#include "test_ota/mota_vectors.h"
+#include <algorithm>
+#include <cassert>
+#include <cstdio>
+#include <vector>
+
+using namespace mesh::ota;
+
+namespace mesh {
+namespace ota {
+struct OtaContext {
+  OtaManager manager;
+  OtaStoreRam<32768> fetch_store;
+  FolderMotaStore* folder_dest = nullptr;
+  const char* folder_dest_info = "test host";
+  bool apply_pending = false, fetch_to_folder = false;
+};
+
+// @CLI_HELPERS@
+
+static bool pull_command(const char* a, char* reply, OtaContext& c) {
+  const char* rest = nullptr;
+  if (false) {
+// @CLI_PULL@
+  } else {
+    return false;
+  }
+  return true;
+}
+}  // namespace ota
+}  // namespace mesh
+
+struct HostStream : Stream {
+  enum Failure { None, Timeout, Truncated, BadChecksum, Error, ShortWrite, WrongOp };
+  std::vector<uint8_t> bytes{SIM_MOTA_1K, SIM_MOTA_1K + SIM_MOTA_1K_LEN};
+  std::vector<uint8_t> response;
+  std::vector<uint8_t> operations;
+  size_t position = 0;
+  bool present = true;
+  uint8_t fail_op = MS_OP_SREAD;
+  uint32_t fail_offset = 0;
+  Failure failure = None;
+  unsigned failures_left = 0, begins = 0, writes = 0;
+
+  int available() override { return response.size() - position; }
+  int read() override { return position < response.size() ? response[position++] : -1; }
+  int peek() override { return -1; }
+  void flush() override { assert(false); }
+  size_t write(uint8_t) override { assert(false); return 0; }
+  size_t write(const uint8_t* request, size_t length) override {
+    assert(length >= 4 && request[0] == MOTA_SEEDER_REQ_MAGIC0
+           && request[1] == MOTA_SEEDER_REQ_MAGIC1);
+    uint8_t checksum = 0;
+    for (size_t i = 2; i < length - 1; ++i) checksum ^= request[i];
+    assert(checksum == request[length - 1]);
+    const uint8_t op = request[2];
+    operations.push_back(op);
+    const bool fail = failures_left && op == fail_op
+        && (op != MS_OP_SREAD || rd_u32le(request + 7) == fail_offset);
+    if (fail) --failures_left;
+    response.clear();
+    position = 0;
+    if (fail && failure == ShortWrite) return length - 1;
+    if (fail && failure == Timeout) return length;
+    response = {MOTA_SEEDER_RSP_MAGIC0, MOTA_SEEDER_RSP_MAGIC1,
+                uint8_t(fail && failure == WrongOp ? op + 1 : op),
+                uint8_t(fail && failure == Error ? MS_STATUS_ERR : MS_STATUS_OK)};
+    if (!(fail && failure == Error)) {
+      if (op == MS_OP_STAT) {
+        response.push_back(present ? 1 : 0);
+        uint8_t size[4]; wr_u32le(size, bytes.size());
+        response.insert(response.end(), size, size + sizeof(size));
+      } else if (op == MS_OP_SREAD) {
+        const uint32_t offset = rd_u32le(request + 7);
+        const uint16_t count = request[11] | (uint16_t(request[12]) << 8);
+        assert(present && offset <= bytes.size() && count <= bytes.size() - offset);
+        response.insert(response.end(), bytes.begin() + offset, bytes.begin() + offset + count);
+      } else if (op == MS_OP_BEGIN) {
+        ++begins;
+        present = true;
+        bytes.assign(rd_u32le(request + 7), 0xFF);
+      } else if (op == MS_OP_WRITE) {
+        ++writes;
+        const uint32_t offset = rd_u32le(request + 7);
+        const uint16_t count = request[11] | (uint16_t(request[12]) << 8);
+        assert(present && offset <= bytes.size() && count <= bytes.size() - offset);
+        std::copy(request + 13, request + 13 + count, bytes.begin() + offset);
+      } else {
+        assert(op == MS_OP_FIN);
+      }
+    }
+    checksum = 0;
+    for (uint8_t value : response) checksum ^= value;
+    response.push_back(fail && failure == BadChecksum ? uint8_t(checksum ^ 1) : checksum);
+    if (fail && failure == Truncated) response.resize(4);
+    return length;
+  }
+};
+
+struct Flight {
+  unsigned manifests = 0;
+  std::vector<uint32_t> requested_blocks;
+  static bool send(void* context, const uint8_t* bytes, uint16_t length, bool) {
+    auto& flight = *static_cast<Flight*>(context);
+    GetManifestMsg manifest{};
+    if (decode_get_manifest(bytes, length, manifest)) ++flight.manifests;
+    ReqWindowMsg request{};
+    if (decode_req_window(bytes, length, request)) {
+      for (uint8_t i = 0; i < request.n_items; ++i)
+        flight.requested_blocks.push_back(request.items[i].block_idx);
+    }
+    return true;
+  }
+};
+
+struct Fixture {
+  HostStream host;
+  FolderMotaStore store{host, MotaStreamWritePolicy::NoFlush, 4};
+  OtaContext context;
+  Flight flight;
+  MotaManifest manifest;
+
+  Fixture() {
+    assert(mota_parse(SIM_MOTA_1K, SIM_MOTA_1K_LEN, manifest));
+    context.folder_dest = &store;
+    context.manager.begin(manifest.target_id, Flight::send, &flight);
+    context.manager.set_wire_v2_enabled(false);
+    uint8_t row[OTA_HAVE_ROW_BYTES] = {};
+    memcpy(row, manifest.merkle_root, 4);
+    wr_u32le(row + 4, manifest.target_id);
+    wr_u32le(row + 8, manifest.fw_version);
+    row[12] = manifest.codec_id;
+    row[13] = manifest.flags;
+    HaveMsg have{};
+    have.seeder_id[0] = 1;
+    have.frag_total = have.n_rows = 1;
+    have.rows = row;
+    uint8_t wire[256];
+    context.manager.on_message(wire, encode_have(wire, sizeof(wire), have));
+    assert(context.manager.catalogCount() == 1);
+  }
+
+  void pull(bool validate = false) {
+    char reply[160];
+    assert(pull_command(validate ? "pull 1 folder validate" : "pull 1 folder", reply, context));
+    assert(strncmp(reply, "OK ", 3) == 0);
+    if (context.manager.fetchState() == OtaManager::PAUSED) {
+      assert(strstr(reply, "OK paused (folder I/O error; reconnect to resume)") == reply);
+      assert(context.fetch_to_folder);
+    }
+  }
+
+  void deliver_manifest() {
+    for (uint32_t offset = 0; offset < MOTA_MFL; offset += OTA_MF_FRAG) {
+      ManifestMsg message{};
+      memcpy(message.manifest_id, manifest.merkle_root, 4);
+      message.frag_idx = offset / OTA_MF_FRAG;
+      message.frag_total = (MOTA_MFL + OTA_MF_FRAG - 1) / OTA_MF_FRAG;
+      message.bytes = manifest.manifest_start + offset;
+      message.len = std::min(uint32_t(OTA_MF_FRAG), MOTA_MFL - offset);
+      uint8_t wire[256];
+      context.manager.on_message(wire, encode_manifest(wire, sizeof(wire), message));
+    }
+  }
+
+  void finish_verification() {
+    for (unsigned guard = 0; guard < 100 && context.manager.fetchState() == OtaManager::VERIFYING_STAGED; ++guard)
+      context.manager.loop();
+    assert(context.manager.fetchState() != OtaManager::VERIFYING_STAGED);
+  }
+
+  void deliver_block(uint32_t block) {
+    const uint32_t block_size = manifest.block_size();
+    const uint32_t length = std::min(block_size, manifest.payload_size - block * block_size);
+    uint8_t wire[256];
+    for (uint32_t offset = 0; offset < length; offset += OTA_FRAG_DATA) {
+      DataMsg message{};
+      memcpy(message.manifest_id, manifest.merkle_root, 4);
+      message.block_idx = block;
+      message.frag_off = offset;
+      message.data = manifest.payload + block * block_size + offset;
+      message.data_len = std::min(uint32_t(OTA_FRAG_DATA), length - offset);
+      assert(context.manager.on_message(wire, encode_data(wire, sizeof(wire), message)));
+    }
+    std::vector<uint8_t> scratch(manifest.block_count * 4);
+    uint8_t siblings[128];
+    ProofMsg proof{};
+    memcpy(proof.manifest_id, manifest.merkle_root, 4);
+    proof.block_idx = block;
+    proof.n_proof = merkle_gen_proof(manifest.leaves, manifest.block_count, block,
+                                    scratch.data(), siblings);
+    proof.proof = siblings;
+    assert(context.manager.on_message(wire, encode_proof(wire, sizeof(wire), proof)));
+  }
+};
+
+static void transient_resume_failure_preserves_bytes(HostStream::Failure failure,
+                                                     uint8_t op, uint32_t offset,
+                                                     bool retry_by_pull) {
+  Fixture f;
+  f.host.failure = failure;
+  f.host.fail_op = op;
+  f.host.fail_offset = offset;
+  f.host.failures_left = 2;
+  const auto saved = f.host.bytes;
+  f.pull();
+  f.deliver_manifest();  // Even a late peer response cannot initialize a paused slot.
+  if (f.host.bytes != saved)
+    fputs("FAIL: resume I/O fallback issued BEGIN and erased staging\n", stderr);
+  assert(f.host.bytes == saved && f.host.begins == 0 && f.host.writes == 0);
+  assert(f.context.manager.fetchState() == OtaManager::PAUSED);
+  assert(f.context.manager.fetchError() == OtaManager::FETCH_ERROR_STORAGE);
+  assert(memcmp(f.context.manager.fetchManifestId(), f.manifest.merkle_root, 4) == 0);
+  assert(f.flight.manifests == 0 && f.host.begins == 0 && f.host.writes == 0);
+  f.context.manager.set_clock(1000000);
+  f.context.manager.loop();
+  assert(f.host.bytes == saved && f.flight.manifests == 0);
+  char reply[160];
+  assert(pull_command("pull 1 folder", reply, f.context));
+  assert(strncmp(reply, "ERR OTA slot is paused", 22) == 0);
+  assert(!f.context.manager.resumeFetchAfterReconnect());
+  assert(f.context.manager.fetchState() == OtaManager::PAUSED);
+  assert(f.host.bytes == saved && f.host.begins == 0 && f.flight.manifests == 0);
+  if (retry_by_pull) {
+    f.context.manager.reset_session(); // The manager action performed by explicit cancellation.
+    f.pull();
+  } else {
+    assert(f.context.manager.resumeFetchAfterReconnect());
+  }
+  assert(f.context.manager.fetchState() == OtaManager::VERIFYING_STAGED);
+  f.finish_verification();
+  assert(f.context.manager.fetchState() == OtaManager::COMPLETE);
+  assert(f.context.manager.blocksHave() == f.manifest.block_count);
+  assert(f.context.manager.fetchError() == OtaManager::FETCH_ERROR_NONE);
+  assert(f.host.bytes == saved && f.host.begins == 0 && f.host.writes == 0);
+}
+
+static void known_absent_or_invalid_staging_starts_fresh(unsigned corruption,
+                                                        bool after_io_failure) {
+  Fixture f;
+  if (after_io_failure) {
+    f.host.failure = HostStream::Timeout;
+    f.host.fail_op = MS_OP_STAT;
+    f.host.failures_left = 1;
+    f.pull();
+    assert(f.context.manager.fetchState() == OtaManager::PAUSED);
+  }
+  if (corruption == 0) f.host.present = false;
+  else if (corruption == 1) f.host.bytes[0] ^= 1;
+  else if (corruption == 2) wr_u32le(f.host.bytes.data() + 4, f.host.bytes.size() + 1);
+  else if (corruption == 3) f.host.bytes[8] = 0xFF; // Invalid manifest format version.
+  else if (corruption == 4) f.host.bytes.back() ^= 1;
+  else if (corruption == 5) f.host.bytes.resize(12);
+  else if (corruption == 6) f.host.bytes[8 + 20] ^= 1; // Another MID, read successfully.
+  else wr_u32le(f.host.bytes.data() + 8 + 3, f.manifest.target_id ^ 1u);
+  if (after_io_failure) assert(f.context.manager.resumeFetchAfterReconnect());
+  else f.pull();
+  assert(f.context.manager.fetchState() == OtaManager::WANT_MANIFEST);
+  assert(f.context.manager.fetchError() == OtaManager::FETCH_ERROR_NONE);
+  assert(!f.store.reopenHadIoError());
+  assert(f.flight.manifests == 1 && f.host.begins == 0);
+  f.deliver_manifest();
+  assert(f.context.manager.fetchState() == OtaManager::FETCHING);
+  assert(f.host.begins == 1);
+}
+
+static void explicit_validate_requests_fresh_initialization() {
+  Fixture f;
+  f.host.failure = HostStream::Timeout;
+  f.host.fail_op = MS_OP_STAT;
+  f.host.failures_left = 1;
+  f.pull(true);
+  assert(f.context.manager.fetchState() == OtaManager::WANT_MANIFEST);
+  assert(f.host.operations.empty() && f.flight.manifests == 1);
+  f.deliver_manifest();
+  assert(f.host.begins == 1 && f.host.failures_left == 1);
+}
+
+static void transient_probe_failure_resumes_only_missing_blocks() {
+  Fixture f;
+  const auto complete = f.host.bytes;
+  const uint32_t missing = 1;
+  const uint32_t leaf = f.manifest.leaves - SIM_MOTA_1K + missing * 4;
+  const uint32_t payload = f.manifest.payload - SIM_MOTA_1K + missing * f.manifest.block_size();
+  std::fill(f.host.bytes.begin() + leaf, f.host.bytes.begin() + leaf + 4, 0xFF);
+  std::fill(f.host.bytes.begin() + payload,
+            f.host.bytes.begin() + payload + f.manifest.block_size(), 0xFF);
+  const auto partial = f.host.bytes;
+  f.host.failure = HostStream::Timeout;
+  f.host.fail_offset = 8;
+  f.host.failures_left = 1;
+  f.pull();
+  assert(f.context.manager.fetchState() == OtaManager::PAUSED);
+  assert(f.context.manager.resumeFetchAfterReconnect());
+  f.finish_verification();
+  assert(f.context.manager.fetchState() == OtaManager::FETCHING);
+  assert(f.context.manager.blocksHave() == f.manifest.block_count - 1);
+  assert(f.host.bytes == partial && f.host.begins == 0 && f.host.writes == 0);
+  assert(!f.flight.requested_blocks.empty());
+  for (uint32_t block : f.flight.requested_blocks) assert(block == missing);
+  f.deliver_block(missing);
+  assert(f.context.manager.fetchState() == OtaManager::COMPLETE);
+  assert(f.host.bytes == complete && f.host.begins == 0);
+}
+
+static void established_capture_still_refuses_replacement_file() {
+  Fixture f;
+  f.pull();
+  assert(f.context.manager.fetchState() == OtaManager::VERIFYING_STAGED);
+  assert(f.context.manager.pauseFetchForDisconnect());
+  f.host.bytes[0] ^= 1;
+  assert(!f.context.manager.resumeFetchAfterReconnect());
+  assert(f.context.manager.fetchState() == OtaManager::PAUSED);
+  assert(f.host.begins == 0 && f.flight.manifests == 0);
+}
+
+int main() {
+  for (auto failure : {HostStream::Timeout, HostStream::Truncated, HostStream::BadChecksum,
+                       HostStream::Error, HostStream::ShortWrite, HostStream::WrongOp}) {
+    for (bool retry_by_pull : {false, true}) {
+      transient_resume_failure_preserves_bytes(failure, MS_OP_STAT, 0, retry_by_pull);
+      for (uint32_t offset : {0u, 8u, SIM_MOTA_1K_LEN - 5u})
+        transient_resume_failure_preserves_bytes(failure, MS_OP_SREAD, offset, retry_by_pull);
+    }
+  }
+  for (unsigned corruption = 0; corruption < 8; ++corruption)
+    for (bool after_io_failure : {false, true})
+      known_absent_or_invalid_staging_starts_fresh(corruption, after_io_failure);
+  explicit_validate_requests_fresh_initialization();
+  transient_probe_failure_resumes_only_missing_blocks();
+  established_capture_still_refuses_replacement_file();
+  puts("PASS: folder resume I/O failures preserve staging and recover safely");
+}

test/test_ota_folder_resume.py

@@ -0,0 +1,71 @@
+#!/usr/bin/env python3
+"""Exercise actual folder store, manager, and CLI pull branch across host failures."""
+from pathlib import Path
+import os
+import shutil
+import subprocess
+import tempfile
+import unittest
+
+from test_ota_cli_listing import function
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+class OtaFolderResumeTest(unittest.TestCase):
+    def compile_run(self, manager_source=None, expect_rejection=False):
+        compiler = os.environ.get("CXX") or shutil.which("g++") or shutil.which("clang++")
+        self.assertIsNotNone(compiler)
+        cli = (ROOT / "src/helpers/ota/OtaCli.cpp").read_text()
+        start = cli.index('  } else if (is_cmd(a, "pull|get|download", &rest)) {')
+        end = cli.index('  // ---- discard the current session', start)
+        harness = (ROOT / "test/fixtures/ota_folder_resume.cpp").read_text()
+        helpers = "\n\n".join(function(cli, name) for name in ("is_cmd", "state_word"))
+        harness = harness.replace("// @CLI_HELPERS@", helpers).replace("// @CLI_PULL@", cli[start:end])
+        with tempfile.TemporaryDirectory(prefix="meshcore-ota-folder-resume-") as temporary:
+            work = Path(temporary)
+            source = work / "resume.cpp"
+            source.write_text(harness)
+            binary = work / "resume"
+            sanitizers = (["-fsanitize=address,undefined", "-fno-sanitize-recover=all",
+                           "-fno-pie", "-no-pie"] if os.name != "nt" else [])
+            manager = ROOT / "src/helpers/ota/OtaManager.cpp"
+            if manager_source is not None:
+                manager = work / "OtaManager.cpp"
+                manager.write_text(manager_source)
+            sources = ("FolderMotaStore.cpp", "OtaProtocol.cpp",
+                       "MotaContainer.cpp", "MerkleTree.cpp")
+            built = subprocess.run([
+                compiler, "-std=c++17", "-O1", "-Wall", "-Wextra", "-Werror",
+                "-Wno-unused-parameter", "-Wno-sign-compare", "-Wno-unused-variable",
+                "-DOTA_SEEDER_ONLY=1", *sanitizers,
+                "-I", str(ROOT / "src"), "-I", str(ROOT / "test/mocks"),
+                "-I", str(ROOT / "src/helpers/ota"),
+                "-I", str(ROOT / "test"), str(source),
+                str(manager),
+                *[str(ROOT / "src/helpers/ota" / name) for name in sources],
+                str(ROOT / "src/Utils.cpp"), "-o", str(binary),
+            ], capture_output=True, text=True, timeout=90)
+            self.assertEqual(built.returncode, 0, built.stdout + built.stderr)
+            checked = subprocess.run([str(binary)], capture_output=True, text=True, timeout=30)
+            if expect_rejection:
+                self.assertNotEqual(checked.returncode, 0, checked.stdout + checked.stderr)
+                self.assertIn("resume I/O fallback issued BEGIN and erased staging", checked.stderr)
+                self.assertIn("Assertion", checked.stderr)
+            else:
+                self.assertEqual(checked.returncode, 0, checked.stdout + checked.stderr)
+                self.assertIn("PASS: folder resume I/O failures preserve staging", checked.stdout)
+
+    def test_failed_probe_preserves_staging_and_allows_later_resume(self):
+        self.compile_run()
+
+    def test_reintroduced_fresh_fetch_after_io_failure_is_rejected(self):
+        source = (ROOT / "src/helpers/ota/OtaManager.cpp").read_text()
+        unsafe = source.replace("    if (_fetch_error == FETCH_ERROR_STORAGE) {",
+                                "    if (false) {", 1)
+        self.assertNotEqual(source, unsafe)
+        self.compile_run(unsafe, expect_rejection=True)
+
+
+if __name__ == "__main__":
+    unittest.main()