Speed up and harden LoRa OTA transfers

docs/ota_easy.md

@@ -286,9 +286,10 @@ Monitor the transfer:
 ota status
 ```
 
-The update is ready when the status says `ready to install`. OTA is deliberately the lowest-priority mesh
-traffic. At the temporary-radio settings in this guide, allow roughly **one hour** for a typical ESP32 full
-image over a quiet, direct link. That is a planning estimate, not an upper bound: repeaters, retries, weak
+The update is ready when the status says `ready to install`. Discovery is background traffic, while an
+active OTA download is primary mesh traffic. At the temporary-radio settings in this guide, allow roughly
+**one hour** for a typical ESP32 full image over a quiet, direct link. That is a planning estimate, not an
+upper bound: repeaters, retries, weak
 links, and normal mesh traffic can extend it well past an hour. The 120-minute example window is
 intentional. If necessary, start another overlapping `tempradio` window; the download resumes rather than
 starting over.

docs/ota_protocol.md

@@ -15,8 +15,8 @@ where a section names a source file, that file is the authoritative reference fo
 - **Trustless mesh relay:** repeaters may forward packets while the source alone serves firmware data;
   integrity is content-addressed
   against a signed merkle root, so a relay need not be trusted and never needs the signing keys.
-- **Lowest priority, always:** OTA traffic is enqueued behind all mesh traffic - "eventually upgradable".
-  A busy node delays OTA indefinitely rather than competing with real traffic.
+- **Primary while transferring:** periodic discovery stays at background priority, but manifest, block,
+  data, and proof packets for an active fetch use primary queue priority at every relay hop.
 - **Portable:** the engine (`src/helpers/ota/OtaManager`) is Arduino/radio/crypto-free and host-testable,
   so the same logic drives a device, a simulation, or a third-party implementation.
 
@@ -68,7 +68,8 @@ where a section names a source file, that file is the authoritative reference fo
 | `PAYLOAD_TYPE_OTA` | `0x0C` | MeshCore packet type (`src/Packet.h`) |
 | `MAX_PACKET_PAYLOAD` | `184` | usable bytes per packet (`src/MeshCore.h`) |
 | Default block size | `1024` | `block_size_log2 = 0x0A` |
-| OTA TX priority | `250` | lowest (`OTA_TX_PRIORITY`, `src/Mesh.h`) |
+| OTA discovery TX priority | `250` | background (`OTA_TX_PRIORITY`, `src/Mesh.h`) |
+| OTA active-transfer TX priority | `0` | primary (`OTA_TRANSFER_TX_PRIORITY`, `src/Mesh.h`) |
 
 ---
 
@@ -366,8 +367,9 @@ Message types:
 
 - **`manifest_id`** = the manifest's `merkle_root` (4 bytes) - a compact content id present in every
   transfer message, so a multi-mota server dispatches each request to the right image.
-- **Priority:** all OTA packets enqueue at `OTA_TX_PRIORITY = 250` (lowest). OTA never competes with mesh
-  traffic; on a busy node it is delayed indefinitely.
+- **Priority:** `OTA_ADV`, `OTA_QUERY`, and `OTA_HAVE` enqueue at background priority 250. Once a fetch is
+  active, manifest, block request, data, and proof messages enqueue at primary priority 0. Relay-only
+  nodes classify the wire message identically, so a transfer stays primary across the complete path.
 - **Reliability is *eventual*:** the fetcher re-requests missing fragments/blocks after a timeout, possibly
   from a different peer. No hard ACKs, no global ordering.
 - **Relay:** replies are flooded, so transparent relay needs no per-requester addressing, and the transfer
@@ -377,9 +379,10 @@ Message types:
 - **Hop limit + duty cycle:** OTA floods accumulate one path-hash per relay (the mesh's flood routing). A
   node with the OTA manager *accepts* a packet only if it arrived within `ota config hops` hops (default 3;
   `0` = direct only) and *relays* it only while still under that limit, appending its own hash. Relay-only
-  repeaters instead use their ordinary flood limits and forwarding filters. Relays are lowest-priority and
-  are skipped when the packet pool runs low (the source retries), so heavy OTA can never monopolise a
-  repeater's RAM or starve real traffic.
+  repeaters instead use their ordinary flood limits and forwarding filters. Discovery relays remain
+  lowest-priority and may be skipped when the packet pool runs low. Active-transfer relays bypass that
+  background pool gate and use priority 0; operators should therefore treat TempRadio as a dedicated OTA
+  maintenance window because the transfer can delay unrelated mesh traffic.
 
 ### 8.1 Two-tier discovery
 
@@ -522,7 +525,10 @@ payload); larger self-images pass a bigger scratch buffer.
 
 OTA packets may cross normal mesh relay hops, but each participating node processes or relays them only while
 its `tempradio` window is actually running. A receiver selects missing blocks in serial order into a bounded
-two-block pipeline. It never serves partial blocks. A normal install receiver never re-advertises its completed
+request pipeline. The default capacity is two blocks. Targets with a larger compiled capacity begin at two,
+grow the live window by one after four clean verified blocks, and shrink it after two consecutive no-progress
+service ticks; the RAK3401 LoRa-OTA target currently adapts from two through four blocks. It never serves
+partial blocks. A normal install receiver never re-advertises its completed
 download. An SD archive node is the deliberate exception: after a fully proof-verified container is published
 to its persistent archive, it registers that complete file as a MotaSource and advertises it as a new seeder.
 This keeps each active transfer as one transmitter and one receiver while still allowing active temporary-radio
@@ -712,6 +718,9 @@ ota dev ...                        bring-up helpers (stage/recv/serve/verify)
   locates the physical EndF, hashes the running app, and compares that value with the package before its
   first app write. A physically absent/corrupt EndF or wrong base therefore returns to the unchanged app;
   it still requires USB recovery if that app does not already contain this command.
+  A chain intended to cross historical firmware must introduce this command in its first bridge and retain
+  it in every later bridge. Manual pulls still use the build-provided target ID when app-side EndF parsing
+  fails, so a rescue-capable bridge can fetch its exact successor before invoking the guarded command.
 - **MeshTower V2 SD nRF52:** the application stores a contiguous `/meshcore-ota.mota` on microSD and
   publishes its raw sector range in a checksummed handoff record outside the MBR partition. The matching
   bootloader reads the card without mounting FAT, supports either a full image or an in-place delta,

docs/ota_user_guide.md

@@ -42,16 +42,16 @@ battery telemetry, and `battery.alert` behavior.
 
 ## The important part first: it's safe
 
-- **Nothing installs by itself.** Your node can *discover* and *download* an update in the background, but
+- **Nothing installs by itself.** Your node can *discover* and download an update, but
   it only **installs** when you say so (unless you deliberately turn on auto-install - see below).
 - **Bad downloads can't sneak in.** Every piece of the firmware is checked against a cryptographic
   fingerprint as it arrives, and the whole image is verified again before install. A corrupt or tampered
   download is rejected, not installed.
 - **You choose who to trust.** Updates can be *signed* by their author. You can tell your node to only
   auto-install firmware signed by keys you've added.
-- **It won't disrupt your mesh.** OTA traffic is always the **lowest priority** - your node only spends
-  spare airtime on it. Messages and routing always come first; a busy node simply updates later. Think of
-  it as *"eventually upgradable."*
+- **Discovery stays quiet; a transfer is deliberate.** Periodic update discovery uses background priority.
+  Once a download starts, its transfer packets are primary traffic across every relay hop, so use
+  TempRadio as an OTA maintenance window when delaying unrelated mesh traffic would matter.
 - **It can recover.** If an install ever fails, the node falls back to a safe recovery mode (you can
   re-flash a known-good firmware over USB) - it won't be left bricked.
 
@@ -149,8 +149,10 @@ re-running a `validate` pull re-begins fresh (it never resumes a stale partial).
 trusted - every kept block is checked against the target's own fingerprints, so a mismatched or missing seed
 just means those blocks are fetched over the radio (correct result, only slower).
 
-The node fetches from one source, **at low priority**, with a bounded two-block receive window. Mesh repeaters may carry the
-packets, but only while their temporary-radio windows are active. Check progress with `ota status`.
+The node fetches from one source as **primary traffic**, with a bounded adaptive block-request window.
+RAK3401 OTA builds start with two concurrent block requests, grow to four on a clean link, and contract on
+repeated stalls. Mesh repeaters carry the packets only while their temporary-radio windows are active.
+Check progress with `ota status`.
 
 If a `folder` pull loses its link mid-transfer, `ota status` shows **paused** - the host keeps the
 partial and the pull resumes (filling only what's missing) the moment you reconnect motatool; it never
@@ -190,6 +192,7 @@ After it reboots, run `ota status` to confirm the new version.
   target mismatch, corrupt payload, and invalid/untrusted signatures. The bootloader independently hashes
   the running app and rejects a wrong base before writing the app. If the physical EndF is absent or this
   command was not already in the running firmware, recover over USB.
+  Release chains should put this command in their first bridge and keep it in every bridge after that.
 - If an **install** fails, the node won't boot a broken image - it lands in **recovery mode**:
   - **nRF52:** it appears as a USB drive; drag a known-good firmware `.uf2` for that exact board onto it
     to recover.
@@ -284,8 +287,8 @@ already active. Other FULL roles with browser OTA support can use the
 
 You don't have to be a gateway to help. Once **any** node finishes downloading an update, it automatically
 offers it to *its* neighbours too. So a new firmware spreads outward node-to-node, instead of everyone
-hammering the one node that had it first - and no node is ever overloaded, because all of this stays
-lowest-priority.
+hammering the one node that had it first. Discovery remains background traffic; an actual transfer is
+primary traffic for the duration of its TempRadio maintenance window.
 
 ---
 

docs/rak3401_mota_chain.md

@@ -1,10 +1,8 @@
 # RAK3401 1W repeater LoRa update chain
 
-> Status: the corrected v1.17.01 26-package release candidate has passed
-> complete offline reconstruction, independent container verification, and
-> both stable OTAFIX 2.4 and Preview 6 C simulations. It still needs its first
-> complete physical RAK3401 run and is gated behind
-> `--accept-test-candidate`; do not use it in production yet.
+> Status: the second corrected v1.17.01 candidate is withdrawn. A physical
+> RAK3401 run passed steps 1 through 15, then step 16 booted but could not
+> validate its EndF. The runner blocks this exact bundle before device access.
 
 This procedure records the exact 26-step update chain from the release
 [`rak3401-mota-v1.16.07-c1caa5ad-to-v1.17.01-c96bdd6e`](https://github.com/mikecarper/MeshCore/releases/tag/rak3401-mota-v1.16.07-c1caa5ad-to-v1.17.01-c96bdd6e).
@@ -19,33 +17,45 @@ It is intentionally specific to the following destination firmware:
   `63d8df6387eaffd2e25db7d2a8ad967a65202182a48d681d7e7a9260f917280d`
 
 Do not use this chain on another node, and do not start it on another RAK3401.
-The artifact checks do not replace the pending complete physical-board run.
-Use this candidate only for the next recoverable lab test.
+The physical test has withdrawn this exact chain. Use it only for offline
+diagnosis; the runner refuses live deployment even with
+`--accept-test-candidate`.
 
-## Corrected test-candidate bundle
+## Withdrawn second v1.17.01 candidate
 
-The replacement rebuilds steps 6 through 14 so both `Utils::sha256` overloads
-use software SHA-256; hardware AES and HMAC stay enabled. Temporary steps 14
-through 25 have monotonic EndF versions `1.16.9.110` through `1.16.9.121`.
-The final endpoint is a real v1.17.01 build from `c96bdd6e`, with both its
-runtime string and EndF version set to `1.17.1.0`. No v1.17.02 endpoint is used.
+The replacement rebuilt steps 6 through 15 around the first observed SHA
+failure. Step 15 used software SHA-256 and passed its physical EndF check.
+Step 16 re-enabled CC310 SHA with checked return codes. It downloaded and
+applied successfully, booted `v1.16.9.112-ea3843e0`, then returned `ERR no
+EndF`. This proves that CC310 can report success yet produce a wrong digest
+for the memory-mapped application image; return-code fallback is insufficient.
 
 ```text
 /home/mesh/git/MeshCore/out-rak3401-mota-v1.17.01-corrected/RAK3401-update-chain-v1.16.7-c1caa5ad-to-v1.17.01-c96bdd6e.zip
 ```
 
 ZIP SHA-256:
-`693f08187e42cce72124f01328983965726bfbbb3fef80de503f06c4cbe9256a`.
+`46c7480ed6bdc2aa01fb23a0f70e34c4012ffdd42b616d07bde66cf66d594630`.
 The largest package is 88,280 bytes, leaving 1,832 bytes below the staging
 limit. All 26 deltas reconstruct with both zero-filled and 0xFF-filled
 workspaces. The endpoint has whole-image SHA-256
 `e923f6209e0f071a5862ffbc7690e0b355c1e74829067e0bfdacd0de6c316065`
 and EndF body hash `AC267B02E055F42E`.
 
-The candidate includes the exact v1.16.7 test-start UF2/Nordic DFU ZIP, a
+The withdrawn bundle includes the exact v1.16.7 test-start UF2/Nordic DFU ZIP, a
 diagnostic corrected step-6 image, and the final recovery UF2/ZIP. Restore the
 lab node locally to the included test-start image before rerunning the chain.
-The bundle's `RUNBOOK.md` contains the direct and two-hop commands.
+The bundle's `RUNBOOK.md` is retained for diagnosis only.
+
+## Requirements for the next chain
+
+The next chain must use software SHA-256 for every bridge. CC310 may remain
+enabled for AES, HMAC, and entropy, but not for either `Utils::sha256`
+overload. The guarded `ota rescue install <base_hash16>` command must be in
+the first installed bridge (step 1) and every target after it. Offline
+validation must reconstruct each target and prove that command is present;
+the live runner also probes `ota help` after every successful transition and
+refuses to expose a bridge without the rescue path to another update.
 
 ## Withdrawn v1.17.02-chain smoke-test result
 
@@ -82,21 +92,24 @@ The lab RAK remains alive on v1.16.08.7. Its normal radio is restored to
 
 ## Recover the lab RAK
 
-Recovery now requires a local data path to the RAK3401 itself. The Heltec
-controller cable cannot recover the remote RAK, and this host currently has no
-Bluetooth adapter.
+The latest physical run stopped on `v1.16.9.112-ea3843e0`: normal radio is
+restored to `910.525,62.5,7,5` and `system.watchdog` is verified `on`, but the
+application reports `ERR no EndF`. That bridge predates the guarded rescue
+command, so recovery requires a local data path to the RAK3401 itself. The
+Heltec controller cable cannot recover the remote RAK.
 
 1. Connect the RAK3401 itself to this host by USB.
 2. Enter its exact-board UF2 bootloader, using a local `uf2reset` command or
    the board's double-reset gesture.
-3. Flash the final RAK3401 UF2 from the corrected bundle's `recovery/final/`
-   directory:
-   `RAK_3401_repeater_lora_ota_no_external_sensors-ota-v1.17.01-halo-keymind-cascade-dev-c96bdd6e.uf2`.
-4. After it boots, require version `v1.17.1`, target `2FA509C1`, hardware
-   `RAK_3401`, a valid `ota self` reply, normal radio settings, and watchdog
-   `on`.
+3. For another chain test, flash the exact c1caa5ad test-start UF2 from the
+   withdrawn bundle's `recovery/test-start/` directory. Do not use its final
+   recovery image as a production recovery target; it predates the new
+   software-only SHA rule.
+4. After it boots, require the c1caa5ad start identity, target `2FA509C1`,
+   hardware `RAK_3401`, a valid `ota self` reply, normal radio settings, and
+   watchdog `on`.
 
-Do not send step 7, force an apply, or issue a remote recovery reboot. Once the
+Do not force an apply or issue a remote recovery reboot. Once the
 RAK USB cable is attached, the port and UF2 mount can be detected and the
 recovery completed without guessing device paths.
 
@@ -114,9 +127,10 @@ The dormant live path implements the release's watchdog sequence:
 7. Re-enable the watchdog only after the exact step 26 image boots and passes
    both version and EndF body-hash verification.
 
-These controls are active for the corrected candidate. They cannot repair the
-old bridge whose own EndF check is broken. The candidate must pass a complete
-physical-board run before its explicit lab-only gate is removed.
+These controls remain requirements for the next candidate. They cannot repair
+a running bridge that lacks both valid app-side EndF validation and the rescue
+command. A replacement must pass a complete physical-board run before its
+lab-only gate is removed.
 
 An interrupted download is resumable. Avoid interrupting power or radio
 coverage while `ota install` is applying a package. Physical USB recovery is
@@ -162,7 +176,7 @@ ordinary radio after each step.
 ## Verify without touching a radio
 
 The chain runner pins the release ZIP SHA-256 to
-`693f08187e42cce72124f01328983965726bfbbb3fef80de503f06c4cbe9256a`.
+`46c7480ed6bdc2aa01fb23a0f70e34c4012ffdd42b616d07bde66cf66d594630`.
 It also pins and verifies the inner checksum list, checks complete checksum
 coverage, parses every manifest, verifies the chain continuity, and runs
 `motatool verify` on all 26 containers.
@@ -179,36 +193,26 @@ python3 tools/lora_ota/rak3401_mota_chain.py \
 An already-downloaded asset can be supplied with `--bundle`. Only the pinned
 ZIP or its extracted root is accepted.
 
-## Corrected-release live preflight
+## Withdrawn-release verification
 
-The remote-admin password for the tested RAK is the lowercase default
-`password`. Prefer the environment variable so it is not placed in the child
-process command line:
+No password or device connection is needed for offline verification:
 
 ```bash
-export MESHCORE_ADMIN_PASSWORD='password'
-
 python3 tools/lora_ota/rak3401_mota_chain.py \
   --work-dir ./rak3401-mota-chain-work \
-  --controller-serial /dev/ttyACM0 \
-  --source-tcp 192.168.1.51:5001 \
-  --source-cli-tcp 192.168.1.51:5002 \
-  --source-shares-controller \
-  --ota-hops 3 \
-  --accept-test-candidate \
   --motatool /path/to/motatool \
-  --preflight-only
+  --verify-only
 ```
 
-This validates the corrected bundle, Full Companion identity, source, and
-destination without changing radio or watchdog settings. Omitting
-`--accept-test-candidate` exits before connecting to a device. The withdrawn
-v1.17.02 chain is also recognized and always blocked before device access.
+Use `--verify-only` for this withdrawn bundle. Every live invocation exits
+before connecting to a device, including one with `--accept-test-candidate`.
+The older failed v1.17.01 and v1.17.02 chains are also recognized and blocked.
 
-## Direct lab template
+## Future direct lab template
 
-Use only on a locally recoverable test RAK3401. Keep the work directory for
-resume, and retain the explicit `--accept-test-candidate` lab gate.
+This template becomes usable only after a new ZIP and its hashes are pinned in
+the runner. Use it only on a locally recoverable test RAK3401. Keep the work
+directory for resume, and retain the explicit `--accept-test-candidate` gate.
 
 ```bash
 export MESHCORE_ADMIN_PASSWORD='password'
@@ -227,9 +231,10 @@ python3 tools/lora_ota/rak3401_mota_chain.py \
 
 The intended runner logs each isolated step attempt below `steps/`, records every
 verified transition in `progress.jsonl`, and prints the source log path. Rerun
-the exact command after a host restart or recoverable RF failure. Its default
-60-second status interval deliberately leaves airtime for the low-priority OTA
-transfer; avoid shortening it on a busy or relayed mesh.
+the exact command after a host restart or recoverable RF failure. Status polling
+starts at 60 seconds, expands adaptively when replies are slow or the link is
+contended, and contracts after quick replies. The transfer itself is primary
+traffic, so reserve the TempRadio window for OTA on a busy or relayed mesh.
 
 ## Relayed lab template
 
@@ -280,7 +285,7 @@ python3 tools/lora_ota/rak3401_mota_chain.py \
   --yes
 ```
 
-Do not begin a production deployment until the corrected bundle passes a
+Do not begin a production deployment until a replacement bundle passes a
 direct physical smoke test from start to endpoint. Run `--preflight-only` with
 the same relay arguments first. The runner sends
 the destination to TempRadio, then the relays farthest-to-nearest while their

platformio.ini

@@ -253,6 +253,7 @@ test_framework = googletest
 build_flags = -std=c++17
   -I src
   -I test/mocks
+  -D OTA_FETCH_PIPELINE=4
 test_build_src = yes
 test_ignore = test_kiss_modem
 build_src_filter =

src/Mesh.cpp

@@ -144,8 +144,17 @@ void Mesh::configureDirectRetryPacket(Packet* retry, const Packet* original, uin
 
   retry->tx_cr = getDirectRetryCodingRateForAttempt(default_cr, retry_attempt);
 }
+
+static bool isPrimaryOtaTraffic(const uint8_t* msg, uint16_t len) {
+  return msg && len && ota::ota_is_transfer_message(msg[0]);
+}
+
+static uint8_t otaTrafficPriority(const uint8_t* msg, uint16_t len) {
+  return isPrimaryOtaTraffic(msg, len) ? OTA_TRANSFER_TX_PRIORITY : OTA_TX_PRIORITY;
+}
+
 #if defined(ENABLE_OTA)
-// Adapter so the portable OtaManager can emit packets through the mesh (lowest priority, hop-capped).
+// Adapter so the portable OtaManager can emit packets through the mesh (message-priority, hop-capped).
 void Mesh::otaSendAdapter(void* ctx, const uint8_t* msg, uint16_t len, bool /*flood*/) {
   Mesh* m = (Mesh*)ctx;
   if (!m->isTempRadioActive()) return;
@@ -1006,6 +1015,8 @@ DispatcherAction Mesh::onRecvPacket(Packet* pkt) {
       // ONLY to avoid re-flooding the same packet more than once.
       bool seen = _tables->wasSeen(pkt);
       if (!seen) _tables->markSeen(pkt);
+      const bool primary_ota = isPrimaryOtaTraffic(pkt->payload, pkt->payload_len);
+      const uint8_t ota_priority = otaTrafficPriority(pkt->payload, pkt->payload_len);
 #if defined(ENABLE_OTA)
       ota::ota_ctx().manager.set_clock(_ms->getMillis());                 // discovery jitter/ages
       ota::ota_ctx().manager.on_message(pkt->payload, pkt->payload_len);  // central OTA receive (beacon/query/
@@ -1013,20 +1024,19 @@ DispatcherAction Mesh::onRecvPacket(Packet* pkt) {
       ota::ota_ctx().track_session(ota::ota_ctx().manager.fetchState(), _ms->getMillis());
       onOtaRecv(pkt);                                                     // optional per-example hook
 #endif
-      // Re-flood at the LOWEST priority and only while still under the hop limit, so OTA never competes with
-      // mesh traffic. The free-pool guard keeps heavy OTA from monopolising the shared packet pool - dropping
-      // a relay is safe (OTA is best-effort; the source retries). A relay-only build has no OTA-specific hop
-      // preference, so its ordinary repeater flood limits in allowPacketForward() remain authoritative.
+      // Re-flood discovery at background priority, but keep an active transfer primary at every relay hop.
+      // The free-pool reserve still sheds periodic discovery under pressure; requested transfer packets are
+      // protected from that background-only gate. Ordinary hop and forwarding policy remain authoritative.
       if (!seen && pkt->isRouteFlood() && !pkt->isMarkedDoNotRetransmit()
 #if defined(ENABLE_OTA)
           && n < getOtaHopLimit()
 #endif
           && (n + 1) * pkt->getPathHashSize() <= MAX_PATH_SIZE
-          && _mgr->getFreeCount() > OTA_FWD_MIN_FREE
+          && (primary_ota || _mgr->getFreeCount() > OTA_FWD_MIN_FREE)
           && allowPacketForward(pkt)) {
         self_id.copyHashTo(&pkt->path[n * pkt->getPathHashSize()], pkt->getPathHashSize());
         pkt->setPathHashCount(n + 1);
-        action = ACTION_RETRANSMIT_DELAYED(OTA_TX_PRIORITY, getOtaRetransmitDelay(pkt));
+        action = ACTION_RETRANSMIT_DELAYED(ota_priority, getOtaRetransmitDelay(pkt));
       }
       break;
     }
@@ -2634,7 +2644,7 @@ void Mesh::sendOtaFlood(Packet* packet, uint32_t delay_millis) {
   packet->header |= ROUTE_TYPE_FLOOD;
   packet->setPathHashSizeAndCount(1, 0);
   _tables->markSeen(packet);   // mark as sent, in case it floods back to us
-  sendPacket(packet, OTA_TX_PRIORITY, delay_millis);
+  sendPacket(packet, otaTrafficPriority(packet->payload, packet->payload_len), delay_millis);
 }
 #endif
 

src/Mesh.h

@@ -4,12 +4,16 @@
 
 // OTA-over-LoRa transport is understood by every Mesh role even when the OTA manager/installer is not
 // compiled in. Repeaters can therefore relay PAYLOAD_TYPE_OTA opaquely while they are on TempRadio.
-// OTA traffic always uses the lowest TX priority (selected only after all real traffic).
+// Periodic OTA discovery stays at the lowest priority. Once an image transfer starts, its manifest,
+// block, and proof exchange is a primary operation at every origin and relay hop.
 #ifndef OTA_TX_PRIORITY
 #define OTA_TX_PRIORITY 250
 #endif
-// Keep this many pool slots free while relaying OTA, so best-effort firmware traffic cannot monopolise the
-// shared packet pool and starve real traffic.
+#ifndef OTA_TRANSFER_TX_PRIORITY
+#define OTA_TRANSFER_TX_PRIORITY 0
+#endif
+// Keep this many pool slots free while relaying background OTA discovery. An explicitly requested transfer
+// is primary and bypasses this discovery-only reserve gate.
 #ifndef OTA_FWD_MIN_FREE
 #define OTA_FWD_MIN_FREE 4
 #endif
@@ -545,7 +549,8 @@ public:
 #if defined(ENABLE_OTA)
   // Build a PAYLOAD_TYPE_OTA packet from raw OTA message bytes (route set by sendOtaFlood).
   Packet* createOtaPacket(const uint8_t* data, size_t len);
-  // Flood-send at the lowest priority (so OTA never competes with mesh traffic).
+  // Flood-send at the priority derived from the OTA message: discovery is background, active transfer is
+  // primary. Relay-only nodes use the same classification.
   void sendOtaFlood(Packet* packet, uint32_t delay_millis = 0);
 #endif
   Packet* createTrace(uint32_t tag, uint32_t auth_code, uint8_t flags = 0);

src/Utils.cpp

@@ -158,53 +158,15 @@ uint32_t RNG::nextInt(uint32_t _min, uint32_t _max) {
 }
 
 void Utils::sha256(uint8_t *hash, size_t hash_len, const uint8_t* msg, int msg_len) {
-#ifdef USE_CC310_HW_CRYPTO
-  if (hash_len <= SHA256::HASH_SIZE && msg_len >= 0) {
-    CC310CryptoSession session;
-    if (session) {
-      static CRYS_HASH_Result_t result;
-      const CRYSError_t rc = CRYS_HASH(
-          CRYS_HASH_SHA256_mode, const_cast<uint8_t*>(msg),
-          static_cast<size_t>(msg_len), result);
-      if (rc == CRYS_OK) {
-        memcpy(hash, result, hash_len);
-        return;
-      }
-    }
-  }
-#endif
+  // CC310 can report CRYS_OK yet return a wrong digest when the input is the
+  // memory-mapped nRF52 application image. SHA-256 protects firmware identity,
+  // Merkle proofs, and install gates, so a return-code fallback is insufficient:
+  // always use the deterministic software implementation. CC310 remains in use
+  // for AES, HMAC, and entropy, where its inputs live in ordinary RAM.
   sha256Software(hash, hash_len, msg, msg_len);
 }
 
 void Utils::sha256(uint8_t *hash, size_t hash_len, const uint8_t* frag1, int frag1_len, const uint8_t* frag2, int frag2_len) {
-#ifdef USE_CC310_HW_CRYPTO
-  if (hash_len <= SHA256::HASH_SIZE && frag1_len > 0 && frag2_len > 0) {
-    CC310CryptoSession session;
-    if (session) {
-      static CRYS_HASHUserContext_t ctx;
-      static CRYS_HASH_Result_t result;
-      CRYSError_t rc = CRYS_HASH_Init(&ctx, CRYS_HASH_SHA256_mode);
-      const bool initialized = rc == CRYS_OK;
-      if (rc == CRYS_OK) {
-        rc = CRYS_HASH_Update(&ctx, const_cast<uint8_t*>(frag1),
-                              static_cast<size_t>(frag1_len));
-      }
-      if (rc == CRYS_OK) {
-        rc = CRYS_HASH_Update(&ctx, const_cast<uint8_t*>(frag2),
-                              static_cast<size_t>(frag2_len));
-      }
-      if (rc == CRYS_OK) rc = CRYS_HASH_Finish(&ctx, result);
-      if (rc == CRYS_OK) {
-        memcpy(hash, result, hash_len);
-        return;
-      }
-      if (initialized) {
-        const bool context_freed = CRYS_HASH_Free(&ctx) == CRYS_OK;
-        (void) context_freed;
-      }
-    }
-  }
-#endif
   sha256Software(hash, hash_len, frag1, frag1_len, frag2, frag2_len);
 }
 

src/helpers/ota/MotaSeederProto.h

@@ -8,8 +8,8 @@
 // it never contends with the line-based text CLI on the main console.
 //
 // The device always initiates; every request gets exactly one response. Framing is resync-safe: the
-// reader scans for the 2-byte magic, so line noise / a half-read frame just times out and is retried
-// (OTA is lowest priority - eventually-upgradable). All multi-byte fields are little-endian.
+// reader scans for the 2-byte magic, so line noise / a half-read frame just times out and is retried.
+// Active transfers are primary mesh traffic; discovery remains background. All multi-byte fields are little-endian.
 //
 //   request  (device -> host):  'M' 'S'  op(1)  args...                     xsum(1 = XOR of op+args)
 //   response (host -> device):  'm' 's'  op(1)  status(1)  payload...       xsum(1 = XOR of all prior)

src/helpers/ota/MotaSourceSerial.h

@@ -6,8 +6,8 @@
 // A MotaSource backed by a host "mota-seeder" daemon over a dedicated Stream (a spare UART / USB-UART).
 // The device pulls catalog + bytes on demand (MotaSeederProto.h); the folder image is never held on the
 // device - it streams through. Use a stream that is NOT the text-CLI console so the binary framing never
-// collides with command/log text. Reads block on the Stream up to `timeout_ms` (OTA is lowest priority,
-// so a serial round-trip's latency is acceptable; keep the daemon on a fast link).
+// collides with command/log text. Reads block on the Stream up to `timeout_ms`; keep the daemon on a fast
+// link so a serial round-trip does not stall the primary transfer.
 
 namespace mesh {
 namespace ota {

src/helpers/ota/OtaCli.cpp

@@ -348,7 +348,7 @@ bool handle_ota_command(const char* command, char* reply, mesh::MainBoard& board
     c.manager.set_fetch_store(store);                        // stage this pull to the chosen destination
     c.manager.pull(selmid, seltgt, validate);                // sets want + begins the manifest fetch now
     char midhx[9]; mesh::Utils::toHex(midhx, selmid, 4);
-    snprintf(reply, 160, "OK pulling mid=%s -> %s (low priority)", midhx, dname);
+    snprintf(reply, 160, "OK pulling mid=%s -> %s (primary traffic)", 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/OtaFormat.h

@@ -87,6 +87,25 @@ enum OtaMsgType : uint8_t {
   OTA_LEAVES       = 0x0B,   // a fragment of the leaves[] array (for host-side seed leaf-diff)
 };
 
+// Discovery is periodic background traffic; every packet that advances an active fetch is primary
+// traffic. Keep this classification in the wire-format layer so origins and opaque relay-only builds make
+// the same queueing decision without needing an OtaManager.
+inline bool ota_is_transfer_message(uint8_t type) {
+  switch (type) {
+    case OTA_GET_MANIFEST:
+    case OTA_MANIFEST:
+    case OTA_REQ:
+    case OTA_DATA:
+    case OTA_REQ_PROOF:
+    case OTA_PROOF:
+    case OTA_GET_LEAVES:
+    case OTA_LEAVES:
+      return true;
+    default:
+      return false;  // ADV / QUERY / HAVE and unknown future messages stay background-safe
+  }
+}
+
 static const uint16_t OTA_DEFAULT_BLOCK_SIZE = 1024;
 static const uint8_t  OTA_DEFAULT_HOP_LIMIT  = 3;
 

src/helpers/ota/OtaManager.cpp

@@ -620,6 +620,9 @@ void OtaManager::clearReassembly() {
   for (uint8_t slot = 0; slot < OTA_FETCH_PIPELINE; slot++) clearReassemblySlot(slot);
   _retry_slot = 0;
   _req_count = 0;
+  _pipeline_width = OTA_FETCH_PIPELINE_INITIAL;
+  _pipeline_success_streak = 0;
+  _pipeline_stall_streak = 0;
 }
 
 int OtaManager::findReassemblySlot(uint32_t block) const {
@@ -630,12 +633,35 @@ int OtaManager::findReassemblySlot(uint32_t block) const {
 }
 
 int OtaManager::findEmptyReassemblySlot() const {
-  for (uint8_t slot = 0; slot < OTA_FETCH_PIPELINE; slot++) {
+  for (uint8_t slot = 0; slot < _pipeline_width; slot++) {
     if (_reasm[slot].block == NO_BLOCK) return slot;
   }
   return -1;
 }
 
+void OtaManager::notePipelineBlockVerified() {
+  _pipeline_stall_streak = 0;
+  if (_pipeline_width >= OTA_FETCH_PIPELINE) return;
+  if (++_pipeline_success_streak < OTA_FETCH_PIPELINE_GROW_BLOCKS) return;
+  _pipeline_success_streak = 0;
+  _pipeline_width++;
+  OTA_DBG("OTA: request pipeline grew to %u/%u after clean block progress\n",
+          (unsigned)_pipeline_width, (unsigned)OTA_FETCH_PIPELINE);
+}
+
+void OtaManager::notePipelineStall() {
+  _pipeline_success_streak = 0;
+  if (_pipeline_width <= OTA_FETCH_PIPELINE_INITIAL) {
+    _pipeline_stall_streak = 0;
+    return;
+  }
+  if (++_pipeline_stall_streak < OTA_FETCH_PIPELINE_SHRINK_TICKS) return;
+  _pipeline_stall_streak = 0;
+  _pipeline_width--;
+  OTA_DBG("OTA: request pipeline shrank to %u/%u after stalled ticks\n",
+          (unsigned)_pipeline_width, (unsigned)OTA_FETCH_PIPELINE);
+}
+
 bool OtaManager::blockInPipeline(uint32_t block) const {
   return findReassemblySlot(block) >= 0;
 }
@@ -1015,6 +1041,7 @@ void OtaManager::handleProof(const uint8_t* m, uint16_t n) {
   }
   _have++;
   OTA_DBG("OTA: block %u OK  have=%u/%u\n", (unsigned)block, (unsigned)_have, (unsigned)_fbc);
+  notePipelineBlockVerified();
   clearReassemblySlot((uint8_t)slot_index);
   // periodically persist progress (meta/leaf page + open payload) so a reboot can resume (no-op for RAM);
   // cadence is runtime-tunable via `ota config checkpoint <N>` (0 = never)
@@ -1079,6 +1106,7 @@ void OtaManager::requestMissing() {
 
   // A full window with no progress is stalled. Retry one slot per service tick, round-robin, so a weak
   // block cannot trigger a simultaneous multi-block re-burst on a half-duplex channel.
+  notePipelineStall();
   for (uint8_t offset = 0; offset < OTA_FETCH_PIPELINE; offset++) {
     uint8_t slot = (uint8_t)((_retry_slot + offset) % OTA_FETCH_PIPELINE);
     if (_reasm[slot].block == NO_BLOCK) continue;

src/helpers/ota/OtaManager.h

@@ -134,6 +134,28 @@ typedef bool (*ServeReadFn)(void* ctx, uint32_t off, uint8_t* buf, uint32_t len)
 #if OTA_FETCH_PIPELINE < 1
 #error "OTA_FETCH_PIPELINE must be at least 1"
 #endif
+#ifndef OTA_FETCH_PIPELINE_INITIAL
+  #if OTA_FETCH_PIPELINE >= 2
+    #define OTA_FETCH_PIPELINE_INITIAL 2
+  #else
+    #define OTA_FETCH_PIPELINE_INITIAL 1
+  #endif
+#endif
+#if OTA_FETCH_PIPELINE_INITIAL < 1 || OTA_FETCH_PIPELINE_INITIAL > OTA_FETCH_PIPELINE
+#error "OTA_FETCH_PIPELINE_INITIAL must be between 1 and OTA_FETCH_PIPELINE"
+#endif
+#ifndef OTA_FETCH_PIPELINE_GROW_BLOCKS
+#define OTA_FETCH_PIPELINE_GROW_BLOCKS 4  // clean verified blocks before adding one concurrent request slot
+#endif
+#if OTA_FETCH_PIPELINE_GROW_BLOCKS < 1
+#error "OTA_FETCH_PIPELINE_GROW_BLOCKS must be at least 1"
+#endif
+#ifndef OTA_FETCH_PIPELINE_SHRINK_TICKS
+#define OTA_FETCH_PIPELINE_SHRINK_TICKS 2 // consecutive 3-second no-progress ticks before removing one slot
+#endif
+#if OTA_FETCH_PIPELINE_SHRINK_TICKS < 1
+#error "OTA_FETCH_PIPELINE_SHRINK_TICKS must be at least 1"
+#endif
 // nRF52 note: a flash page-erase halts the CPU (~85 ms, code runs from flash) and starves the LoRa RX,
 // so writing to flash on every received packet drops in-flight DATA and the transfer stalls. The SD-safe
 // driver (Adafruit flash_nrf5x) always erases on flush, so there is no erase-free write; instead
@@ -332,6 +354,8 @@ public:
   FetchState fetchState() const { return _fstate; }
   uint32_t blocksHave() const { return _have; }
   uint32_t blocksTotal() const { return _fbc; }
+  uint8_t fetchPipelineWidth() const { return _pipeline_width; }
+  static constexpr uint8_t fetchPipelineCapacity() { return OTA_FETCH_PIPELINE; }
   const uint8_t* fetchManifestId() const { return _fid; }
 
   // --- discovery catalog (for `ota neighbors`): mOTAs heard around us via OTA_HAVE, deduped by mid ---
@@ -382,6 +406,8 @@ private:
   int findReassemblySlot(uint32_t block) const;
   int findEmptyReassemblySlot() const;
   bool blockInPipeline(uint32_t block) const;
+  void notePipelineBlockVerified();
+  void notePipelineStall();
   uint32_t pipelineProgress() const;
   uint32_t pickMissingBlock();                            // choose the next missing block not already in flight
   int  serveEntryIndex(const uint8_t* mid) const;         // registry slot serving this mid (-1 if none)
@@ -465,9 +491,9 @@ private:
   uint16_t   _advert_mins = OTA_ADVERT_INTERVAL_MINS;    // beacon re-advertise cadence, minutes; 0=off (persisted)
   uint8_t    _max_hops = OTA_HOP_LIMIT_DEFAULT;          // OTA flood reach in hops; 0=direct only (persisted)
   uint8_t    _fflags = 0;                       // flags of the manifest currently being fetched
-  // Bounded multi-block client pipeline. Each slot independently reassembles DATA and awaits its Merkle
-  // proof, so proof/flash work for one block can overlap radio delivery of the next. The wire protocol and
-  // opaque repeater behavior are unchanged.
+  // Bounded adaptive multi-block client pipeline. Each slot independently reassembles DATA and awaits its
+  // Merkle proof, so proof/flash work for one block can overlap radio delivery of the next. The live window
+  // grows on clean verified blocks and contracts after stalled service ticks; the wire protocol is unchanged.
   struct ReassemblySlot {
     uint32_t block = NO_BLOCK;
     uint16_t mask = 0;                         // received FRAG_DATA-slice bitmap
@@ -478,6 +504,9 @@ private:
   ReassemblySlot _reasm[OTA_FETCH_PIPELINE];
   uint32_t   _loop_last_progress = 0;           // aggregate pipeline progress for stall detection
   uint8_t    _retry_slot = 0;                   // round-robin stalled-slot retry cursor
+  uint8_t    _pipeline_width = OTA_FETCH_PIPELINE_INITIAL; // live request window, adaptive up to array capacity
+  uint8_t    _pipeline_success_streak = 0;      // verified blocks since the last grow/stall decision
+  uint8_t    _pipeline_stall_streak = 0;        // consecutive no-progress service ticks
   // multi-fragment manifest reassembly (a signed v2 manifest exceeds one packet)
   uint8_t    _mf_buf[OTA_MF_MAXFRAG * OTA_MF_FRAG];   // sized to the fragment cap so no valid manifest is silently dropped
   uint16_t   _mf_retries = 0;                          // GET_MANIFEST retries while WANT_MANIFEST (give up after a cap)

src/helpers/ota/OtaSource.h

@@ -47,8 +47,8 @@ public:
   // Cheap metadata + region offsets for mota `idx`. False if idx is out of range or unparsable.
   virtual bool    describe(uint8_t idx, MotaDesc& out) = 0;
   // Random-access read of `len` bytes at absolute offset `off` of mota `idx` into `buf`. Returns true iff
-  // exactly `len` bytes were produced. May block on the transport (serial round-trip); OTA is lowest
-  // priority so latency here is acceptable.
+  // exactly `len` bytes were produced. May block on the transport (serial round-trip); implementations
+  // should keep active-transfer reads prompt because requested OTA traffic is primary traffic.
   virtual bool    read(uint8_t idx, uint32_t off, uint8_t* buf, uint32_t len) = 0;
 };
 

src/helpers/ota/OtaStoreFlashEsp32.h

@@ -27,8 +27,8 @@
 // COALESCED to the 4 KB sector and each sector is programmed once. The meta region (header+manifest+
 // leaves -- written all transfer long as blocks/leaves arrive, often out of order) is PINNED in RAM and
 // flushed at finalize() with the radio idle; the bulk payload streams through ONE sliding sector buffer,
-// flushing the sector it leaves behind (~1 flush per 4 KB, off the per-packet path). OTA is also the
-// lowest-priority TX, so a brief stall yields to real traffic. A small delta whose container fits the
+// flushing the sector it leaves behind (~1 flush per 4 KB, off the per-packet path). A brief stall still
+// yields between radio packets. A small delta whose container fits the
 // pinned meta region does ZERO flash I/O until COMPLETE.
 
 namespace mesh {

test/README.md

@@ -48,6 +48,8 @@ does not reflect the GoogleTest count -- run the built binary directly
 | `test_companion_frame_queue` | `src/helpers/CompanionFrameQueue.h` | response/required/best-effort classification; reserved capacity; stable priority; safe eviction; message-waiting coalescing |
 | `test_serial_mode_switch` | `src/helpers/ArduinoSerialInterface.cpp`, `src/helpers/MultiSerialInterface.h` | independent terminal/seeder control-sequence recognition across reads and binary-frame boundaries; passthrough ownership of USB input and suppression of binary output; Bluetooth-only connection and pairing-request routing |
 | `test_ble_tx_stall_watchdog` | `src/helpers/BleTxStallWatchdog.h` | exact BLE fragment progress; blocked-reply timeout; rollover-safe elapsed time; disconnect recovery retry and completion |
+| `test_ota` | `src/helpers/ota/` | container and EndF integrity; protocol codecs; transfer, resume, and apply safety; adaptive 2-to-4 block-request window growth and stall contraction; active-transfer priority classification |
+| `test_trace_retry` | `src/Mesh.cpp` retry and relay policy | opaque OTA relay behavior during TempRadio; background discovery priority; primary active-transfer priority and packet-reserve bypass; trace and flood retry timing |
 | `test_utils` | `src/Utils.cpp` | `Utils::toHex` (upstream) |
 
 ## Conventions (and how to add a suite)

test/test_ota/test_ota_core.cpp

@@ -413,6 +413,21 @@ TEST(OtaMerkle, GenProofMatchesPythonAndVerifies) {
 
 // --- protocol codec round-trips ---------------------------------------------------------------
 
+TEST(OtaProtocol, ClassifiesOnlyActiveTransferMessagesAsPrimary) {
+  EXPECT_FALSE(ota_is_transfer_message(OTA_ADV));
+  EXPECT_FALSE(ota_is_transfer_message(OTA_QUERY));
+  EXPECT_FALSE(ota_is_transfer_message(OTA_HAVE));
+  EXPECT_TRUE(ota_is_transfer_message(OTA_GET_MANIFEST));
+  EXPECT_TRUE(ota_is_transfer_message(OTA_MANIFEST));
+  EXPECT_TRUE(ota_is_transfer_message(OTA_REQ));
+  EXPECT_TRUE(ota_is_transfer_message(OTA_DATA));
+  EXPECT_TRUE(ota_is_transfer_message(OTA_REQ_PROOF));
+  EXPECT_TRUE(ota_is_transfer_message(OTA_PROOF));
+  EXPECT_TRUE(ota_is_transfer_message(OTA_GET_LEAVES));
+  EXPECT_TRUE(ota_is_transfer_message(OTA_LEAVES));
+  EXPECT_FALSE(ota_is_transfer_message(0xFF));
+}
+
 TEST(OtaProtocol, CodecRoundTrips) {
   uint8_t buf[200];
 
@@ -775,6 +790,77 @@ TEST(OtaTransfer, ClientPipelinesAndRefillsOutOfOrderBlocks) {
   EXPECT_EQ(refill.block_idx, 2);
 }
 
+#if OTA_FETCH_PIPELINE >= 3
+TEST(OtaTransfer, RequestWindowGrowsOnCleanBlocksAndShrinksOnStalls) {
+  MotaManifest manifest;
+  ASSERT_TRUE(mota_parse(SIM_MOTA, SIM_MOTA_LEN, manifest));
+  ASSERT_GE(manifest.block_count, 7u);
+
+  OtaManager client;
+  OtaStoreRam<4096> store;
+  CapturedMessages sent;
+  client.begin(SIM_TARGET_ID, capture_send, &sent);
+  client.set_fetch_store(&store);
+  client.pull(manifest.merkle_root, manifest.target_id);
+  ASSERT_EQ(client.fetchPipelineCapacity(), 4u);
+  sent.items.clear();
+  deliver_manifest_fragment(client, manifest.merkle_root, 0,
+                            manifest.manifest_start, OTA_MF_FRAG);
+  deliver_manifest_fragment(client, manifest.merkle_root, 1,
+                            manifest.manifest_start + OTA_MF_FRAG,
+                            (uint16_t)(MOTA_MFL - OTA_MF_FRAG));
+  client.loop();
+  ASSERT_EQ(client.fetchPipelineWidth(), 2u);
+
+  auto deliver_verified_block = [&](uint32_t block) {
+    const uint32_t block_size = manifest.block_size();
+    const uint32_t block_len = block + 1 < manifest.block_count
+        ? block_size : manifest.payload_size - block * block_size;
+    uint8_t wire[MAX_PACKET_PAYLOAD];
+    for (uint32_t offset = 0; offset < block_len; offset += OTA_FRAG_DATA) {
+      uint32_t length = block_len - offset;
+      if (length > OTA_FRAG_DATA) length = OTA_FRAG_DATA;
+      DataMsg data;
+      memcpy(data.manifest_id, manifest.merkle_root, 4);
+      data.block_idx = block;
+      data.frag_off = (uint16_t)offset;
+      data.data = manifest.payload + block * block_size + offset;
+      data.data_len = (uint16_t)length;
+      uint16_t wire_len = encode_data(wire, sizeof(wire), data);
+      ASSERT_GT(wire_len, 0);
+      client.on_message(wire, wire_len);
+    }
+
+    std::vector<uint8_t> scratch(manifest.block_count * 4);
+    uint8_t siblings[32 * 4];
+    uint8_t sibling_count = merkle_gen_proof(
+        manifest.leaves, manifest.block_count, block, scratch.data(), siblings);
+    ProofMsg proof;
+    memcpy(proof.manifest_id, manifest.merkle_root, 4);
+    proof.block_idx = block;
+    proof.n_proof = sibling_count;
+    proof.proof = siblings;
+    uint16_t wire_len = encode_proof(wire, sizeof(wire), proof);
+    ASSERT_GT(wire_len, 0);
+    client.on_message(wire, wire_len);
+  };
+
+  for (uint32_t block = 0; block < 3; block++) {
+    deliver_verified_block(block);
+    EXPECT_EQ(client.fetchPipelineWidth(), 2u);
+  }
+  deliver_verified_block(3);
+  EXPECT_EQ(client.blocksHave(), 4u);
+  EXPECT_EQ(client.fetchPipelineWidth(), 3u);
+
+  client.loop();  // establish the new committed-block/progress baseline
+  client.loop();  // first no-progress service tick
+  EXPECT_EQ(client.fetchPipelineWidth(), 3u);
+  client.loop();  // second no-progress service tick contracts the request window
+  EXPECT_EQ(client.fetchPipelineWidth(), 2u);
+}
+#endif
+
 TEST(OtaTransfer, PipelineRetriesOneSlotAndOnlyItsMissingFragments) {
   MotaManifest manifest;
   ASSERT_TRUE(mota_parse(SIM_MOTA_1K, SIM_MOTA_1K_LEN, manifest));
@@ -1341,7 +1427,7 @@ TEST(OtaCatalog, DistinctSeederCountAndHaveCount) {
 }
 
 // An unanswered GET_MANIFEST must not pin the fetch slot forever: after OTA_MANIFEST_MAX_RETRY ticks with
-// no manifest, the session gives up (FAILED) so a new pull can take the slot. (Lowest-priority + bounded.)
+// no manifest, the session gives up (FAILED) so a new pull can take the slot. (Bounded primary operation.)
 TEST(OtaTransfer, ManifestGiveUpAfterRetries) {
   g_q.clear();
   OtaManager client; OtaStoreRam<4096> store; SendTo to_none{&client};

test/test_trace_retry/test_trace_retry.cpp

@@ -4,6 +4,7 @@
 #include <Mesh.h>
 #include <helpers/ClockSyncUtils.h>
 #include <helpers/StaticPoolPacketManager.h>
+#include <helpers/ota/OtaFormat.h>
 
 class TraceTestClock : public mesh::MillisecondClock {
 public:
@@ -195,7 +196,7 @@ TEST(RepeaterTransport, UnknownFloodPayloadHonorsReceiveAndForwardingRejections)
   EXPECT_EQ(1, tables.mark_seen_calls);
 }
 
-TEST(RepeaterTransport, OtaFloodRelaysWithoutOtaManagerOnlyDuringTempRadio) {
+TEST(RepeaterTransport, OtaDiscoveryRelaysInBackgroundOnlyDuringTempRadio) {
   TraceTestClock clock;
   TraceTestRTC rtc;
   TraceTestRNG rng;
@@ -206,6 +207,7 @@ TEST(RepeaterTransport, OtaFloodRelaysWithoutOtaManagerOnlyDuringTempRadio) {
   node.forwardFloods = true;
 
   mesh::Packet packet = makeFloodPacket(PAYLOAD_TYPE_OTA);
+  packet.payload[0] = mesh::ota::OTA_ADV;
   EXPECT_FALSE(node.canTransmit(&packet));
   EXPECT_EQ(ACTION_RELEASE, node.receivePacket(&packet));
   EXPECT_EQ(0, tables.mark_seen_calls);
@@ -223,6 +225,27 @@ TEST(RepeaterTransport, OtaFloodRelaysWithoutOtaManagerOnlyDuringTempRadio) {
   EXPECT_FALSE(node.canTransmit(&packet));  // queued-near-expiry packets cannot leak onto the normal channel
 }
 
+TEST(RepeaterTransport, OtaTransferRelaysAsPrimaryTrafficWithoutOtaManager) {
+  TraceTestClock clock;
+  TraceTestRTC rtc;
+  TraceTestRNG rng;
+  TraceTestRadio radio;
+  ForwardingTestTables tables;
+  StaticPoolPacketManager manager(OTA_FWD_MIN_FREE);  // exactly at the discovery-shedding threshold
+  TraceTestMesh node(radio, clock, rng, rtc, manager, tables);
+  node.forwardFloods = true;
+  node.tempRadioActive = true;
+
+  mesh::Packet request = makeFloodPacket(PAYLOAD_TYPE_OTA);
+  request.payload[0] = mesh::ota::OTA_REQ;
+  ASSERT_EQ(OTA_FWD_MIN_FREE, manager.getFreeCount());
+  mesh::DispatcherAction action = node.receivePacket(&request);
+
+  EXPECT_NE(ACTION_RELEASE, action);
+  EXPECT_EQ(OTA_TRANSFER_TX_PRIORITY, (action >> 24) - 1);
+  EXPECT_EQ(1, request.getPathHashCount());
+}
+
 TEST(RepeaterTransport, OtaRelayDelayIsQuarterToHalfAirtime) {
   TraceTestClock clock;
   TraceTestRTC rtc;

tools/lora_ota/lora_ota.py

@@ -14,12 +14,14 @@ Run through ``lora_ota.sh`` or ``lora_ota.ps1``; see
 from __future__ import annotations
 
 import argparse
+import atexit
 import getpass
 import hashlib
 import json
 import math
 import os
 from pathlib import Path
+import queue
 import re
 import secrets
 import shlex
@@ -30,6 +32,7 @@ import struct
 import subprocess
 import sys
 import tempfile
+import threading
 import time
 from dataclasses import dataclass
 from datetime import datetime
@@ -52,8 +55,10 @@ MAX_FIRMWARE_IMAGE_SIZE = 64 * 1024 * 1024
 MOTA_MAX_BLOCK_SIZE = 1024
 TRANSMISSION_RETRY_LIMIT = 3
 TRANSMISSION_RETRY_WINDOW_SECONDS = 90
-TRANSMISSION_RETRY_DELAY_SECONDS = 1
+TRANSMISSION_RETRY_DELAY_SECONDS = 2
+TRANSMISSION_RETRY_MAX_DELAY_SECONDS = 8
 TRANSMISSION_PROMPT_SECONDS = 10
+ADAPTIVE_POLL_MAX_FACTOR = 3
 TEMP_RADIO_SWITCH_DELAY_SECONDS = 3
 TEMP_RADIO_RETURN_MINUTES = 1
 TEMP_RADIO_RETURN_MARGIN_SECONDS = 15
@@ -246,7 +251,7 @@ def retry_transmission(action: Callable[[], T], label: str) -> T:
                     f"[transmission] {label} failed; retry "
                     f"{retries}/{TRANSMISSION_RETRY_LIMIT}: {exc}"
                 )
-                time.sleep(TRANSMISSION_RETRY_DELAY_SECONDS)
+                time.sleep(transmission_retry_delay(retries))
                 continue
             if not prompt_after_transmission_failure(label, exc):
                 raise TransmissionStopped(
@@ -256,6 +261,33 @@ def retry_transmission(action: Callable[[], T], label: str) -> T:
             retries = 0
 
 
+def transmission_retry_delay(retry_number: int) -> int:
+    """Return a bounded exponential delay for consecutive RF failures."""
+    exponent = max(0, retry_number - 1)
+    return min(
+        TRANSMISSION_RETRY_MAX_DELAY_SECONDS,
+        TRANSMISSION_RETRY_DELAY_SECONDS * (2 ** exponent),
+    )
+
+
+def adaptive_poll_interval(
+    current_seconds: float,
+    baseline_seconds: float,
+    query_seconds: float,
+    reply_timeout: float,
+) -> float:
+    """Back status polling off after contention and recover after quick replies."""
+    ceiling = max(baseline_seconds, baseline_seconds * ADAPTIVE_POLL_MAX_FACTOR)
+    contention_threshold = max(5.0, reply_timeout * 0.5)
+    if query_seconds >= contention_threshold:
+        return min(ceiling, max(baseline_seconds, current_seconds * 1.5))
+    return max(baseline_seconds, current_seconds * 0.75)
+
+
+def adaptive_poll_ceiling(baseline_seconds: float) -> float:
+    return max(baseline_seconds, baseline_seconds * ADAPTIVE_POLL_MAX_FACTOR)
+
+
 def format_decimal(value: float) -> str:
     return f"{value:.6f}".rstrip("0").rstrip(".")
 
@@ -926,8 +958,159 @@ def extract_reply_version(reply: str) -> int | None:
     return parse_version(match.group(1)) if match else None
 
 
+class PersistentMeshcliSession:
+    """Run meshcli scripts over one long-lived Companion connection."""
+
+    def __init__(self, command: list[str]):
+        self.command = command
+        self.process: subprocess.Popen[bytes] | None = None
+        self.output_queue: queue.Queue[bytes | None] = queue.Queue()
+        self.pending = bytearray()
+        self.reader_thread: threading.Thread | None = None
+        self.announced = False
+
+    @staticmethod
+    def _read_output(
+        stream,
+        output_queue: queue.Queue[bytes | None],
+    ) -> None:
+        try:
+            while True:
+                chunk = os.read(stream.fileno(), 4096)
+                if not chunk:
+                    break
+                output_queue.put(chunk)
+        except OSError:
+            pass
+        finally:
+            output_queue.put(None)
+
+    def _start(self) -> None:
+        self.close()
+        self.output_queue = queue.Queue()
+        self.pending = bytearray()
+        try:
+            self.process = subprocess.Popen(
+                [*self.command, "-C", "-i"],
+                stdin=subprocess.PIPE,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.STDOUT,
+                bufsize=0,
+            )
+        except OSError as exc:
+            raise OtaError(
+                f"could not start persistent meshcli session: {exc}"
+            ) from exc
+        assert self.process.stdout is not None
+        self.reader_thread = threading.Thread(
+            target=self._read_output,
+            args=(self.process.stdout, self.output_queue),
+            name="meshcli-output",
+            daemon=True,
+        )
+        self.reader_thread.start()
+
+    @staticmethod
+    def _line_pattern(marker: str) -> re.Pattern[bytes]:
+        return re.compile(
+            rb"(?:^|\r?\n)" + re.escape(marker.encode("ascii")) + rb"\r?\n"
+        )
+
+    def _read_frame(self, start: str, end: str, timeout: float) -> str:
+        start_pattern = self._line_pattern(start)
+        end_pattern = self._line_pattern(end)
+        deadline = time.monotonic() + timeout
+        while True:
+            start_match = start_pattern.search(self.pending)
+            if start_match is not None:
+                end_match = end_pattern.search(self.pending, start_match.end())
+                if end_match is not None:
+                    frame = bytes(self.pending[start_match.end():end_match.start()])
+                    del self.pending[:end_match.end()]
+                    return frame.decode("utf-8", "replace")
+            elif len(self.pending) > 256 * 1024:
+                del self.pending[:-64 * 1024]
+
+            if len(self.pending) > 8 * 1024 * 1024:
+                raise OtaError("persistent meshcli output exceeded 8 MiB")
+            remaining = deadline - time.monotonic()
+            if remaining <= 0:
+                self.close()
+                raise OtaError("persistent meshcli command timed out")
+            try:
+                chunk = self.output_queue.get(timeout=min(remaining, 1.0))
+            except queue.Empty:
+                if self.process is not None and self.process.poll() is not None:
+                    detail = bytes(self.pending[-4096:]).decode("utf-8", "replace")
+                    self.close()
+                    raise OtaError(
+                        "persistent meshcli session exited"
+                        + (f": {detail.strip()}" if detail.strip() else "")
+                    )
+                continue
+            if chunk is None:
+                detail = bytes(self.pending[-4096:]).decode("utf-8", "replace")
+                self.close()
+                raise OtaError(
+                    "persistent meshcli session closed"
+                    + (f": {detail.strip()}" if detail.strip() else "")
+                )
+            self.pending.extend(chunk)
+
+    def run_script(
+        self,
+        script_path: Path,
+        start_marker: str,
+        end_marker: str,
+        timeout: float,
+    ) -> str:
+        if self.process is None or self.process.poll() is not None:
+            self._start()
+        assert self.process is not None and self.process.stdin is not None
+        command = shlex.join(["script", str(script_path)]) + "\n"
+        try:
+            self.process.stdin.write(command.encode("utf-8"))
+            self.process.stdin.flush()
+        except (BrokenPipeError, OSError) as exc:
+            self.close()
+            raise OtaError(f"persistent meshcli input failed: {exc}") from exc
+        output = self._read_frame(start_marker, end_marker, timeout)
+        if not self.announced:
+            print("[controller] persistent meshcli connection established")
+            self.announced = True
+        return output
+
+    def close(self) -> None:
+        process = self.process
+        self.process = None
+        if process is None:
+            return
+        if process.poll() is None and process.stdin is not None:
+            try:
+                process.stdin.write(b"quit\n")
+                process.stdin.flush()
+                process.wait(timeout=2)
+            except (BrokenPipeError, OSError, subprocess.TimeoutExpired):
+                process.terminate()
+                try:
+                    process.wait(timeout=2)
+                except subprocess.TimeoutExpired:
+                    process.kill()
+                    process.wait(timeout=2)
+        if process.stdout is not None:
+            process.stdout.close()
+        if process.stdin is not None:
+            process.stdin.close()
+
+
 class Controller:
-    def __init__(self, args: argparse.Namespace, password: str):
+    def __init__(
+        self,
+        args: argparse.Namespace,
+        password: str,
+        *,
+        persistent: bool = True,
+    ):
         self.meshcli = args.meshcli
         self.password = password
         self.reply_timeout = args.reply_timeout
@@ -942,6 +1125,58 @@ class Controller:
             self.connection = ["-t", host, "-p", str(port)]
         else:
             self.connection = ["-a", args.controller_ble]
+        # Remote-admin authentication belongs to the radio node, not to each
+        # host command. Reuse it across application reboots; only repeated
+        # silence suggests that the session needs to be refreshed.
+        self._authenticated_targets: set[str] = set()
+        self._authenticated_target_failures: dict[str, int] = {}
+        self._meshcli_session = (
+            PersistentMeshcliSession([
+                self.meshcli, "-j", "-c", "off", *self.connection,
+            ])
+            if persistent else None
+        )
+        if self._meshcli_session is not None:
+            atexit.register(self.close)
+
+    def close(self) -> None:
+        session = getattr(self, "_meshcli_session", None)
+        if session is not None:
+            session.close()
+
+    def forget_remote_auth(self, target: str) -> None:
+        authenticated = getattr(self, "_authenticated_targets", None)
+        if authenticated is not None:
+            authenticated.discard(target)
+        failures = getattr(self, "_authenticated_target_failures", None)
+        if failures is not None:
+            failures.pop(target, None)
+
+    def _remote_auth_is_cached(self, target: str) -> bool:
+        return target in getattr(self, "_authenticated_targets", set())
+
+    def _cache_remote_auth(self, target: str) -> None:
+        if not hasattr(self, "_authenticated_targets"):
+            self._authenticated_targets = set()
+        if not hasattr(self, "_authenticated_target_failures"):
+            self._authenticated_target_failures = {}
+        self._authenticated_targets.add(target)
+        self._authenticated_target_failures.pop(target, None)
+
+    def _note_remote_silence(self, target: str) -> bool:
+        if not hasattr(self, "_authenticated_target_failures"):
+            self._authenticated_target_failures = {}
+        failures = self._authenticated_target_failures.get(target, 0) + 1
+        self._authenticated_target_failures[target] = failures
+        if failures < 2:
+            return False
+        self.forget_remote_auth(target)
+        return True
+
+    def _note_remote_reply(self, target: str) -> None:
+        failures = getattr(self, "_authenticated_target_failures", None)
+        if failures is not None:
+            failures.pop(target, None)
 
     def _execute(
         self, commands: list[str], label: str
@@ -951,25 +1186,38 @@ class Controller:
         # us one safely quoted command line. The temporary file is removed even
         # when meshcli times out or fails.
         script_path: Path | None = None
+        frame_start = f"MESHCORE_HOST_START_{secrets.token_hex(16)}"
+        frame_end = f"MESHCORE_HOST_END_{secrets.token_hex(16)}"
         try:
             with tempfile.NamedTemporaryFile(
                 mode="w", encoding="utf-8", prefix="meshcore-ota-",
                 suffix=".meshcli", delete=False,
             ) as script:
+                if self._meshcli_session is not None:
+                    script.write(shlex.join(["echo", frame_start]))
+                    script.write("\n")
                 script.write(shlex.join(commands))
                 script.write("\n")
+                if self._meshcli_session is not None:
+                    script.write(shlex.join(["echo", frame_end]))
+                    script.write("\n")
                 script_path = Path(script.name)
             if os.name != "nt":
                 script_path.chmod(0o600)
-            command = [
-                self.meshcli, "-j", "-c", "off", *self.connection,
-                "script", str(script_path),
-            ]
-            result = run_checked(
-                command,
-                label=label,
-                timeout=max(90, self.reply_timeout + 60),
-            )
+            timeout = max(90, self.reply_timeout + 60)
+            if self._meshcli_session is not None:
+                stdout = self._meshcli_session.run_script(
+                    script_path, frame_start, frame_end, timeout
+                )
+                result = subprocess.CompletedProcess(
+                    [], 0, stdout=stdout, stderr=""
+                )
+            else:
+                command = [
+                    self.meshcli, "-j", "-c", "off", *self.connection,
+                    "script", str(script_path),
+                ]
+                result = run_checked(command, label=label, timeout=timeout)
         finally:
             if script_path is not None:
                 script_path.unlink(missing_ok=True)
@@ -1064,17 +1312,20 @@ class Controller:
         password: str,
     ) -> str:
         marker = f"MESHCORE_OTA_{secrets.token_hex(16)}"
+        login_required = not self._remote_auth_is_cached(target)
+        commands = ["contact_info", target]
+        if login_required:
+            commands.extend(["login", target, password])
+        commands.extend([
+            "sync_msgs",
+            "echo", marker,
+            "cmd", target, command_text,
+            "trywait_msg", str(self.reply_timeout),
+            "sync_msgs",
+        ])
         try:
             objects, post_objects = self._run_marked(
-                [
-                    "contact_info", target,
-                    "login", target, password,
-                    "sync_msgs",
-                    "echo", marker,
-                    "cmd", target, command_text,
-                    "trywait_msg", str(self.reply_timeout),
-                    "sync_msgs",
-                ],
+                commands,
                 f"remote command on {target}",
                 marker,
             )
@@ -1090,10 +1341,13 @@ class Controller:
         ):
             raise OtaError(f"controller has no contact named {target!r}")
         login_results = [item for item in objects if "login_success" in item]
-        if login_results and not login_results[-1].get("login_success"):
-            raise OtaError(f"admin login failed for {target}")
-        if not login_results:
-            raise TransmissionError(f"no admin-login result from {target}")
+        if login_required:
+            if login_results and not login_results[-1].get("login_success"):
+                raise OtaError(f"admin login failed for {target}")
+            if not login_results:
+                raise TransmissionError(f"no admin-login result from {target}")
+            self._cache_remote_auth(target)
+            print(f"[auth] remote admin session established for {target}")
 
         target_key = None
         for item in objects:
@@ -1115,11 +1369,17 @@ class Controller:
             and reply_matches_command(command_text, item["text"])
         ]
         if not messages:
+            refresh = self._note_remote_silence(target)
+            refresh_note = (
+                "; cached admin session will be refreshed on the next retry"
+                if refresh else ""
+            )
             raise TransmissionError(
                 f"no matching CLI reply from {target} for {command_text!r}; "
-                "check its path and reply timeout"
+                f"check its path and reply timeout{refresh_note}"
             )
         reply = messages[-1]["text"]
+        self._note_remote_reply(target)
         print(f"[{target}] {reply}")
         return reply
 
@@ -1396,9 +1656,12 @@ def verify_shared_source_identity(
         controller_baud=args.controller_baud,
         reply_timeout=args.reply_timeout,
     )
-    source_api = Controller(api_args, "")
     controller_key = controller.get_public_key()
-    source_key = source_api.get_public_key()
+    if controller.connection == ["-t", source_host, "-p", "5000"]:
+        source_key = controller_key
+    else:
+        source_api = Controller(api_args, "", persistent=False)
+        source_key = source_api.get_public_key()
     if source_key != controller_key:
         raise OtaError(
             "--source-shares-controller identity mismatch: controller is "
@@ -1730,15 +1993,29 @@ def monitor_download(
     seeder: SeederProcess | None = None,
 ) -> str:
     deadline = time.monotonic() + args.transfer_timeout_minutes * 60
+    poll_seconds = float(args.poll_seconds)
     while time.monotonic() < deadline:
+        query_started = time.monotonic()
         status = remote_command_with_seeder(
             controller, args.target, "ota status", seeder, "during transfer"
         )
+        query_seconds = time.monotonic() - query_started
         require_package_session(status, package)
         lowered = status.lower()
         if "ready to install" in lowered:
             return status
-        time.sleep(args.poll_seconds)
+        next_poll = adaptive_poll_interval(
+            poll_seconds, float(args.poll_seconds), query_seconds,
+            float(args.reply_timeout),
+        )
+        if round(next_poll) != round(poll_seconds):
+            print(
+                f"[download] adaptive status interval "
+                f"{round(poll_seconds)}s -> {round(next_poll)}s "
+                f"(query {query_seconds:.1f}s)"
+            )
+        poll_seconds = next_poll
+        time.sleep(poll_seconds)
     raise OtaError(
         "transfer timeout; the partial download remains staged and can resume when the same mOTA is served again"
     )
@@ -1881,7 +2158,7 @@ def request_install(
                     f"[install] destination is still staged; safe retry "
                     f"{retries}/{TRANSMISSION_RETRY_LIMIT}"
                 )
-                time.sleep(TRANSMISSION_RETRY_DELAY_SECONDS)
+                time.sleep(transmission_retry_delay(retries))
                 continue
             if not prompt_after_transmission_failure(
                 f"'ota install' on {args.target}", exc
@@ -2131,6 +2408,14 @@ def build_parser() -> argparse.ArgumentParser:
         metavar="HASH",
         help="16-hex running EndF body hash required by --clear-completed-manifest",
     )
+    parser.add_argument(
+        "--expected-installed-body-hash",
+        metavar="HASH",
+        help=(
+            "require this exact 16-hex EndF body hash after installation; "
+            "used when a delta does not expose its reconstructed target hash"
+        ),
+    )
     parser.add_argument("--yes", action="store_true", help="skip the destructive-action confirmation")
     parser.add_argument(
         "--require-system-watchdog-off",
@@ -2178,6 +2463,21 @@ def validate_args(args: argparse.Namespace, parser: argparse.ArgumentParser) ->
         args.clear_completed_on_body_hash = clear_body_hash.upper()
         if args.prepare_only:
             parser.error("completed-manifest cleanup is only valid during a live run")
+    expected_installed_hash = args.expected_installed_body_hash
+    if expected_installed_hash:
+        if not re.fullmatch(r"[0-9A-Fa-f]{16}", expected_installed_hash):
+            parser.error(
+                "--expected-installed-body-hash must be 16 hexadecimal characters"
+            )
+        args.expected_installed_body_hash = expected_installed_hash.upper()
+        if args.prepare_only:
+            parser.error(
+                "--expected-installed-body-hash is only valid during a live install"
+            )
+        if args.no_install:
+            parser.error(
+                "--expected-installed-body-hash cannot be used with --no-install"
+            )
     if args.prepare_only:
         if not args.platform or not args.target_id:
             parser.error("--prepare-only requires --platform and --target-id")
@@ -2247,7 +2547,7 @@ def validate_args(args: argparse.Namespace, parser: argparse.ArgumentParser) ->
             + args.seeder_start_wait
             + args.discovery_timeout
             + args.transfer_timeout_minutes * 60
-            + args.poll_seconds
+            + adaptive_poll_ceiling(args.poll_seconds)
             + args.reply_timeout * final_reply_count
         )
         temp_seconds = args.temp_values[4] * 60
@@ -2309,12 +2609,16 @@ def offline_target(args: argparse.Namespace) -> TargetInfo:
     )
 
 
-def main(argv: list[str] | None = None) -> int:
+def main(
+    argv: list[str] | None = None,
+    *,
+    controller_override: Controller | None = None,
+) -> int:
     parser = build_parser()
     args = parser.parse_args(argv)
     validate_args(args, parser)
     work_dir: Path | None = None
-    controller: Controller | None = None
+    controller: Controller | None = controller_override
     original_radio: RadioSettings | None = None
     controller_changed = False
     seeder: SeederProcess | None = None
@@ -2334,7 +2638,8 @@ def main(argv: list[str] | None = None) -> int:
         args.relay_values = [parse_relay(value, password) for value in args.relay]
         if not args.prepare_only:
             preflight_source_cli(args)
-            controller = Controller(args, password)
+            if controller is None:
+                controller = Controller(args, password)
             verify_shared_source_identity(controller, args)
             target = query_target(controller, args)
             original_radio = controller.get_radio()
@@ -2350,6 +2655,17 @@ def main(argv: list[str] | None = None) -> int:
         package_path, package, expected_body_hash = prepare_package(
             args, target, work_dir
         )
+        if args.expected_installed_body_hash:
+            asserted_body_hash = bytes.fromhex(args.expected_installed_body_hash)
+            if (
+                expected_body_hash is not None
+                and expected_body_hash != asserted_body_hash
+            ):
+                raise OtaError(
+                    "--expected-installed-body-hash conflicts with the body hash "
+                    "exposed by the selected package"
+                )
+            expected_body_hash = asserted_body_hash
         print(
             f"[package] {package_path.name}: {package.version} {package.kind} "
             f"target={package.target_id:08X} mid={package.manifest_id}"

tools/lora_ota/rak3401_mota_chain.py

@@ -100,6 +100,11 @@ KNOWN_FAILED_V11701_STEP15_IMAGE_SHA256 = (
 SAFE_CANDIDATE_STEP15_IMAGE_SHA256 = (
     "1124247f65772f11f9527408e51971eb9633ed656206276fa95019275bb8fdd2"
 )
+KNOWN_FAILED_V11701_STEP16 = 16
+KNOWN_FAILED_V11701_STEP16_VERSION = "1.16.9.112"
+KNOWN_FAILED_V11701_STEP16_IMAGE_SHA256 = (
+    "35211ea70be635376b366a90afc74fcd8a7695f744f9c21a50bc872455ec5b21"
+)
 KNOWN_UNSAFE_RELEASE_MESSAGE = (
     f"live installation of {KNOWN_UNSAFE_RELEASE_TAG} is disabled: a physical RAK3401 "
     f"test reached step {KNOWN_UNSAFE_STEP} (v{KNOWN_UNSAFE_VERSION}) but "
@@ -114,6 +119,15 @@ KNOWN_FAILED_V11701_MESSAGE = (
     "unchecked CC310 SHA path failed. Use --verify-only for artifact inspection. "
     "Do not deploy this replaced candidate."
 )
+KNOWN_FAILED_V11701_STEP16_MESSAGE = (
+    "live installation of the second v1.17.01 candidate is disabled: a "
+    "physical RAK3401 test passed steps 1-15, but step "
+    f"{KNOWN_FAILED_V11701_STEP16} (v{KNOWN_FAILED_V11701_STEP16_VERSION}) "
+    "booted without a usable EndF. Its checked CC310 SHA call can report "
+    "success while returning a wrong digest for memory-mapped application "
+    "flash, so return-code fallback is not sufficient. Use --verify-only for "
+    "artifact inspection. Do not deploy this replaced candidate."
+)
 
 
 class KnownUnsafeReleaseError(ota.OtaError):
@@ -140,12 +154,13 @@ def require_live_release_safe(
             "live installation is disabled: step 15 is not a recognized, "
             "audited RAK3401 SHA-safe bridge image"
         )
-    if not args.accept_test_candidate:
-        raise KnownUnsafeReleaseError(
-            "this corrected bundle has passed offline reconstruction and bootloader "
-            "simulation but still needs its first end-to-end board test; rerun with "
-            "--accept-test-candidate only on a recoverable lab RAK3401"
-        )
+    step16_sha256 = steps[KNOWN_FAILED_V11701_STEP16 - 1].target_sha256
+    if step16_sha256 == KNOWN_FAILED_V11701_STEP16_IMAGE_SHA256:
+        raise KnownUnsafeReleaseError(KNOWN_FAILED_V11701_STEP16_MESSAGE)
+    raise KnownUnsafeReleaseError(
+        "live installation is disabled: step 16 is not a recognized, "
+        "physically tested software-SHA bridge image"
+    )
 
 
 @dataclass(frozen=True)
@@ -649,6 +664,20 @@ def enforce_ota_hops(
     print(f"[radio] destination OTA reach verified at {current} hops")
 
 
+def require_rescue_capability(
+    controller: ota.Controller,
+    target_name: str,
+) -> None:
+    """Require the guarded no-EndF recovery command before another transition."""
+    reply = controller.remote_command(target_name, "ota help")
+    if "rescue install <hash16>" not in reply:
+        raise ota.OtaError(
+            "running bridge does not advertise `ota rescue install <hash16>`; "
+            "refusing to expose it to another chain transition"
+        )
+    print("[rescue] guarded no-EndF recovery command is present")
+
+
 def wait_with_label(seconds: int, label: str) -> None:
     deadline = time.monotonic() + seconds
     while True:
@@ -781,7 +810,9 @@ def run_step(
     target_name: str,
     step: ChainStep,
     previous_step: ChainStep | None,
+    expected_body_hash: bytes,
     work_dir: Path,
+    controller: ota.Controller,
 ) -> None:
     command = [
         str(step.path),
@@ -801,6 +832,7 @@ def run_step(
         "--reboot-wait", str(args.reboot_wait),
         "--work-dir", str(next_attempt_dir(work_dir, step.number)),
         "--require-system-watchdog-off",
+        "--expected-installed-body-hash", expected_body_hash.hex().upper(),
         # Some audited bridge binaries retain a later historical runtime
         # version string than their pinned EndF chain version. The exact base
         # hash, target hash, package ID, and ordered step still gate the move.
@@ -814,7 +846,7 @@ def run_step(
         ])
     for relay in args.relay:
         command.extend(["--relay", relay])
-    result = ota.main(command)
+    result = ota.main(command, controller_override=controller)
     if result != 0:
         raise ota.OtaError(f"chain step {step.number} exited with status {result}")
 
@@ -842,8 +874,8 @@ def build_parser() -> argparse.ArgumentParser:
         "--accept-test-candidate",
         action="store_true",
         help=(
-            "permit the corrected, offline-validated chain on a recoverable lab "
-            "RAK3401 before its first complete physical test"
+            "reserved lab gate for a future pinned candidate; it never overrides "
+            "a bundle known to have failed physical testing"
         ),
     )
 
@@ -951,7 +983,7 @@ def validate_args(args: argparse.Namespace, parser: argparse.ArgumentParser) ->
         + args.seeder_start_wait
         + args.discovery_timeout
         + args.transfer_timeout_minutes * 60
-        + args.poll_seconds
+        + ota.adaptive_poll_ceiling(args.poll_seconds)
         + args.reply_timeout * (4 + len(args.relay))
     )
     if not args.verify_only and temp_values[4] * 60 <= required_seconds:
@@ -1018,10 +1050,18 @@ def main(argv: list[str] | None = None) -> int:
                 == KNOWN_FAILED_V11701_STEP15_IMAGE_SHA256
             ):
                 print(f"WARNING: {KNOWN_FAILED_V11701_MESSAGE}", file=sys.stderr)
+            elif (
+                steps[KNOWN_FAILED_V11701_STEP16 - 1].target_sha256
+                == KNOWN_FAILED_V11701_STEP16_IMAGE_SHA256
+            ):
+                print(
+                    f"WARNING: {KNOWN_FAILED_V11701_STEP16_MESSAGE}",
+                    file=sys.stderr,
+                )
             else:
                 print(
-                    "Corrected test candidate verified offline; a complete physical-board "
-                    "run is still required before production use."
+                    "Unfailed candidate verified offline; complete physical-board testing "
+                    "is still required before production use."
                 )
             return 0
 
@@ -1083,10 +1123,13 @@ def main(argv: list[str] | None = None) -> int:
                 f"{step.from_version} -> {step.to_version}"
             )
             previous_step = steps[index - 1] if index > 0 else None
-            run_step(args, target_name, step, previous_step, work_dir)
+            expected_hash = expected_hash_after(steps, final_body_hash, index)
+            run_step(
+                args, target_name, step, previous_step, expected_hash,
+                work_dir, controller
+            )
 
             target = query_live_target(controller, args, target_name)
-            expected_hash = expected_hash_after(steps, final_body_hash, index)
             if target.base_hash != expected_hash:
                 raise ota.OtaError(
                     f"step {step.number} returned body hash {target.base_hash.hex().upper()}, "
@@ -1111,6 +1154,7 @@ def main(argv: list[str] | None = None) -> int:
                     f"runtime label {target.current_version} is historical and is "
                     f"not the EndF chain version {step.to_version}"
                 )
+            require_rescue_capability(controller, target_name)
             clear_completed_download(controller, target_name, step)
             require_watchdog_state(controller, target_name, "off")
             append_progress(work_dir, step, target.base_hash)

tools/lora_ota/test_lora_ota.py

@@ -170,6 +170,31 @@ class FormatTests(unittest.TestCase):
         with self.assertRaises(SystemExit), contextlib.redirect_stderr(io.StringIO()):
             ota.validate_args(args, parser)
 
+    def test_expected_installed_body_hash_is_normalized(self) -> None:
+        parser = ota.build_parser()
+        args = parser.parse_args([
+            "release.mota", "remote",
+            "--controller-serial", "/dev/controller",
+            "--source-serial", "/dev/source",
+            "--expected-installed-body-hash", "aabbccddeeff0011",
+        ])
+        ota.validate_args(args, parser)
+        self.assertEqual(
+            args.expected_installed_body_hash, "AABBCCDDEEFF0011"
+        )
+
+    def test_expected_installed_body_hash_rejects_stage_only(self) -> None:
+        parser = ota.build_parser()
+        args = parser.parse_args([
+            "release.mota", "remote",
+            "--controller-serial", "/dev/controller",
+            "--source-serial", "/dev/source",
+            "--expected-installed-body-hash", "AABBCCDDEEFF0011",
+            "--no-install",
+        ])
+        with self.assertRaises(SystemExit), contextlib.redirect_stderr(io.StringIO()):
+            ota.validate_args(args, parser)
+
     def test_intel_hex_rejects_an_excessive_address_span(self) -> None:
         def record(address: int, record_type: int, data: bytes) -> str:
             raw = bytes((len(data), address >> 8, address & 0xFF, record_type)) + data
@@ -434,7 +459,7 @@ class DownloadSessionTests(unittest.TestCase):
             "OK dropped session",
             "OTA | no download",
             "Updates 1/1",
-            f"OK pulling mid={self.package.manifest_id} -> flash (low priority)",
+            f"OK pulling mid={self.package.manifest_id} -> flash (primary traffic)",
         ])
         ota.find_and_start_pull(controller, args, self.package)
         self.assertEqual(
@@ -450,7 +475,7 @@ class DownloadSessionTests(unittest.TestCase):
             "OTA | download: downloading 3/9 id=DEADBEEF 2s",
             "OK dropped session",
             "Updates 1/1",
-            f"OK pulling mid={self.package.manifest_id} -> flash (low priority)",
+            f"OK pulling mid={self.package.manifest_id} -> flash (primary traffic)",
         ])
         ota.find_and_start_pull(controller, self.args(replace=True), self.package)
         self.assertEqual(
@@ -599,6 +624,21 @@ class ReliabilityTests(unittest.TestCase):
             ota.retry_transmission(action, "test command")
         self.assertEqual(calls, 4)
 
+    def test_retry_delay_is_bounded_exponential(self) -> None:
+        self.assertEqual(
+            [ota.transmission_retry_delay(number) for number in range(1, 6)],
+            [2, 4, 8, 8, 8],
+        )
+
+    def test_status_poll_interval_adapts_to_contention(self) -> None:
+        backed_off = ota.adaptive_poll_interval(60, 60, 30, 45)
+        self.assertEqual(backed_off, 90)
+        self.assertEqual(
+            ota.adaptive_poll_interval(backed_off, 60, 2, 45),
+            67.5,
+        )
+        self.assertEqual(ota.adaptive_poll_ceiling(60), 180)
+
     def test_marked_output_excludes_queued_messages(self) -> None:
         controller = object.__new__(ota.Controller)
         controller._execute = lambda _commands, _label: subprocess.CompletedProcess(
@@ -640,6 +680,67 @@ class ReliabilityTests(unittest.TestCase):
         reply = controller._remote_command_once("remote", "ota status", "secret")
         self.assertTrue(reply.startswith("OTA |"))
 
+    def test_remote_admin_login_is_reused_until_explicit_refresh(self) -> None:
+        controller = object.__new__(ota.Controller)
+        controller.reply_timeout = 20
+        controller._authenticated_targets = set()
+        controller._authenticated_target_failures = {}
+        key = "A1" * 32
+        commands_seen: list[list[str]] = []
+
+        def run_marked(commands: list[str], _label: str, _marker: str):
+            commands_seen.append(commands)
+            objects = [{"adv_name": "remote", "public_key": key}]
+            if "login" in commands:
+                objects.append({"login_success": True})
+            return objects, [{
+                "txt_type": 1,
+                "text": "OTA | no download | target:1234ABCD",
+                "pubkey_prefix": key[:12],
+            }]
+
+        controller._run_marked = run_marked
+        controller._remote_command_once("remote", "ota status", "secret")
+        controller._remote_command_once("remote", "ota status", "secret")
+        self.assertIn("login", commands_seen[0])
+        self.assertNotIn("login", commands_seen[1])
+
+        controller.forget_remote_auth("remote")
+        controller._remote_command_once("remote", "ota status", "secret")
+        self.assertIn("login", commands_seen[2])
+
+    def test_two_silent_commands_refresh_cached_admin_session(self) -> None:
+        controller = object.__new__(ota.Controller)
+        controller.reply_timeout = 20
+        controller._authenticated_targets = {"remote"}
+        controller._authenticated_target_failures = {}
+        key = "A1" * 32
+        commands_seen: list[list[str]] = []
+
+        def run_marked(commands: list[str], _label: str, _marker: str):
+            commands_seen.append(commands)
+            objects = [{"adv_name": "remote", "public_key": key}]
+            if "login" in commands:
+                objects.append({"login_success": True})
+                replies = [{
+                    "txt_type": 1,
+                    "text": "OTA | no download | target:1234ABCD",
+                    "pubkey_prefix": key[:12],
+                }]
+            else:
+                replies = []
+            return objects, replies
+
+        controller._run_marked = run_marked
+        with self.assertRaises(ota.TransmissionError):
+            controller._remote_command_once("remote", "ota status", "secret")
+        with self.assertRaises(ota.TransmissionError):
+            controller._remote_command_once("remote", "ota status", "secret")
+        controller._remote_command_once("remote", "ota status", "secret")
+        self.assertNotIn("login", commands_seen[0])
+        self.assertNotIn("login", commands_seen[1])
+        self.assertIn("login", commands_seen[2])
+
     def test_generic_retry_rejects_state_changing_ota_commands(self) -> None:
         controller = object.__new__(ota.Controller)
         controller.password = "secret"
@@ -1065,6 +1166,18 @@ class Rak3401KnownUnsafeReleaseTests(unittest.TestCase):
         with self.assertRaisesRegex(ota.OtaError, "retained mOTA DEADBEEF"):
             rak_chain.clear_completed_download(Controller(), "remote", step)
 
+    def test_chain_requires_rescue_command_before_another_transition(self) -> None:
+        controller = mock.Mock()
+        controller.remote_command.return_value = (
+            "OTA: status | install | rescue install <hash16> | cancel"
+        )
+        rak_chain.require_rescue_capability(controller, "remote")
+        controller.remote_command.assert_called_once_with("remote", "ota help")
+
+        controller.remote_command.return_value = "OTA: status | install | cancel"
+        with self.assertRaisesRegex(ota.OtaError, "refusing to expose"):
+            rak_chain.require_rescue_capability(controller, "remote")
+
     def test_live_chain_is_blocked_after_failed_physical_step_6(self) -> None:
         self.assertEqual(rak_chain.KNOWN_UNSAFE_STEP, 6)
         self.assertEqual(rak_chain.KNOWN_UNSAFE_VERSION, "1.16.8.7")
@@ -1079,19 +1192,41 @@ class Rak3401KnownUnsafeReleaseTests(unittest.TestCase):
             )
 
     def test_corrected_chain_requires_explicit_lab_gate(self) -> None:
-        steps = [mock.Mock(target_sha256="") for _ in range(15)]
+        steps = [mock.Mock(target_sha256="") for _ in range(16)]
         steps[5].target_sha256 = rak_chain.SAFE_CANDIDATE_STEP6_IMAGE_SHA256
         steps[14].target_sha256 = rak_chain.SAFE_CANDIDATE_STEP15_IMAGE_SHA256
+        steps[15].target_sha256 = (
+            rak_chain.KNOWN_FAILED_V11701_STEP16_IMAGE_SHA256
+        )
         with self.assertRaisesRegex(
             rak_chain.KnownUnsafeReleaseError,
-            "needs its first end-to-end board test",
+            "physical RAK3401 test passed steps 1-15",
         ):
             rak_chain.require_live_release_safe(
                 argparse.Namespace(accept_test_candidate=False), steps
             )
-        rak_chain.require_live_release_safe(
-            argparse.Namespace(accept_test_candidate=True), steps
+        with self.assertRaisesRegex(
+            rak_chain.KnownUnsafeReleaseError,
+            "physical RAK3401 test passed steps 1-15",
+        ):
+            rak_chain.require_live_release_safe(
+                argparse.Namespace(accept_test_candidate=True), steps
+            )
+
+    def test_checked_cc310_step16_is_always_blocked(self) -> None:
+        steps = [mock.Mock(target_sha256="") for _ in range(16)]
+        steps[5].target_sha256 = rak_chain.SAFE_CANDIDATE_STEP6_IMAGE_SHA256
+        steps[14].target_sha256 = rak_chain.SAFE_CANDIDATE_STEP15_IMAGE_SHA256
+        steps[15].target_sha256 = (
+            rak_chain.KNOWN_FAILED_V11701_STEP16_IMAGE_SHA256
         )
+        with self.assertRaisesRegex(
+            rak_chain.KnownUnsafeReleaseError,
+            "return-code fallback is not sufficient",
+        ):
+            rak_chain.require_live_release_safe(
+                argparse.Namespace(accept_test_candidate=True), steps
+            )
 
     def test_first_v11701_candidate_is_blocked_after_failed_step15(self) -> None:
         steps = [mock.Mock(target_sha256="") for _ in range(15)]

variants/rak3401/platformio.ini

@@ -64,6 +64,7 @@ build_flags =
   -D ENABLE_OTA=1
   -D OTA_FLASH_STORE=1
   -D OTA_FOLDER_SERIAL
+  -D OTA_FETCH_PIPELINE=4                 ; adapt from 2 to 4 concurrent block requests on a clean OTA link
 build_src_filter = ${rak3401.build_src_filter}
   +<helpers/ota/*.cpp>
   +<helpers/ui/SSD1306Display.cpp>