diff --git a/feature/firmware/src/commonMain/kotlin/org/meshtastic/feature/firmware/ota/dfu/DfuUploadTransport.kt b/feature/firmware/src/commonMain/kotlin/org/meshtastic/feature/firmware/ota/dfu/DfuUploadTransport.kt
index a772165a2..9b8d38172 100644
--- a/feature/firmware/src/commonMain/kotlin/org/meshtastic/feature/firmware/ota/dfu/DfuUploadTransport.kt
+++ b/feature/firmware/src/commonMain/kotlin/org/meshtastic/feature/firmware/ota/dfu/DfuUploadTransport.kt
@@ -43,7 +43,10 @@ interface DfuUploadTransport {
      */
     suspend fun transferFirmware(firmware: ByteArray, onProgress: suspend (Float) -> Unit): Result<Unit>
 
-    /** Best-effort abort of any in-flight transfer (for cancellation / error recovery). Never throws. */
+    /**
+     * Best-effort abort. Operational transport exceptions are swallowed; structured-concurrency cancellation and Error
+     * subtypes propagate.
+     */
     suspend fun abort()
 
     /** Disconnect and release resources. */
diff --git a/feature/firmware/src/commonMain/kotlin/org/meshtastic/feature/firmware/ota/dfu/LegacyDfuProtocol.kt b/feature/firmware/src/commonMain/kotlin/org/meshtastic/feature/firmware/ota/dfu/LegacyDfuProtocol.kt
index a58d4a3de..06ad9fbd3 100644
--- a/feature/firmware/src/commonMain/kotlin/org/meshtastic/feature/firmware/ota/dfu/LegacyDfuProtocol.kt
+++ b/feature/firmware/src/commonMain/kotlin/org/meshtastic/feature/firmware/ota/dfu/LegacyDfuProtocol.kt
@@ -166,6 +166,43 @@ internal fun legacyPrnRequestPayload(packets: Int): ByteArray = byteArrayOf(
     ((packets ushr 8) and 0xFF).toByte(),
 )
 
+// ---------------------------------------------------------------------------
+// Stream profiles
+// ---------------------------------------------------------------------------
+
+/**
+ * Default packet-receipt-notification interval (packets between flow-control ACKs). Higher values mean fewer
+ * notification round-trips per byte and therefore faster throughput, at the cost of a slightly longer recovery window
+ * if a packet is dropped.
+ *
+ * Capped at 10 to match Nordic's own Legacy DFU implementation, which force-limits legacy PRN to ≤10 with the comment:
+ * "DFU bootloaders from SDK 6.0.0 or older were unable to save incoming data to flash as fast as they are being sent …
+ * PRN = 10 may be the highest supported value" and treats status 6 (OPERATION_FAILED) as "data sent too fast — reduce
+ * PRN to 10 or less." The stock Adafruit bootloader shares this SDK11 flash-write path, so a higher value risks
+ * OPERATION_FAILED mid-stream on stock bootloaders. 10 is the safe ceiling that still batches flow-control ACKs.
+ */
+internal const val PRN_INTERVAL_PACKETS = 10
+
+/**
+ * Tighter PRN interval used only on a recovery upload that follows a mid-stream disconnect. Halving the interval
+ * reduces the burst depth between flow-control checkpoints. Used exclusively by [LegacyDfuStreamProfile.RECOVERY] as a
+ * recovery heuristic; it does not by itself prevent a bootloader-side stall.
+ */
+internal const val RECOVERY_PRN_INTERVAL_PACKETS = 5
+
+/**
+ * Legacy upload profile. Selects the PRN interval used in BOTH the `PACKET_RECEIPT_NOTIF_REQ` request and the
+ * receipt-await threshold inside [org.meshtastic.feature.firmware.ota.dfu.LegacyDfuTransport.streamFirmware].
+ * - [NORMAL]: the healthy first-attempt transfer profile — fewer PRN round-trips, higher throughput.
+ * - [RECOVERY]: selected after a [LegacyDfuException.MidStreamDisconnect] so subsequent attempts on the same run use
+ *   the tighter interval. A [LegacyDfuException.StaleSessionReset] alone does NOT switch the profile, since the link
+ *   did not actually drop mid-upload.
+ */
+internal enum class LegacyDfuStreamProfile(val prnIntervalPackets: Int) {
+    NORMAL(PRN_INTERVAL_PACKETS),
+    RECOVERY(RECOVERY_PRN_INTERVAL_PACKETS),
+}
+
 // ---------------------------------------------------------------------------
 // Exceptions
 // ---------------------------------------------------------------------------
@@ -200,12 +237,36 @@ sealed class LegacyDfuException(message: String, cause: Throwable? = null) : Dfu
         LegacyDfuException("Packet receipt mismatch: expected $expected bytes received, device reports $actual")
 
     /**
-     * START_DFU was rejected with `INVALID_STATE` because the bootloader still holds a previous, interrupted DFU
-     * session. The transport has issued a RESET; retrying the whole session (fresh reconnect) will start over on the
-     * now-clean bootloader. Retryable — distinct from a hard [ProtocolError].
+     * START_DFU was rejected with INVALID_STATE because the bootloader still holds state from an interrupted session.
+     * The current connection is closed without attempting RESET; SecureDfuHandler performs a bounded reset-prime over a
+     * fresh connection before retrying.
      */
     class StaleSessionReset :
         LegacyDfuException(
             "Bootloader rejected START with INVALID_STATE (leftover from an interrupted flash); reset and retrying.",
         )
+
+    /**
+     * The BLE link dropped while firmware bytes were in flight. Carries the host in-flight offset (the end of the write
+     * the host had dispatched or was in flight at the moment of the drop) so the retry coordinator can distinguish a
+     * mid-stream failure (which switches subsequent Legacy uploads to [LegacyDfuStreamProfile.RECOVERY]) from a
+     * handshake/connect failure (which does not).
+     *
+     * [bytesSent] is the end offset of the host write currently dispatched or in flight. It may include the current
+     * chunk when the disconnect occurs during `service.write()`. It does NOT prove the host stack accepted the chunk,
+     * and it does NOT prove the bootloader received or committed it. [lastConfirmedBytes] is the authoritative
+     * PRN-confirmed checkpoint (the last byte the bootloader explicitly acknowledged); `-1` means no PRN was received
+     * before the drop. Do not treat [bytesSent] as confirmed device progress.
+     *
+     * Typed — callers must NOT parse the message to detect a mid-stream drop.
+     */
+    class MidStreamDisconnect(
+        val bytesSent: Int,
+        val totalBytes: Int,
+        val connectionState: String,
+        val lastConfirmedBytes: Int,
+    ) : LegacyDfuException(
+        "BLE link dropped mid-upload at host in-flight offset $bytesSent/$totalBytes " +
+            "(lastConfirmed=$lastConfirmedBytes, state=$connectionState)",
+    )
 }
diff --git a/feature/firmware/src/commonMain/kotlin/org/meshtastic/feature/firmware/ota/dfu/LegacyDfuTransport.kt b/feature/firmware/src/commonMain/kotlin/org/meshtastic/feature/firmware/ota/dfu/LegacyDfuTransport.kt
index 5c11f11dc..0886854ca 100644
--- a/feature/firmware/src/commonMain/kotlin/org/meshtastic/feature/firmware/ota/dfu/LegacyDfuTransport.kt
+++ b/feature/firmware/src/commonMain/kotlin/org/meshtastic/feature/firmware/ota/dfu/LegacyDfuTransport.kt
@@ -34,11 +34,14 @@ import kotlinx.coroutines.SupervisorJob
 import kotlinx.coroutines.TimeoutCancellationException
 import kotlinx.coroutines.cancel
 import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.currentCoroutineContext
 import kotlinx.coroutines.delay
+import kotlinx.coroutines.ensureActive
 import kotlinx.coroutines.flow.catch
 import kotlinx.coroutines.flow.launchIn
 import kotlinx.coroutines.flow.onEach
 import kotlinx.coroutines.withTimeout
+import kotlinx.coroutines.withTimeoutOrNull
 import org.meshtastic.core.ble.BleConnectionFactory
 import org.meshtastic.core.ble.BleConnectionState
 import org.meshtastic.core.ble.BleDevice
@@ -48,10 +51,12 @@ import org.meshtastic.core.common.util.safeCatching
 import org.meshtastic.feature.firmware.ota.calculateMacPlusOne
 import org.meshtastic.feature.firmware.ota.scanForBleDevice
 import org.meshtastic.feature.firmware.ota.withDisconnectTripwire
+import kotlin.concurrent.Volatile
 import kotlin.time.Duration
 import kotlin.time.Duration.Companion.milliseconds
 import kotlin.time.Duration.Companion.minutes
 import kotlin.time.Duration.Companion.seconds
+import kotlin.time.TimeSource
 
 /**
  * Kable-based transport for the Nordic **Legacy DFU** protocol (Nordic SDK 11/12 / Adafruit `BLEDfu`).
@@ -66,21 +71,41 @@ import kotlin.time.Duration.Companion.seconds
  *
  * Phase-1 buttonless trigger is shared with [SecureDfuTransport] (see `triggerButtonlessDfu` there).
  */
-class LegacyDfuTransport(
+class LegacyDfuTransport
+internal constructor(
     private val scanner: BleScanner,
     connectionFactory: BleConnectionFactory,
     private val address: String,
     dispatcher: CoroutineDispatcher,
+    private val streamProfile: LegacyDfuStreamProfile = LegacyDfuStreamProfile.NORMAL,
 ) : DfuUploadTransport {
     private val transportScope = CoroutineScope(SupervisorJob() + dispatcher)
     private val bleConnection = connectionFactory.create(transportScope, "Legacy DFU")
 
+    /**
+     * Stream progress captured by the disconnect-tripwire callback. The disconnect watcher runs in a child coroutine on
+     * the caller's dispatcher, which is IO in production but may differ in tests. These fields provide visible
+     * diagnostic snapshots between the streaming and watcher coroutines. A fresh transport is created per upload
+     * attempt, so these never leak across sessions.
+     */
+    @Volatile private var streamOffset: Int = 0
+
+    @Volatile private var streamLastPrnOffset: Int = -1
+
+    @Volatile private var streamLastPrnLatencyMs: Long = -1
+
     /** Receives parsed responses from the Control Point characteristic. */
     private val notificationChannel = Channel<LegacyDfuResponse>(Channel.UNLIMITED)
 
     /** Name advertised by the device in DFU mode (e.g. `4631_DFU`). Captured in [connectToDfuMode]. */
     private var dfuAdvertisedName: String? = null
 
+    /**
+     * DFU Version reported by the optional DFU Version characteristic during [connectToDfuMode]. `-1` when the
+     * characteristic is absent or unreadable. Captured here so the stream-start log can include it without re-reading.
+     */
+    private var dfuVersion: Int = -1
+
     // ---------------------------------------------------------------------------
     // Phase 2: Connect to device in DFU mode
     // ---------------------------------------------------------------------------
@@ -144,6 +169,7 @@ class LegacyDfuTransport(
                         if (bytes.size >= 2) (bytes[0].toInt() and 0xFF) or ((bytes[1].toInt() and 0xFF) shl 8) else -1
                     }
                     .getOrElse { -1 }
+            dfuVersion = version
             Logger.i { "Legacy DFU: DFU Version characteristic = $version (-1 ⇒ absent / unreadable)" }
             if (version in 1..MIN_SUPPORTED_DFU_VERSION - 1) {
                 throw LegacyDfuException.UnsupportedBootloader(version)
@@ -204,7 +230,9 @@ class LegacyDfuTransport(
      * 7. Stream firmware in MTU-sized chunks. Every PRN packets, await `PacketReceipt(bytesReceived)` and verify count.
      * 8. After last byte, await final response for `RECEIVE_FIRMWARE_IMAGE`.
      * 9. `VALIDATE`, await response.
-     * 10. `ACTIVATE_AND_RESET` — device reboots; write may fail with disconnect, treat as success.
+     * 10. `ACTIVATE_AND_RESET` — an ordinary disconnect/write Exception may occur because the device resets before the
+     *     acknowledgement; treat that operational Exception as expected success (structured cancellation and Error
+     *     subtypes still propagate).
      */
     @Suppress("LongMethod")
     override suspend fun transferFirmware(firmware: ByteArray, onProgress: suspend (Float) -> Unit): Result<Unit> =
@@ -232,7 +260,7 @@ class LegacyDfuTransport(
             Logger.i { "Legacy DFU: requestHighConnectionPriority -> $highPriorityRequested" }
 
             // ── 3. PRN setup ────────────────────────────────────────────────────
-            writeControlPoint(legacyPrnRequestPayload(PRN_INTERVAL_PACKETS))
+            writeControlPoint(legacyPrnRequestPayload(streamProfile.prnIntervalPackets))
 
             // ── 4. RECEIVE_FIRMWARE_IMAGE ──────────────────────────────────────
             writeControlPoint(byteArrayOf(LegacyDfuOpcode.RECEIVE_FIRMWARE_IMAGE))
@@ -248,24 +276,17 @@ class LegacyDfuTransport(
             requireSuccess(LegacyDfuOpcode.VALIDATE, awaitResponse(VALIDATE_TIMEOUT))
 
             // ── 8. ACTIVATE_AND_RESET ──────────────────────────────────────────
-            // Device may reset before the GATT write ACK lands — treat any write failure / disconnect as success.
+            // The device may reset before the GATT write ACK lands; an ordinary disconnect/write Exception is expected
+            // because of that reset — safeCatching treats it as success. Structured cancellation and Error subtypes
+            // still propagate.
             Logger.i { "Legacy DFU: Sending ACTIVATE_AND_RESET (disconnect during write is expected)" }
-            runCatching { writeControlPoint(byteArrayOf(LegacyDfuOpcode.ACTIVATE_AND_RESET)) }
+            safeCatching { writeControlPoint(byteArrayOf(LegacyDfuOpcode.ACTIVATE_AND_RESET)) }
                 .onFailure { Logger.i(it) { "Legacy DFU: ACTIVATE write reported failure (expected on reset)" } }
 
             onProgress(1f)
             Logger.i { "Legacy DFU: Upload complete, device rebooting into new firmware." }
         }
 
-    /**
-     * Stream [firmware] to the Packet characteristic, awaiting a [LegacyDfuResponse.PacketReceipt] every
-     * [PRN_INTERVAL_PACKETS] packets and verifying the bytes-received count.
-     *
-     * Watches the connection state in parallel with the write loop; if the link drops mid-stream we cancel the write
-     * coroutine and surface a [DfuException.ConnectionFailed] immediately rather than waiting indefinitely for a write
-     * that will never complete.
-     */
-
     /**
      * Low-speed when the bootloader did not negotiate a larger MTU, leaving us on the 20-byte packet floor. Valid once
      * [connectToDfuMode] has established the connection (the MTU is known by then).
@@ -291,66 +312,113 @@ class LegacyDfuTransport(
         return sized - (sized % DFU_PACKET_WORD_ALIGNMENT)
     }
 
+    /**
+     * Stream [firmware] to the Packet characteristic under a single outer disconnect tripwire. The outer watcher is the
+     * sole streaming disconnect classifier — PRN waits inside the loop intentionally do NOT install another tripwire,
+     * so a link drop during a PRN wait surfaces as a typed [LegacyDfuException.MidStreamDisconnect] carrying host-side
+     * and confirmed-progress diagnostics, not a generic handshake-style [DfuException.ConnectionFailed].
+     *
+     * Every [streamProfile.prnIntervalPackets] packets the loop awaits a [LegacyDfuResponse.PacketReceipt] and verifies
+     * the bootloader's bytes-received count. The [streamOffset], [streamLastPrnOffset], and [streamLastPrnLatencyMs]
+     * snapshots give the watcher's onDrop callback visible diagnostic values.
+     */
     @Suppress("CyclomaticComplexMethod", "NestedBlockDepth", "LongMethod")
     private suspend fun streamFirmware(firmware: ByteArray, onProgress: suspend (Float) -> Unit) {
         // Packet size = negotiated ATT MTU − 3, word-aligned and capped at 244 (see computeStreamPacketSize). Falls
         // back to 20 bytes when the bootloader did not negotiate a larger MTU, which is the self-gating safety against
         // bootloaders that can't accept large DFU writes — the 20-byte path is the slow but universally-safe default.
         val mtu = computeStreamPacketSize()
-        var offset = 0
+        val prnInterval = streamProfile.prnIntervalPackets
+        // Reset stream progress for this attempt. These are @Volatile instance properties so the disconnect-tripwire
+        // callback (running in a child coroutine on the caller's dispatcher) has a visible snapshot.
+        streamOffset = 0
+        streamLastPrnOffset = -1
+        streamLastPrnLatencyMs = -1
         Logger.i {
             "Legacy DFU: Streaming ${firmware.size} bytes with packet size $mtu " +
-                "(advertised='${dfuAdvertisedName ?: "?"}')"
+                "(advertised='${dfuAdvertisedName ?: "?"}', dfuVersion=$dfuVersion, " +
+                "profile=$streamProfile, prnInterval=$prnInterval)"
         }
 
         bleConnection.withDisconnectTripwire(
             onDrop = { state ->
-                Logger.w { "Legacy DFU: Link dropped mid-stream at offset $offset/${firmware.size} (state=$state)" }
-                DfuException.ConnectionFailed("BLE link dropped mid-upload at byte $offset/${firmware.size}")
+                Logger.w {
+                    "Legacy DFU: Link dropped mid-stream at host in-flight offset $streamOffset/${firmware.size} " +
+                        "(state=$state, lastConfirmedPrn=$streamLastPrnOffset, lastPrnLatencyMs=$streamLastPrnLatencyMs)"
+                }
+                LegacyDfuException.MidStreamDisconnect(
+                    bytesSent = streamOffset,
+                    totalBytes = firmware.size,
+                    connectionState = state.toString(),
+                    lastConfirmedBytes = streamLastPrnOffset,
+                )
             },
         ) {
             var packetsSincePrn = 0
             var bytesAtLastPrn = 0L
             bleConnection.profile(LegacyDfuUuids.SERVICE, timeout = STREAM_TIMEOUT) { service ->
                 val packetChar = service.characteristic(LEGACY_DFU_PACKET_UUID)
-                while (offset < firmware.size) {
-                    val end = minOf(offset + mtu, firmware.size)
+                while (streamOffset < firmware.size) {
+                    val end = minOf(streamOffset + mtu, firmware.size)
+                    // Publish the in-flight boundary BEFORE invoking service.write() so the disconnect watcher
+                    // observes the current attempted chunk when a disconnect fires synchronously inside the write
+                    // (e.g. a test harness or a real bootloader that drops on receive). The write may not complete
+                    // and the host stack may not accept it; this boundary is the dispatched boundary, not a
+                    // confirmed one — the authoritative checkpoint is the PRN-confirmed offset (streamLastPrnOffset).
+                    val chunk = firmware.copyOfRange(streamOffset, end)
+                    streamOffset = end
                     try {
-                        service.write(packetChar, firmware.copyOfRange(offset, end), BleWriteType.WITHOUT_RESPONSE)
+                        service.write(packetChar, chunk, BleWriteType.WITHOUT_RESPONSE)
                     } catch (e: CancellationException) {
                         Logger.w(e) {
-                            "Legacy DFU: Write CANCELLED at offset $offset/${firmware.size} cause=${e.cause}"
+                            "Legacy DFU: Write CANCELLED at offset $streamOffset/${firmware.size} cause=${e.cause}"
                         }
                         throw e
                     } catch (@Suppress("TooGenericExceptionCaught") e: Throwable) {
-                        Logger.w(e) { "Legacy DFU: Write FAILED at offset $offset/${firmware.size}: ${e.message}" }
+                        Logger.w(e) {
+                            "Legacy DFU: Write FAILED at offset $streamOffset/${firmware.size}: ${e.message}"
+                        }
                         throw e
                     }
-                    offset = end
                     packetsSincePrn++
 
-                    if (packetsSincePrn >= PRN_INTERVAL_PACKETS && offset < firmware.size) {
-                        Logger.d { "Legacy DFU: Awaiting PRN at offset $offset" }
+                    if (packetsSincePrn >= prnInterval && streamOffset < firmware.size) {
+                        val awaitMark = TimeSource.Monotonic.markNow()
+                        Logger.d { "Legacy DFU: Awaiting PRN at offset $streamOffset" }
                         val receipt =
                             try {
-                                awaitPacketReceipt()
+                                awaitPacketReceiptDuringStream()
                             } catch (e: CancellationException) {
                                 Logger.w(e) {
-                                    "Legacy DFU: awaitPacketReceipt CANCELLED at offset $offset cause=${e.cause}"
+                                    "Legacy DFU: awaitPacketReceiptDuringStream CANCELLED at offset $streamOffset " +
+                                        "cause=${e.cause}"
                                 }
                                 throw e
                             }
-                        val expected = offset.toLong()
+                        val latencyMs = awaitMark.elapsedNow().inWholeMilliseconds
+                        if (latencyMs >= PRN_LATENCY_WARN_THRESHOLD_MS) {
+                            Logger.w {
+                                "Legacy DFU: PRN receipt latency ${latencyMs}ms at offset $streamOffset " +
+                                    "(>= ${PRN_LATENCY_WARN_THRESHOLD_MS}ms — possible bootloader backpressure or BLE scheduling delay)"
+                            }
+                        } else {
+                            Logger.d { "Legacy DFU: PRN receipt at offset $streamOffset latency=${latencyMs}ms" }
+                        }
+                        val expected = streamOffset.toLong()
                         if (receipt.bytesReceived != expected) {
                             throw LegacyDfuException.PacketReceiptMismatch(expected, receipt.bytesReceived)
                         }
+                        // Record the checkpoint only after the receipt validates — a mismatched PRN must
+                        // not be reported as the last successful checkpoint in the link-drop log.
+                        streamLastPrnOffset = streamOffset
+                        streamLastPrnLatencyMs = latencyMs
                         bytesAtLastPrn = receipt.bytesReceived
                         packetsSincePrn = 0
-                        onProgress(offset.toFloat() / firmware.size)
+                        onProgress(streamOffset.toFloat() / firmware.size)
                     }
                 }
             }
-            Logger.d { "Legacy DFU: Streamed $offset/${firmware.size} bytes (lastPRN=$bytesAtLastPrn)" }
+            Logger.d { "Legacy DFU: Streamed $streamOffset/${firmware.size} bytes (lastPRN=$bytesAtLastPrn)" }
         }
     }
 
@@ -361,13 +429,33 @@ class LegacyDfuTransport(
     /**
      * Send `RESET` to the device, instructing it to discard any in-progress transfer and reboot. Best-effort — the
      * device may disconnect before the write ACK lands; that's expected.
+     *
+     * The RESET op remains a `WITH_RESPONSE` write (the bootloader is contractually allowed to act on it before the
+     * GATT ACK lands, but we still request one so the link-layer queues it reliably), but we bound the wait with a
+     * short timeout. A missing acknowledgement is ambiguous: the device may have accepted RESET and rebooted, become
+     * unresponsive, disconnected, or simply failed to receive or complete the write. We report the outcome
+     * (acknowledged, unacknowledged, or operational failure) without claiming the RESET landed. Operational Exceptions
+     * are best-effort and non-fatal; structured-concurrency cancellation and Error subtypes propagate. The caller
+     * (`SecureDfuHandler`) tears the connection down afterwards regardless.
+     *
+     * Parent cancellation is preserved: a [CancellationException] that escapes `withTimeoutOrNull` is propagated.
      */
     override suspend fun abort() {
-        safeCatching {
-            writeControlPoint(byteArrayOf(LegacyDfuOpcode.RESET))
-            Logger.i { "Legacy DFU: RESET sent." }
+        val write =
+            try {
+                withTimeoutOrNull(RESET_WRITE_TIMEOUT) { writeControlPoint(byteArrayOf(LegacyDfuOpcode.RESET)) }
+            } catch (e: CancellationException) {
+                Logger.w(e) { "Legacy DFU: RESET write cancelled; disconnecting" }
+                throw e
+            } catch (@Suppress("TooGenericExceptionCaught") e: Exception) {
+                Logger.w(e) { "Legacy DFU: RESET write failed; disconnecting" }
+                return
+            }
+        if (write != null) {
+            Logger.i { "Legacy DFU: RESET write acknowledged; disconnecting" }
+        } else {
+            Logger.w { "Legacy DFU: RESET write unacknowledged within the timeout; disconnecting" }
         }
-            .onFailure { Logger.w(it) { "Legacy DFU: Failed to send RESET (device may already be disconnected)" } }
     }
 
     override suspend fun close() {
@@ -408,12 +496,11 @@ class LegacyDfuTransport(
 
     private suspend fun awaitResponse(timeout: Duration): LegacyDfuResponse = try {
         withTimeout(timeout) {
-            // Fail fast + accurately if the bootloader drops the link mid-handshake instead of answering. The stock
-            // same-MAC nRF Legacy bootloader (e.g. AdaDFU) commonly disconnects on the first Control Point command
-            // (stale-bond encryption mismatch); without this the receive() below just blocks until `timeout`, so
-            // the
-            // user waited the full 30s and saw a misleading "No response from Control Point" for what was really an
-            // immediate disconnect.
+            // Fail fast + accurately if the bootloader drops the link mid-handshake instead of answering.
+            // Some Legacy bootloader variants disconnect on the first Control Point command
+            // (e.g. stale-bond encryption mismatch); without this the receive() below just blocks until
+            // `timeout`, so the user waited for the full command timeout and saw a misleading
+            // "No response from Control Point" for what was really an immediate disconnect.
             bleConnection.withDisconnectTripwire(onDrop = ::handshakeDropError) {
                 // Drain any stray PRNs that arrive before the response we want.
                 while (true) {
@@ -425,32 +512,43 @@ class LegacyDfuTransport(
             }
         }
     } catch (_: TimeoutCancellationException) {
+        currentCoroutineContext().ensureActive()
         throw DfuException.Timeout("No response from Legacy DFU Control Point after $timeout")
     }
 
-    private suspend fun awaitPacketReceipt(): LegacyDfuResponse.PacketReceipt = try {
+    /**
+     * PRN wait used inside [streamFirmware]. Deliberately does NOT install its own [withDisconnectTripwire]:
+     * [streamFirmware] owns the sole disconnect watcher during PRN waits, and a nested tripwire here would race it and
+     * could surface a generic [DfuException.ConnectionFailed] from [handshakeDropError] instead of the typed
+     * [LegacyDfuException.MidStreamDisconnect] that drives the recovery profile. The handshake-level disconnect watcher
+     * lives in [awaitResponse] (its [withDisconnectTripwire] classifies link drops during Control Point responses).
+     *
+     * A [TimeoutCancellationException] may be raised by an outer [withTimeout] (structured cancellation) rather than
+     * the local 30 s missing-PRN timeout. We re-check [ensureActive] first so parent/outer cancellation is rethrown,
+     * and only then surface the local [DfuException.Timeout].
+     */
+    private suspend fun awaitPacketReceiptDuringStream(): LegacyDfuResponse.PacketReceipt = try {
         withTimeout(COMMAND_TIMEOUT) {
-            bleConnection.withDisconnectTripwire(onDrop = ::handshakeDropError) {
-                while (true) {
-                    val r = notificationChannel.receive()
-                    if (r is LegacyDfuResponse.PacketReceipt) return@withDisconnectTripwire r
-                    if (r is LegacyDfuResponse.Failure) {
-                        throw LegacyDfuException.ProtocolError(r.requestOpcode, r.status)
-                    }
-                    // Stray Success or Unknown → ignore.
+            while (true) {
+                val r = notificationChannel.receive()
+                if (r is LegacyDfuResponse.PacketReceipt) return@withTimeout r
+                if (r is LegacyDfuResponse.Failure) {
+                    throw LegacyDfuException.ProtocolError(r.requestOpcode, r.status)
                 }
-                @Suppress("UNREACHABLE_CODE")
-                error("unreachable")
+                // Stray Success or Unknown → ignore.
             }
+            @Suppress("UNREACHABLE_CODE")
+            error("unreachable")
         }
     } catch (_: TimeoutCancellationException) {
+        currentCoroutineContext().ensureActive()
         throw DfuException.Timeout("No packet receipt notification after $COMMAND_TIMEOUT")
     }
 
     /** Error for a link drop while awaiting a Control Point response — distinguishes a disconnect from a true stall. */
     private fun handshakeDropError(state: BleConnectionState): Throwable = DfuException.ConnectionFailed(
-        "BLE link dropped during DFU handshake (state=$state). The device disconnected before answering a DFU " +
-            "command — most often the stock Adafruit bootloader rebooting to the app. Retry, or use USB recovery.",
+        "BLE link dropped during DFU handshake (state=$state). The device disconnected before answering a Legacy DFU " +
+            "command; some bootloader variants reboot to the app or drop the link in this state.",
     )
 
     /**
@@ -460,11 +558,11 @@ class LegacyDfuTransport(
      * transfer state and rejects a fresh `START_DFU` with `INVALID_STATE` until it is reset. This is the common case
      * when *recovering* a device stranded in the bootloader.
      *
-     * We do NOT try to RESET on this connection: the stock Adafruit bootloader goes unresponsive immediately after
-     * emitting INVALID_STATE (the link dies by supervision timeout ~5 s later), so a write here never lands. Instead we
-     * fast-fail with [LegacyDfuException.StaleSessionReset]; [SecureDfuHandler] then resets the bootloader over a
-     * *fresh* connection (which is responsive up until START) before retrying. Mirrors the intent of Nordic
-     * `LegacyDfuImpl.resetAndRestart()`.
+     * We do NOT try to RESET on this connection: some Legacy bootloader variants become unresponsive after returning
+     * INVALID_STATE. Skip a potentially blocking same-connection RESET and use the bounded fresh-connection reset-prime
+     * path instead. We fast-fail with [LegacyDfuException.StaleSessionReset]; [SecureDfuHandler] then resets the
+     * bootloader over a *fresh* connection (which is responsive up until START) before retrying. Mirrors the intent of
+     * Nordic `LegacyDfuImpl.resetAndRestart()`.
      */
     private fun handleStartResponse(response: LegacyDfuResponse) {
         if (response is LegacyDfuResponse.Failure && response.status == LegacyDfuStatus.INVALID_STATE) {
@@ -514,6 +612,15 @@ class LegacyDfuTransport(
         private val CONNECT_TIMEOUT = 15.seconds
         private val COMMAND_TIMEOUT = 30.seconds
 
+        /**
+         * Best-effort timeout around the Legacy `RESET` (`0x06`) Control-Point write. Legacy bootloaders may disconnect
+         * before the RESET write acknowledgement returns. A missing acknowledgement is ambiguous: the device may have
+         * accepted RESET and rebooted, become unresponsive, disconnected, or failed to receive/complete the write.
+         * Bound the acknowledgement wait to one second, report it as unacknowledged, then rely on connection teardown
+         * and the subsequent retry/re-advertisement flow.
+         */
+        internal val RESET_WRITE_TIMEOUT = 1.seconds
+
         /**
          * Time to wait for the START_DFU response notification.
          *
@@ -537,18 +644,13 @@ class LegacyDfuTransport(
         private val STREAM_TIMEOUT = 15.minutes
 
         /**
-         * Packet-receipt-notification interval (packets between flow-control ACKs). Higher values mean fewer
-         * notification round-trips per byte and therefore faster throughput, at the cost of a slightly longer recovery
-         * window if a packet is dropped (we have to wait until the next PRN boundary to detect the gap).
-         *
-         * Capped at 10 to match Nordic's own Legacy DFU implementation, which force-limits legacy PRN to ≤10 with the
-         * comment: "DFU bootloaders from SDK 6.0.0 or older were unable to save incoming data to flash as fast as they
-         * are being sent … PRN = 10 may be the highest supported value" and treats status 6 (OPERATION_FAILED) as "data
-         * sent too fast — reduce PRN to 10 or less." The stock Adafruit bootloader shares this SDK11 flash-write path,
-         * so a higher value (we previously used 30, tuned for the faster OTAFIX fork) risks OPERATION_FAILED mid-stream
-         * on stock bootloaders. 10 is the safe ceiling that still batches flow-control ACKs.
+         * Per-receipt latency threshold above which a PRN wait is logged at WARN. The Legacy bootloader normally ACKs
+         * each PRN window inside a few tens of milliseconds. A latency at or above this threshold is diagnostic only —
+         * it may indicate bootloader backpressure, flash activity, BLE scheduling delay, or degraded link conditions.
+         * It is not by itself a definite precursor to a supervision-timeout drop. Below this threshold receipts log at
+         * DEBUG.
          */
-        internal const val PRN_INTERVAL_PACKETS = 10
+        internal const val PRN_LATENCY_WARN_THRESHOLD_MS = 1_000L
 
         /**
          * Universally-safe Legacy DFU packet size (20 bytes — the original ATT_MTU minus the 3-byte ATT header). Used
diff --git a/feature/firmware/src/commonMain/kotlin/org/meshtastic/feature/firmware/ota/dfu/SecureDfuHandler.kt b/feature/firmware/src/commonMain/kotlin/org/meshtastic/feature/firmware/ota/dfu/SecureDfuHandler.kt
index e920e204d..9dc20b61b 100644
--- a/feature/firmware/src/commonMain/kotlin/org/meshtastic/feature/firmware/ota/dfu/SecureDfuHandler.kt
+++ b/feature/firmware/src/commonMain/kotlin/org/meshtastic/feature/firmware/ota/dfu/SecureDfuHandler.kt
@@ -64,11 +64,42 @@ private const val DFU_REBOOT_WAIT_MS = 3_000L
 private const val RETRY_DELAY_MS = 2_000L
 private const val CONNECT_ATTEMPTS = 4
 
-/** Legacy DFU can't resume, so retry the whole session this many times before giving up. */
-private const val LEGACY_SESSION_ATTEMPTS = 3
+/**
+ * Real upload attempts a confirmed Legacy primary is allowed before declaring failure. Legacy DFU has no resume, so a
+ * failed upload requires a whole fresh session (new transport + reconnect + re-handshake). Stale-session cleanup cycles
+ * do NOT consume this budget (see [MAX_LEGACY_STALE_RESETS]).
+ */
+internal const val LEGACY_SESSION_ATTEMPTS = 3
+
+/**
+ * Maximum number of stale-session cleanup cycles a Legacy run will perform. A [LegacyDfuException.StaleSessionReset]
+ * surfaces when the bootloader rejects START_DFU with INVALID_STATE (a half-finished previous session is wedged in the
+ * bootloader). The cleanup connects a fresh transport, RESETs, and waits for the bootloader to reboot into a clean OTA
+ * session. This budget is separate from [LEGACY_SESSION_ATTEMPTS] so a run that legitimately needs to reset-prime more
+ * than once (e.g. a mid-stream drop that left the bootloader wedged twice in a row) is not punished by losing an upload
+ * attempt for a cleanup that did not actually try to upload.
+ */
+internal const val MAX_LEGACY_STALE_RESETS = 2
 
 /** Limited probe/fallback budget for inconclusive detection or alternate-protocol attempts. */
-private const val LIMITED_SESSION_ATTEMPTS = 1
+internal const val LIMITED_SESSION_ATTEMPTS = 1
+
+/**
+ * Two-stage session budget.
+ *
+ * [preEngagementAttempts] limits speculative connection attempts before the selected DFU service has been confirmed.
+ * [engagedAttempts] applies after a connection to that protocol's DFU service succeeds. Engagement means the selected
+ * protocol's DFU service was connected to; it does not by itself universally prevent fallback (see
+ * [DfuFallbackCoordinator] for the fallback policy). Secure always carries [preEngagementAttempts] == [engagedAttempts]
+ * so engagement never expands its budget.
+ */
+internal data class DfuAttemptBudget(val preEngagementAttempts: Int, val engagedAttempts: Int) {
+    init {
+        require(preEngagementAttempts in 1..engagedAttempts) {
+            "preEngagementAttempts ($preEngagementAttempts) must be in 1..engagedAttempts ($engagedAttempts)"
+        }
+    }
+}
 
 /**
  * Transport-dispatch key: which [DfuUploadTransport] implementation ([SecureDfuTransport] or [LegacyDfuTransport]) to
@@ -105,12 +136,12 @@ internal sealed class DfuUploadResult {
 }
 
 /**
- * Owns fallback ordering and retry budget policy. Fallback is intentionally limited to failures that happen before the
- * selected transport has connected and engaged its DFU protocol session.
+ * Owns fallback ordering and retry budget policy. Whether an alternate protocol is tried after a primary failure
+ * depends on the [detection] outcome and engagement state — see [shouldTryAlternateAfterFailure].
  */
 internal class DfuFallbackCoordinator(private val detection: BootloaderDetection) {
 
-    suspend fun execute(uploadWithRetry: suspend (DfuProtocolKind, Int) -> DfuUploadResult) {
+    suspend fun execute(uploadWithRetry: suspend (DfuProtocolKind, DfuAttemptBudget) -> DfuUploadResult) {
         val orderedProtocols = detection.orderedProtocols()
         Logger.i { "DFU: detection=$detection → protocols=$orderedProtocols" }
 
@@ -118,17 +149,17 @@ internal class DfuFallbackCoordinator(private val detection: BootloaderDetection
         for ((index, protocol) in orderedProtocols.withIndex()) {
             val isPrimary = index == 0
             val hasAlternateProtocol = index < orderedProtocols.lastIndex
-            val sessionAttempts = sessionAttemptsFor(protocol, isPrimary)
+            val budget = budgetFor(protocol, isPrimary)
             if (isPrimary) {
-                Logger.i { "DFU: primary protocol=$protocol ($sessionAttempts session attempt(s))" }
-            } else {
-                Logger.w {
-                    "DFU: falling back to alternate protocol=$protocol before protocol engagement " +
-                        "(detection=$detection, sessionAttempts=$sessionAttempts)"
+                Logger.i {
+                    "DFU: primary protocol=$protocol " +
+                        "(upload attempts pre=${budget.preEngagementAttempts}, engaged=${budget.engagedAttempts})"
                 }
+            } else {
+                Logger.w { "DFU: trying alternate protocol=$protocol (detection=$detection, budget=$budget)" }
             }
 
-            when (val result = uploadWithRetry(protocol, sessionAttempts)) {
+            when (val result = uploadWithRetry(protocol, budget)) {
                 DfuUploadResult.Success -> return
 
                 is DfuUploadResult.Failure -> {
@@ -150,7 +181,7 @@ internal class DfuFallbackCoordinator(private val detection: BootloaderDetection
 
     /**
      * The ordered list of protocols to attempt for this detection outcome. The first element is the primary; the second
-     * is the alternate, attempted only if the primary fails before protocol engagement.
+     * is the alternate. Whether the alternate is reached depends on [shouldTryAlternateAfterFailure], not ordering.
      */
     private fun BootloaderDetection.orderedProtocols(): List<DfuProtocolKind> = when (this) {
         BootloaderDetection.LegacyObserved -> listOf(DfuProtocolKind.LEGACY, DfuProtocolKind.SECURE)
@@ -159,29 +190,45 @@ internal class DfuFallbackCoordinator(private val detection: BootloaderDetection
     }
 
     /**
-     * Session retry budget for the given protocol in this detection context.
-     * - Confirmed Legacy primary ([BootloaderDetection.LegacyObserved]): [LEGACY_SESSION_ATTEMPTS] (3) to cover the
-     *   stale-session reset-prime recovery path.
-     * - Unknown Legacy primary and alternate protocols: [LIMITED_SESSION_ATTEMPTS] (1) to keep speculative probes short
-     *   enough that a Secure bootloader still has time to accept its fallback attempt.
+     * Two-stage upload-attempt budget for the given protocol in this detection context.
+     * - Confirmed Legacy primary ([BootloaderDetection.LegacyObserved]): full [LEGACY_SESSION_ATTEMPTS] (3) budget
+     *   before AND after engagement — the protocol was conclusively observed.
+     * - Unknown Legacy primary: [LIMITED_SESSION_ATTEMPTS] (1) pre-engagement probe so Secure fallback stays timely,
+     *   promoted to [LEGACY_SESSION_ATTEMPTS] (3) once Legacy engages and is conclusively identified.
+     * - Legacy alternate (e.g. SecureObserved → Legacy fallback): same Unknown-style two-stage budget — pre-engagement
+     *   probe of 1, full Legacy budget of 3 once engaged.
+     * - Secure: 1/1 — never expanded by engagement, since the Secure recovery path is single-shot.
      */
-    private fun sessionAttemptsFor(protocol: DfuProtocolKind, isPrimary: Boolean): Int = when {
-        protocol == DfuProtocolKind.LEGACY && isPrimary && detection == BootloaderDetection.LegacyObserved ->
-            LEGACY_SESSION_ATTEMPTS
+    private fun budgetFor(protocol: DfuProtocolKind, isPrimary: Boolean): DfuAttemptBudget = when {
+        protocol == DfuProtocolKind.SECURE ->
+            DfuAttemptBudget(
+                preEngagementAttempts = LIMITED_SESSION_ATTEMPTS,
+                engagedAttempts = LIMITED_SESSION_ATTEMPTS,
+            )
 
-        else -> LIMITED_SESSION_ATTEMPTS
+        protocol == DfuProtocolKind.LEGACY && isPrimary && detection == BootloaderDetection.LegacyObserved ->
+            DfuAttemptBudget(
+                preEngagementAttempts = LEGACY_SESSION_ATTEMPTS,
+                engagedAttempts = LEGACY_SESSION_ATTEMPTS,
+            )
+
+        // Unknown Legacy primary and Legacy alternate: speculative probe that promotes on engagement.
+        else ->
+            DfuAttemptBudget(
+                preEngagementAttempts = LIMITED_SESSION_ATTEMPTS,
+                engagedAttempts = LEGACY_SESSION_ATTEMPTS,
+            )
     }
 
     private fun shouldTryAlternateAfterFailure(isPrimary: Boolean, protocolEngaged: Boolean): Boolean {
         if (!isPrimary) return false
         return when (detection) {
-            // Unknown can mean the primary Legacy advertisement was missed, so a pre-engagement failure is the useful
-            // fallback signal.
+            // Unknown detection: Legacy is tried first, Secure fallback only if Legacy fails before engagement.
             BootloaderDetection.Unknown -> !protocolEngaged
 
-            // Conclusive detections already observed the selected protocol's service. If it never engages, retrying the
-            // opposite service is usually a long doomed scan. Keep fallback only as insurance for a stale/wrong
-            // conclusive signal after the observed service was actually reached.
+            // Conclusive detection: a pre-engagement failure of the observed protocol does NOT immediately try the
+            // opposite protocol. Alternate fallback is retained only after engagement failure — insurance against a
+            // stale or misleading conclusive detection.
             BootloaderDetection.LegacyObserved,
             BootloaderDetection.SecureObserved,
             -> protocolEngaged
@@ -203,6 +250,183 @@ private fun DfuProtocolKind.serviceUuid(): Uuid = when (this) {
     DfuProtocolKind.SECURE -> SecureDfuUuids.SERVICE
 }
 
+/**
+ * Drives the bounded Legacy/Secure DFU upload retry loop. Extracted from [SecureDfuHandler.runDfuUploadWithRetry] so
+ * the retry policy can be unit-tested without bringing up the full BLE stack — callers supply a [runUploadSession]
+ * lambda that returns the next [DfuUploadResult] and a [resetStaleBootloader] lambda that performs the fresh-connection
+ * RESET-prime cycle.
+ *
+ * Policy:
+ * - Active attempt limit starts at [DfuAttemptBudget.preEngagementAttempts]; the first [DfuUploadResult.Failure] whose
+ *   `protocolEngaged=true` promotes the active limit to [DfuAttemptBudget.engagedAttempts] BEFORE the budget-exhaustion
+ *   check runs, so an engaged Legacy probe receives its full retry budget.
+ * - Non-[LegacyDfuException.StaleSessionReset] failures consume one upload attempt.
+ * - A [LegacyDfuException.StaleSessionReset] consumes one of [maxStaleResets], NOT an upload attempt — the cleanup
+ *   cycle never tried to upload.
+ * - After the first [LegacyDfuException.MidStreamDisconnect] on Legacy, every subsequent Legacy attempt uses
+ *   [LegacyDfuStreamProfile.RECOVERY]; a stale-session alone never switches the profile.
+ * - All exit paths are bounded: the loop stops when the upload-attempt budget is exhausted or the stale-response budget
+ *   is exceeded. Up to [maxStaleResets] reset-prime cleanups run; the next stale response terminates without another
+ *   cleanup.
+ * - [DfuProtocolKind.SECURE] carries pre==engaged budgets, so engagement promotion is a no-op there.
+ *
+ * If a mid-stream drop preceded the terminal failure, the first [LegacyDfuException.MidStreamDisconnect] is attached as
+ * a suppressed exception on the surfaced error so the underlying cause stays visible.
+ */
+@Suppress("CyclomaticComplexMethod", "NestedBlockDepth", "LongMethod")
+internal suspend fun runDfuRetryLoop(
+    protocol: DfuProtocolKind,
+    budget: DfuAttemptBudget,
+    maxStaleResets: Int,
+    runUploadSession: suspend (LegacyDfuStreamProfile) -> DfuUploadResult,
+    resetStaleBootloader: suspend () -> Unit,
+    interAttemptDelay: suspend () -> Unit,
+): DfuUploadResult {
+    var uploadAttempts = 0
+    var staleResets = 0
+    var midStreamSeen = false
+    var firstMidStreamError: LegacyDfuException.MidStreamDisconnect? = null
+    var lastError: Throwable? = null
+    var protocolEngaged = false
+    var activeAttempts = budget.preEngagementAttempts
+
+    while (uploadAttempts < activeAttempts) {
+        var skipInterAttemptDelay = false
+
+        val profile =
+            if (protocol == DfuProtocolKind.LEGACY && midStreamSeen) {
+                LegacyDfuStreamProfile.RECOVERY
+            } else {
+                LegacyDfuStreamProfile.NORMAL
+            }
+
+        uploadAttempts++
+        Logger.i { "DFU: upload attempt $uploadAttempts/$activeAttempts ($protocol, profile=$profile)" }
+        when (val result = runUploadSession(profile)) {
+            DfuUploadResult.Success -> return result
+
+            is DfuUploadResult.Failure -> {
+                lastError = result.error
+                val wasEngaged = protocolEngaged
+                protocolEngaged = protocolEngaged || result.protocolEngaged
+
+                // Promote the upload budget the moment a session first reports protocolEngaged. Promotion happens
+                // BEFORE the budget-exhaustion check (the while condition on the next iteration) so an engaged
+                // Legacy probe receives its full retry budget. For Secure (pre==engaged) this is a no-op.
+                if (!wasEngaged && protocolEngaged && activeAttempts < budget.engagedAttempts) {
+                    activeAttempts = budget.engagedAttempts
+                    Logger.i {
+                        "DFU: protocol engaged; promoting upload budget " +
+                            "(${budget.preEngagementAttempts} → ${budget.engagedAttempts}) ($protocol)"
+                    }
+                }
+
+                // Legacy-only stale-session handling. StaleSessionReset consumes a stale-reset cycle, NOT an upload
+                // attempt — the cycle never tried to upload.
+                if (protocol == DfuProtocolKind.LEGACY && result.error is LegacyDfuException.StaleSessionReset) {
+                    uploadAttempts--
+                    staleResets++
+                    if (staleResets > maxStaleResets) {
+                        Logger.w {
+                            "DFU: stale cleanup exhausted ($staleResets > $maxStaleResets) — giving up on $protocol"
+                        }
+                        break
+                    }
+                    if (midStreamSeen) {
+                        // Expected: a stale session is the natural residue of a mid-stream drop, so an informational
+                        // log keeps the operator calm.
+                        Logger.i {
+                            "DFU: stale cleanup cycle $staleResets/$maxStaleResets after mid-stream drop " +
+                                "(expected) ($protocol)"
+                        }
+                    } else {
+                        // Unexpected: the very first session came up stale with no preceding drop, which usually
+                        // means the device was stranded in the bootloader before this update began.
+                        Logger.w {
+                            "DFU: stale cleanup cycle $staleResets/$maxStaleResets — unexpected initial stale " +
+                                "session ($protocol)"
+                        }
+                    }
+                    resetStaleBootloader()
+                    skipInterAttemptDelay = true
+                }
+
+                if (!skipInterAttemptDelay) {
+                    // Mid-stream disconnect: switch subsequent Legacy uploads to the recovery profile and surface the
+                    // host in-flight offset in the log so it is not hidden by later cleanup failures.
+                    val error = result.error
+                    if (protocol == DfuProtocolKind.LEGACY && error is LegacyDfuException.MidStreamDisconnect) {
+                        if (!midStreamSeen) {
+                            midStreamSeen = true
+                            firstMidStreamError = error
+                        }
+                        Logger.w(error) {
+                            "DFU: mid-stream disconnect at host in-flight offset " +
+                                "${error.bytesSent}/${error.totalBytes} " +
+                                "(state=${error.connectionState}); subsequent $protocol attempts will use " +
+                                "${LegacyDfuStreamProfile.RECOVERY}"
+                        }
+                    } else {
+                        Logger.w(error) {
+                            "DFU: upload attempt $uploadAttempts/$activeAttempts failed ($protocol): " +
+                                "${error::class.simpleName}"
+                        }
+                    }
+
+                    if (uploadAttempts < activeAttempts) interAttemptDelay()
+                }
+            }
+        }
+    }
+
+    val error = lastError ?: DfuException.TransferFailed("DFU upload failed after $activeAttempts attempts ($protocol)")
+    // Keep the original mid-stream failure visible even when the terminal failure is a later cleanup/transfer error.
+    firstMidStreamError?.let { drop ->
+        if (error !== drop) {
+            error.addSuppressed(drop)
+        }
+    }
+    return DfuUploadResult.Failure(error, protocolEngaged)
+}
+
+/**
+ * Cleanup policy for a DFU upload session's transport. Extracted from [SecureDfuHandler.runUploadSession]'s finally
+ * block so the skip-abort / always-close contract can be tested directly.
+ *
+ * Semantics:
+ * - Executes in [NonCancellable] so close always runs even if the caller was cancelled.
+ * - Skips [DfuUploadTransport.abort] after a completed session — nothing to abort.
+ * - Skips [DfuUploadTransport.abort] after a [LegacyDfuException.StaleSessionReset] — some Legacy variants become
+ *   unresponsive after INVALID_STATE, so the same-connection RESET may block or be ineffective; the fresh-connection
+ *   reset-prime path owns recovery.
+ * - Skips [DfuUploadTransport.abort] after a [LegacyDfuException.MidStreamDisconnect] — the link has already dropped,
+ *   so there is no live connection over which to deliver an abort.
+ * - Attempts [DfuUploadTransport.abort] after a [DfuException.ConnectionFailed] — [DfuException.ConnectionFailed] is
+ *   produced by [connectWithRetry] for any exhausted connection or setup failure, including cases where a GATT
+ *   connection was established but a subsequent step (e.g. unsupported bootloader version, profile setup) failed. A
+ *   usable connection may still exist, so the bounded best-effort abort is attempted before close.
+ * - Otherwise calls [DfuUploadTransport.abort], then [DfuUploadTransport.close] in a nested finally.
+ * - [DfuUploadTransport.abort]'s own cancellation/Error propagation contract is preserved: the [NonCancellable] context
+ *   prevents the outer coroutine cancellation from interrupting close, but an [Error] thrown by abort still propagates
+ *   after close runs.
+ */
+internal suspend fun cleanupDfuSessionTransport(
+    transport: DfuUploadTransport,
+    completed: Boolean,
+    sessionFailure: Throwable?,
+) {
+    withContext(NonCancellable) {
+        val skipAbort =
+            sessionFailure is LegacyDfuException.StaleSessionReset ||
+                sessionFailure is LegacyDfuException.MidStreamDisconnect
+        try {
+            if (!completed && !skipAbort) transport.abort()
+        } finally {
+            transport.close()
+        }
+    }
+}
+
 /**
  * KMP [FirmwareUpdateHandler] for nRF52 devices.
  *
@@ -285,10 +509,10 @@ class SecureDfuHandler(
                 // reconnect + re-handshake), mirroring Nordic's DFU library ("the Legacy DFU will start again"). A
                 // stock bootloader that leaves the first session's control-point handshake unanswered often responds
                 // after a clean reconnect. Secure DFU resumes in place, so it runs a single session.
-                // DfuFallbackCoordinator resolves the detection into an ordered protocol list and may try the
-                // alternate protocol only before the selected protocol has connected and engaged its DFU session.
-                DfuFallbackCoordinator(detection).execute { protocol, sessionAttempts ->
-                    runDfuUploadWithRetry(protocol, target, pkg, sessionAttempts, updateState)
+                // DfuFallbackCoordinator resolves the detection into an ordered protocol list; whether the alternate
+                // is tried depends on detection-specific engagement rules (shouldTryAlternateAfterFailure).
+                DfuFallbackCoordinator(detection).execute { protocol, budget ->
+                    runDfuUploadWithRetry(protocol, target, pkg, budget, updateState)
                 }
                 zipFile
             }
@@ -370,59 +594,44 @@ class SecureDfuHandler(
     }
 
     /**
-     * Run the connect + init + firmware upload, retrying the whole session up to [attempts] times. Each attempt uses a
-     * fresh [DfuUploadTransport] (new GATT connection + re-handshake) since Legacy DFU can't resume mid-stream.
-     *
-     * Returns the last failure with whether this protocol ever connected and started its DFU session, allowing fallback
-     * orchestration to switch protocols only for pre-engagement failures.
+     * Run the connect + init + firmware upload, retrying within the [DfuFallbackCoordinator] budget. Delegates the
+     * bounded retry logic to [runDfuRetryLoop] (testable without bringing up the full BLE stack); the lambdas here
+     * supply the real [runUploadSession] and [resetStaleBootloader] implementations.
      */
     private suspend fun runDfuUploadWithRetry(
         protocol: DfuProtocolKind,
         target: String,
         pkg: DfuZipPackage,
-        attempts: Int,
+        budget: DfuAttemptBudget,
         updateState: (FirmwareUpdateState) -> Unit,
-    ): DfuUploadResult {
-        var lastError: Throwable? = null
-        var protocolEngaged = false
-        repeat(attempts) { i ->
-            val attempt = i + 1
-            Logger.i { "DFU: upload session attempt $attempt/$attempts ($protocol)" }
-            when (val result = runUploadSession(protocol, target, pkg, updateState)) {
-                DfuUploadResult.Success -> return result
+    ): DfuUploadResult = runDfuRetryLoop(
+        protocol = protocol,
+        budget = budget,
+        maxStaleResets = MAX_LEGACY_STALE_RESETS,
+        runUploadSession = { profile -> runUploadSession(protocol, target, pkg, profile, updateState) },
+        resetStaleBootloader = { resetStaleBootloader(protocol, target) },
+        interAttemptDelay = { delay(SESSION_RETRY_DELAY_MS) },
+    )
 
-                is DfuUploadResult.Failure -> {
-                    lastError = result.error
-                    protocolEngaged = protocolEngaged || result.protocolEngaged
-                    Logger.w(result.error) {
-                        "DFU: upload session $attempt/$attempts failed ($protocol): ${result.error::class.simpleName}"
-                    }
-                    // A stock bootloader holding a wedged session from an interrupted flash rejects START with
-                    // INVALID_STATE, then goes unresponsive (the RESET can't land on that connection). A *fresh*
-                    // connection is responsive up until START, so reset it there — the device reboots into a clean OTA
-                    // session (GPREGRET OTA flag is retained) that the next attempt can flash normally.
-                    if (result.error is LegacyDfuException.StaleSessionReset && attempt < attempts) {
-                        resetStaleBootloader(protocol, target)
-                    }
-                    if (attempt < attempts) delay(SESSION_RETRY_DELAY_MS)
-                }
-            }
-        }
-        return DfuUploadResult.Failure(
-            lastError ?: DfuException.TransferFailed("DFU upload failed after $attempts attempts ($protocol)"),
-            protocolEngaged,
-        )
-    }
+    private fun createTransport(
+        protocol: DfuProtocolKind,
+        target: String,
+        legacyProfile: LegacyDfuStreamProfile = LegacyDfuStreamProfile.NORMAL,
+    ): DfuUploadTransport = when (protocol) {
+        // Legacy transport receives the selected stream profile; default NORMAL keeps prior behavior for any caller
+        // that does not pass an explicit profile (e.g. reset-prime, which is a control-point operation only).
+        DfuProtocolKind.LEGACY ->
+            LegacyDfuTransport(bleScanner, bleConnectionFactory, target, dispatchers.default, legacyProfile)
 
-    private fun createTransport(protocol: DfuProtocolKind, target: String): DfuUploadTransport = when (protocol) {
-        DfuProtocolKind.LEGACY -> LegacyDfuTransport(bleScanner, bleConnectionFactory, target, dispatchers.default)
         DfuProtocolKind.SECURE -> SecureDfuTransport(bleScanner, bleConnectionFactory, target, dispatchers.default)
     }
 
     /**
      * Reboot a bootloader wedged in a stale DFU session. Connects a fresh transport (which is responsive before any
-     * START) and issues RESET (0x06) via [DfuUploadTransport.abort], then waits for the reboot + re-advertise. Best
-     * effort: any failure here is non-fatal — the caller retries the upload regardless.
+     * START) and issues RESET (0x06) via the bounded [DfuUploadTransport.abort], then closes the connection and waits
+     * for the configured [RESET_PRIME_REBOOT_WAIT_MS] reboot/re-advertisement interval. Operational [Exception]s during
+     * abort are best-effort (swallowed by [DfuUploadTransport.abort]); structured-concurrency cancellation and [Error]
+     * subtypes propagate per the abort contract.
      */
     private suspend fun resetStaleBootloader(protocol: DfuProtocolKind, target: String) {
         Logger.i { "DFU: reset-priming stale bootloader before retry" }
@@ -432,7 +641,7 @@ class SecureDfuHandler(
                 .connectToDfuMode()
                 .onSuccess {
                     transport.abort()
-                    Logger.i { "DFU: reset-prime RESET sent; waiting for clean reboot" }
+                    Logger.i { "DFU: reset-prime RESET attempted; disconnecting and waiting for re-advertisement" }
                 }
                 .onFailure { Logger.w(it) { "DFU: reset-prime connect failed: ${it.message}" } }
         } finally {
@@ -446,11 +655,13 @@ class SecureDfuHandler(
         protocol: DfuProtocolKind,
         target: String,
         pkg: DfuZipPackage,
+        legacyProfile: LegacyDfuStreamProfile,
         updateState: (FirmwareUpdateState) -> Unit,
     ): DfuUploadResult {
-        val transport: DfuUploadTransport = createTransport(protocol, target)
+        val transport: DfuUploadTransport = createTransport(protocol, target, legacyProfile)
         var completed = false
         var protocolEngaged = false
+        var sessionFailure: Throwable? = null
         try {
             connectWithRetry(transport, protocol, updateState)
             protocolEngaged = true
@@ -495,13 +706,11 @@ class SecureDfuHandler(
             return DfuUploadResult.Success
         } catch (e: CancellationException) {
             throw e
-        } catch (@Suppress("TooGenericExceptionCaught") e: Throwable) {
+        } catch (@Suppress("TooGenericExceptionCaught") e: Exception) {
+            sessionFailure = e
             return DfuUploadResult.Failure(e, protocolEngaged)
         } finally {
-            withContext(NonCancellable) {
-                if (!completed) transport.abort()
-                transport.close()
-            }
+            cleanupDfuSessionTransport(transport, completed, sessionFailure)
         }
     }
 
diff --git a/feature/firmware/src/commonTest/kotlin/org/meshtastic/feature/firmware/ota/dfu/DfuFallbackCoordinatorTest.kt b/feature/firmware/src/commonTest/kotlin/org/meshtastic/feature/firmware/ota/dfu/DfuFallbackCoordinatorTest.kt
index a177c13b4..b113a4428 100644
--- a/feature/firmware/src/commonTest/kotlin/org/meshtastic/feature/firmware/ota/dfu/DfuFallbackCoordinatorTest.kt
+++ b/feature/firmware/src/commonTest/kotlin/org/meshtastic/feature/firmware/ota/dfu/DfuFallbackCoordinatorTest.kt
@@ -63,11 +63,11 @@ class DfuFallbackCoordinatorTest {
     fun `Unknown falls back from Legacy to Secure on pre-engagement failure`() = runTest {
         val coordinator = DfuFallbackCoordinator(BootloaderDetection.Unknown)
         val protocols = mutableListOf<DfuProtocolKind>()
-        val attemptCounts = mutableListOf<Int>()
+        val budgets = mutableListOf<DfuAttemptBudget>()
         assertFailsWith<RuntimeException> {
-            coordinator.execute { protocol, attempts ->
+            coordinator.execute { protocol, budget ->
                 protocols.add(protocol)
-                attemptCounts.add(attempts)
+                budgets.add(budget)
                 if (protocol == DfuProtocolKind.LEGACY) {
                     DfuUploadResult.Failure(RuntimeException("connect failed"), protocolEngaged = false)
                 } else {
@@ -81,7 +81,9 @@ class DfuFallbackCoordinatorTest {
                 assertEquals("also failed", thrown.suppressedExceptions.first().message)
             }
         assertEquals(listOf(DfuProtocolKind.LEGACY, DfuProtocolKind.SECURE), protocols)
-        assertEquals(listOf(1, 1), attemptCounts)
+        // Unknown Legacy primary budget shape: pre=1 probe, engaged=3; promotion is not exercised because this test
+        // fails before engagement. Secure remains 1/1.
+        assertEquals(listOf(DfuAttemptBudget(1, 3), DfuAttemptBudget(1, 1)), budgets)
     }
 
     @Test
@@ -98,14 +100,17 @@ class DfuFallbackCoordinatorTest {
     }
 
     @Test
-    fun `LegacyObserved gives Legacy 3 session attempts`() = runTest {
+    fun `LegacyObserved gives Legacy 3 upload attempts`() = runTest {
         val coordinator = DfuFallbackCoordinator(BootloaderDetection.LegacyObserved)
-        val attemptCounts = mutableListOf<Int>()
-        coordinator.execute { _, attempts ->
-            attemptCounts.add(attempts)
+        val budgets = mutableListOf<DfuAttemptBudget>()
+        coordinator.execute { _, budget ->
+            budgets.add(budget)
             DfuUploadResult.Success
         }
-        assertEquals(listOf(3), attemptCounts) // LEGACY_SESSION_ATTEMPTS
+        assertEquals(
+            listOf(DfuAttemptBudget(3, 3)),
+            budgets,
+        ) // LEGACY_SESSION_ATTEMPTS — upload-attempt budget; stale cleanup is separate
     }
 
     @Test
@@ -123,11 +128,11 @@ class DfuFallbackCoordinatorTest {
     }
 
     @Test
-    fun `LegacyObserved gives Secure fallback 1 session attempt after Legacy engages`() = runTest {
+    fun `LegacyObserved gives Secure fallback budget of 1 upload attempt after Legacy engages`() = runTest {
         val coordinator = DfuFallbackCoordinator(BootloaderDetection.LegacyObserved)
-        val attemptCounts = mutableListOf<Int>()
-        coordinator.execute { protocol, attempts ->
-            attemptCounts.add(attempts)
+        val budgets = mutableListOf<DfuAttemptBudget>()
+        coordinator.execute { protocol, budget ->
+            budgets.add(budget)
             if (protocol == DfuProtocolKind.LEGACY) {
                 DfuUploadResult.Failure(RuntimeException("legacy protocol failed"), protocolEngaged = true)
             } else {
@@ -135,20 +140,23 @@ class DfuFallbackCoordinatorTest {
             }
         }
         assertEquals(
-            listOf(3, 1),
-            attemptCounts,
-        ) // Legacy primary=LEGACY_SESSION_ATTEMPTS, Secure fallback=LIMITED_SESSION_ATTEMPTS
+            listOf(DfuAttemptBudget(3, 3), DfuAttemptBudget(1, 1)),
+            budgets,
+        ) // Legacy primary=LEGACY_SESSION_ATTEMPTS (upload attempts), Secure fallback=LIMITED_SESSION_ATTEMPTS
     }
 
     @Test
-    fun `Unknown gives Legacy primary 1 session attempt`() = runTest {
+    fun `Unknown gives Legacy primary a pre-engagement probe budget of 1 with engaged promotion to 3`() = runTest {
         val coordinator = DfuFallbackCoordinator(BootloaderDetection.Unknown)
-        val attemptCounts = mutableListOf<Int>()
-        coordinator.execute { _, attempts ->
-            attemptCounts.add(attempts)
+        val budgets = mutableListOf<DfuAttemptBudget>()
+        coordinator.execute { _, budget ->
+            budgets.add(budget)
             DfuUploadResult.Success
         }
-        assertEquals(listOf(1), attemptCounts) // LIMITED_SESSION_ATTEMPTS keeps speculative Unknown probes bounded
+        assertEquals(
+            listOf(DfuAttemptBudget(1, 3)),
+            budgets,
+        ) // LIMITED_SESSION_ATTEMPTS keeps speculative Unknown probes bounded; engagement promotes to 3
     }
 
     @Test
diff --git a/feature/firmware/src/commonTest/kotlin/org/meshtastic/feature/firmware/ota/dfu/LegacyDfuRetryPolicyTest.kt b/feature/firmware/src/commonTest/kotlin/org/meshtastic/feature/firmware/ota/dfu/LegacyDfuRetryPolicyTest.kt
new file mode 100644
index 000000000..120cadb9e
--- /dev/null
+++ b/feature/firmware/src/commonTest/kotlin/org/meshtastic/feature/firmware/ota/dfu/LegacyDfuRetryPolicyTest.kt
@@ -0,0 +1,635 @@
+/*
+ * Copyright (c) 2026 Meshtastic LLC
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+@file:Suppress("MagicNumber")
+
+package org.meshtastic.feature.firmware.ota.dfu
+
+import kotlinx.coroutines.test.runTest
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertFailsWith
+import kotlin.test.assertFalse
+import kotlin.test.assertIs
+import kotlin.test.assertTrue
+
+/**
+ * Tests the Legacy DFU retry policy extracted into [runDfuRetryLoop]. The retry loop is driven entirely by lambdas — a
+ * session-result supplier and a stale-reset cleanup hook — so the policy can be exercised without bringing up the full
+ * BLE stack. Covers the drop/stale/success sequences that the mid-stream-disconnect recovery design depends on.
+ */
+class LegacyDfuRetryPolicyTest {
+
+    /**
+     * Sequence A: NORMAL upload drops mid-stream → stale cleanup → RECOVERY upload succeeds.
+     *
+     * Verifies:
+     * - only 1 upload attempt is consumed before success (stale cleanup does NOT consume an upload attempt)
+     * - RECOVERY profile is selected for the post-drop attempts
+     * - exactly one stale-reset cleanup cycle runs
+     * - the surfaced result is Success
+     */
+    @Test
+    fun `drop then stale then success consumes one upload attempt and switches to RECOVERY`() = runTest {
+        val sessions = mutableListOf<LegacyDfuStreamProfile>()
+        var staleResets = 0
+
+        val outcome =
+            runDfuRetryLoop(
+                protocol = DfuProtocolKind.LEGACY,
+                // Confirmed-Legacy style budget: full budget before and after engagement.
+                budget = DfuAttemptBudget(LEGACY_SESSION_ATTEMPTS, LEGACY_SESSION_ATTEMPTS),
+                maxStaleResets = MAX_LEGACY_STALE_RESETS,
+                runUploadSession = { profile ->
+                    sessions.add(profile)
+                    when (sessions.size) {
+                        1 ->
+                            DfuUploadResult.Failure(
+                                LegacyDfuException.MidStreamDisconnect(
+                                    20,
+                                    200,
+                                    "Disconnected",
+                                    lastConfirmedBytes = -1,
+                                ),
+                                false,
+                            )
+
+                        2 -> DfuUploadResult.Failure(LegacyDfuException.StaleSessionReset(), true)
+
+                        else -> DfuUploadResult.Success
+                    }
+                },
+                resetStaleBootloader = { staleResets++ },
+                interAttemptDelay = {},
+            )
+
+        assertEquals(DfuUploadResult.Success, outcome)
+        assertEquals(
+            listOf(LegacyDfuStreamProfile.NORMAL, LegacyDfuStreamProfile.RECOVERY, LegacyDfuStreamProfile.RECOVERY),
+            sessions,
+        )
+        assertEquals(1, staleResets)
+    }
+
+    /**
+     * Sequence B: drop → stale → RECOVERY drop → stale → RECOVERY success.
+     *
+     * Verifies the run fits within 3 upload attempts + 2 stale resets (the budget), each stale-reset cycle runs exactly
+     * once per drop, and the final RECOVERY attempt succeeds.
+     */
+    @Test
+    fun `drop stale drop stale success fits within 3 uploads and 2 stale resets`() = runTest {
+        val sessions = mutableListOf<LegacyDfuStreamProfile>()
+        var staleResets = 0
+
+        val outcome =
+            runDfuRetryLoop(
+                protocol = DfuProtocolKind.LEGACY,
+                budget = DfuAttemptBudget(LEGACY_SESSION_ATTEMPTS, LEGACY_SESSION_ATTEMPTS),
+                maxStaleResets = MAX_LEGACY_STALE_RESETS,
+                runUploadSession = { profile ->
+                    sessions.add(profile)
+                    // Alternating: attempts 1, 3 drop mid-stream; attempts 2, 4 are stale; attempt 5 succeeds.
+                    when (sessions.size) {
+                        1,
+                        3,
+                        ->
+                            DfuUploadResult.Failure(
+                                LegacyDfuException.MidStreamDisconnect(
+                                    20,
+                                    200,
+                                    "Disconnected",
+                                    lastConfirmedBytes = -1,
+                                ),
+                                false,
+                            )
+
+                        2,
+                        4,
+                        -> DfuUploadResult.Failure(LegacyDfuException.StaleSessionReset(), true)
+
+                        else -> DfuUploadResult.Success
+                    }
+                },
+                resetStaleBootloader = { staleResets++ },
+                interAttemptDelay = {},
+            )
+
+        assertEquals(DfuUploadResult.Success, outcome)
+        // 5 sessions total, all but the first are RECOVERY.
+        assertEquals(5, sessions.size)
+        assertEquals(LegacyDfuStreamProfile.NORMAL, sessions.first())
+        sessions.drop(1).forEach { assertEquals(LegacyDfuStreamProfile.RECOVERY, it) }
+        assertEquals(MAX_LEGACY_STALE_RESETS, staleResets)
+    }
+
+    /**
+     * Repeated stale responses must stop at MAX_LEGACY_STALE_RESETS even though none of them consume an upload attempt
+     * — otherwise a wedged bootloader would loop forever.
+     */
+    @Test
+    fun `repeated stale responses stop at MAX_LEGACY_STALE_RESETS`() = runTest {
+        var sessions = 0
+        var staleResets = 0
+
+        val outcome =
+            runDfuRetryLoop(
+                protocol = DfuProtocolKind.LEGACY,
+                budget = DfuAttemptBudget(LEGACY_SESSION_ATTEMPTS, LEGACY_SESSION_ATTEMPTS),
+                maxStaleResets = MAX_LEGACY_STALE_RESETS,
+                runUploadSession = {
+                    sessions++
+                    DfuUploadResult.Failure(LegacyDfuException.StaleSessionReset(), true)
+                },
+                resetStaleBootloader = { staleResets++ },
+                interAttemptDelay = {},
+            )
+
+        assertIs<DfuUploadResult.Failure>(outcome)
+        assertIs<LegacyDfuException.StaleSessionReset>(outcome.error)
+        // The loop tolerates MAX_LEGACY_STALE_RESETS stale responses that trigger cleanup; the next stale response
+        // terminates the loop.
+        assertEquals(MAX_LEGACY_STALE_RESETS + 1, sessions)
+        assertEquals(MAX_LEGACY_STALE_RESETS, staleResets)
+    }
+
+    /**
+     * Ordinary failures (not MidStreamDisconnect, not StaleSessionReset) consume the upload-attempt budget. After
+     * `attempts` consecutive ordinary failures, the loop exits with the last failure.
+     */
+    @Test
+    fun `ordinary failures consume the three upload attempts`() = runTest {
+        var sessions = 0
+        var staleResets = 0
+
+        val outcome =
+            runDfuRetryLoop(
+                protocol = DfuProtocolKind.LEGACY,
+                budget = DfuAttemptBudget(LEGACY_SESSION_ATTEMPTS, LEGACY_SESSION_ATTEMPTS),
+                maxStaleResets = MAX_LEGACY_STALE_RESETS,
+                runUploadSession = {
+                    sessions++
+                    DfuUploadResult.Failure(DfuException.TransferFailed("ordinary failure"), false)
+                },
+                resetStaleBootloader = { staleResets++ },
+                interAttemptDelay = {},
+            )
+
+        assertIs<DfuUploadResult.Failure>(outcome)
+        assertIs<DfuException.TransferFailed>(outcome.error)
+        assertEquals(LEGACY_SESSION_ATTEMPTS, sessions)
+        assertEquals(0, staleResets, "ordinary failures must NOT trigger stale cleanup")
+    }
+
+    /**
+     * A StaleSessionReset must NOT consume an upload attempt — the cleanup cycle never tried to upload. After the stale
+     * cleanup, the same upload-attempt budget remains, and a subsequent successful attempt succeeds.
+     */
+    @Test
+    fun `stale cleanup does not consume an upload attempt`() = runTest {
+        var sessions = 0
+        var staleResets = 0
+
+        val outcome =
+            runDfuRetryLoop(
+                protocol = DfuProtocolKind.LEGACY,
+                budget = DfuAttemptBudget(LEGACY_SESSION_ATTEMPTS, LEGACY_SESSION_ATTEMPTS),
+                maxStaleResets = MAX_LEGACY_STALE_RESETS,
+                runUploadSession = {
+                    sessions++
+                    if (sessions == 1) {
+                        DfuUploadResult.Failure(LegacyDfuException.StaleSessionReset(), true)
+                    } else {
+                        DfuUploadResult.Success
+                    }
+                },
+                resetStaleBootloader = { staleResets++ },
+                interAttemptDelay = {},
+            )
+
+        assertEquals(DfuUploadResult.Success, outcome)
+        assertEquals(2, sessions, "stale cleanup must not consume an upload attempt — 2 sessions for 1 success")
+        assertEquals(1, staleResets)
+    }
+
+    /**
+     * The profile switches from NORMAL to RECOVERY only after a MidStreamDisconnect. A StaleSessionReset by itself must
+     * NOT switch the profile — the link did not actually drop mid-upload.
+     */
+    @Test
+    fun `NORMAL switches to RECOVERY only after MidStreamDisconnect`() = runTest {
+        val sessions = mutableListOf<LegacyDfuStreamProfile>()
+
+        // Initial stale session, then mid-stream drop, then success — only the post-drop attempt(s) should be RECOVERY.
+        runDfuRetryLoop(
+            protocol = DfuProtocolKind.LEGACY,
+            budget = DfuAttemptBudget(LEGACY_SESSION_ATTEMPTS, LEGACY_SESSION_ATTEMPTS),
+            maxStaleResets = MAX_LEGACY_STALE_RESETS,
+            runUploadSession = { profile ->
+                sessions.add(profile)
+                when (sessions.size) {
+                    1 -> DfuUploadResult.Failure(LegacyDfuException.StaleSessionReset(), true)
+
+                    2 ->
+                        DfuUploadResult.Failure(
+                            LegacyDfuException.MidStreamDisconnect(20, 200, "Disconnected", lastConfirmedBytes = -1),
+                            false,
+                        )
+
+                    else -> DfuUploadResult.Success
+                }
+            },
+            resetStaleBootloader = {},
+            interAttemptDelay = {},
+        )
+
+        // Session 1 (stale) stayed NORMAL; session 2 (drop) was still NORMAL; session 3 (post-drop) switched to
+        // RECOVERY.
+        assertEquals(
+            listOf(LegacyDfuStreamProfile.NORMAL, LegacyDfuStreamProfile.NORMAL, LegacyDfuStreamProfile.RECOVERY),
+            sessions,
+        )
+    }
+
+    /**
+     * An initial StaleSessionReset (no prior drop) keeps NORMAL — the most common case of a device stranded in the
+     * bootloader from a previous interrupted flash.
+     */
+    @Test
+    fun `initial StaleSessionReset keeps NORMAL profile`() = runTest {
+        val sessions = mutableListOf<LegacyDfuStreamProfile>()
+
+        runDfuRetryLoop(
+            protocol = DfuProtocolKind.LEGACY,
+            budget = DfuAttemptBudget(LEGACY_SESSION_ATTEMPTS, LEGACY_SESSION_ATTEMPTS),
+            maxStaleResets = MAX_LEGACY_STALE_RESETS,
+            runUploadSession = { profile ->
+                sessions.add(profile)
+                when (sessions.size) {
+                    1 -> DfuUploadResult.Failure(LegacyDfuException.StaleSessionReset(), true)
+                    else -> DfuUploadResult.Success
+                }
+            },
+            resetStaleBootloader = {},
+            interAttemptDelay = {},
+        )
+
+        sessions.forEach { assertEquals(LegacyDfuStreamProfile.NORMAL, it) }
+    }
+
+    /**
+     * The original mid-stream failure stays visible even when the terminal failure is a later cleanup/transfer error.
+     * The retry loop attaches the first MidStreamDisconnect as a suppressed exception on the surfaced error.
+     */
+    @Test
+    fun `original MidStreamDisconnect stays visible when terminal failure differs`() = runTest {
+        val outcome =
+            runDfuRetryLoop(
+                protocol = DfuProtocolKind.LEGACY,
+                budget = DfuAttemptBudget(LEGACY_SESSION_ATTEMPTS, LEGACY_SESSION_ATTEMPTS),
+                maxStaleResets = MAX_LEGACY_STALE_RESETS,
+                runUploadSession = {
+                    when (it) {
+                        LegacyDfuStreamProfile.NORMAL ->
+                            DfuUploadResult.Failure(
+                                LegacyDfuException.MidStreamDisconnect(
+                                    50,
+                                    200,
+                                    "Disconnected",
+                                    lastConfirmedBytes = -1,
+                                ),
+                                false,
+                            )
+
+                        // All RECOVERY attempts fail with ordinary errors until the budget is exhausted.
+                        else -> DfuUploadResult.Failure(DfuException.TransferFailed("recovery transfer failed"), false)
+                    }
+                },
+                resetStaleBootloader = {},
+                interAttemptDelay = {},
+            )
+
+        assertIs<DfuUploadResult.Failure>(outcome)
+        val terminal = outcome.error
+        assertIs<DfuException.TransferFailed>(terminal)
+        assertEquals(1, terminal.suppressedExceptions.size)
+        val suppressed = terminal.suppressedExceptions.single()
+        assertIs<LegacyDfuException.MidStreamDisconnect>(suppressed)
+        assertEquals(50, suppressed.bytesSent)
+        assertEquals(200, suppressed.totalBytes)
+    }
+
+    /**
+     * Secure DFU never throws the Legacy-only exceptions, so it just runs the `attempts` loop — no stale cleanup, no
+     * profile switching.
+     */
+    @Test
+    fun `Secure protocol runs the attempts loop without stale handling`() = runTest {
+        val sessions = mutableListOf<LegacyDfuStreamProfile>()
+        var staleResets = 0
+
+        val outcome =
+            runDfuRetryLoop(
+                protocol = DfuProtocolKind.SECURE,
+                // Secure budget is always 1/1 — pre==engaged, so engagement promotion is a no-op.
+                budget = DfuAttemptBudget(LIMITED_SESSION_ATTEMPTS, LIMITED_SESSION_ATTEMPTS),
+                maxStaleResets = MAX_LEGACY_STALE_RESETS,
+                runUploadSession = { profile ->
+                    sessions.add(profile)
+                    DfuUploadResult.Failure(DfuException.TransferFailed("secure transfer failed"), false)
+                },
+                resetStaleBootloader = { staleResets++ },
+                interAttemptDelay = {},
+            )
+
+        assertIs<DfuUploadResult.Failure>(outcome)
+        assertIs<DfuException.TransferFailed>(outcome.error)
+        assertEquals(LIMITED_SESSION_ATTEMPTS, sessions.size)
+        // Secure never switches profile even if a MidStreamDisconnect-shaped failure surfaced (it cannot —
+        // MidStreamDisconnect is a LegacyDfuException subtype).
+        sessions.forEach { assertEquals(LegacyDfuStreamProfile.NORMAL, it) }
+        assertEquals(0, staleResets)
+    }
+
+    /** Secure DFU succeeds as soon as the first session succeeds — no fallback, no profile switching. */
+    @Test
+    fun `Secure protocol returns Success on first successful session`() = runTest {
+        val outcome =
+            runDfuRetryLoop(
+                protocol = DfuProtocolKind.SECURE,
+                budget = DfuAttemptBudget(LIMITED_SESSION_ATTEMPTS, LIMITED_SESSION_ATTEMPTS),
+                maxStaleResets = MAX_LEGACY_STALE_RESETS,
+                runUploadSession = { DfuUploadResult.Success },
+                resetStaleBootloader = {},
+                interAttemptDelay = {},
+            )
+
+        assertEquals(DfuUploadResult.Success, outcome)
+    }
+
+    // -----------------------------------------------------------------------
+    // Two-stage budget: pre-engagement probe → engaged promotion
+    // -----------------------------------------------------------------------
+
+    /**
+     * Unknown Legacy pre-engagement connect failure: budget stays at the speculative 1-attempt cap so Secure fallback
+     * stays timely. The probe never engaged, so no promotion occurs.
+     */
+    @Test
+    fun `Unknown Legacy pre-engagement connect failure stops after 1 attempt for timely Secure fallback`() = runTest {
+        var sessions = 0
+        val outcome =
+            runDfuRetryLoop(
+                protocol = DfuProtocolKind.LEGACY,
+                budget = DfuAttemptBudget(LIMITED_SESSION_ATTEMPTS, LEGACY_SESSION_ATTEMPTS),
+                maxStaleResets = MAX_LEGACY_STALE_RESETS,
+                runUploadSession = {
+                    sessions++
+                    DfuUploadResult.Failure(DfuException.ConnectionFailed("connect failed"), false)
+                },
+                resetStaleBootloader = {},
+                interAttemptDelay = {},
+            )
+
+        assertIs<DfuUploadResult.Failure>(outcome)
+        assertEquals(1, sessions, "pre-engagement budget must be 1 so Secure can fall back in time")
+    }
+
+    /**
+     * Unknown Legacy engages then MidStreamDisconnect: budget promotes from 1 → 3 inside the loop, and subsequent
+     * attempts use RECOVERY.
+     */
+    @Test
+    fun `Unknown Legacy engagement during MidStreamDisconnect promotes budget to 3 and uses RECOVERY`() = runTest {
+        val sessions = mutableListOf<LegacyDfuStreamProfile>()
+        val outcome =
+            runDfuRetryLoop(
+                protocol = DfuProtocolKind.LEGACY,
+                budget = DfuAttemptBudget(LIMITED_SESSION_ATTEMPTS, LEGACY_SESSION_ATTEMPTS),
+                maxStaleResets = MAX_LEGACY_STALE_RESETS,
+                runUploadSession = { profile ->
+                    sessions.add(profile)
+                    if (sessions.size == 1) {
+                        // First attempt: engages, then drops mid-stream.
+                        DfuUploadResult.Failure(
+                            LegacyDfuException.MidStreamDisconnect(50, 200, "Disconnected", lastConfirmedBytes = -1),
+                            true,
+                        )
+                    } else {
+                        // Subsequent RECOVERY attempts fail ordinarily to exhaust the promoted budget.
+                        DfuUploadResult.Failure(DfuException.TransferFailed("recovery failed"), true)
+                    }
+                },
+                resetStaleBootloader = {},
+                interAttemptDelay = {},
+            )
+
+        assertIs<DfuUploadResult.Failure>(outcome)
+        assertEquals(3, sessions.size, "promoted engaged budget must allow 3 attempts")
+        assertEquals(LegacyDfuStreamProfile.NORMAL, sessions.first())
+        sessions.drop(1).forEach { assertEquals(LegacyDfuStreamProfile.RECOVERY, it) }
+    }
+
+    /**
+     * Unknown Legacy engages, drop → stale → success: the stale cleanup after engagement does NOT consume the promoted
+     * budget. Three upload sessions total (drop, stale, success) but only one upload-attempt budget slot consumed
+     * before success.
+     */
+    @Test
+    fun `Unknown Legacy stale cleanup after engagement does not consume promoted budget`() = runTest {
+        val sessions = mutableListOf<LegacyDfuStreamProfile>()
+        var staleResets = 0
+        val outcome =
+            runDfuRetryLoop(
+                protocol = DfuProtocolKind.LEGACY,
+                budget = DfuAttemptBudget(LIMITED_SESSION_ATTEMPTS, LEGACY_SESSION_ATTEMPTS),
+                maxStaleResets = MAX_LEGACY_STALE_RESETS,
+                runUploadSession = { profile ->
+                    sessions.add(profile)
+                    when (sessions.size) {
+                        1 ->
+                            DfuUploadResult.Failure(
+                                LegacyDfuException.MidStreamDisconnect(
+                                    50,
+                                    200,
+                                    "Disconnected",
+                                    lastConfirmedBytes = -1,
+                                ),
+                                true,
+                            )
+
+                        2 -> DfuUploadResult.Failure(LegacyDfuException.StaleSessionReset(), true)
+
+                        else -> DfuUploadResult.Success
+                    }
+                },
+                resetStaleBootloader = { staleResets++ },
+                interAttemptDelay = {},
+            )
+
+        assertEquals(DfuUploadResult.Success, outcome)
+        assertEquals(3, sessions.size, "drop+stale+success = 3 sessions; stale cleanup does not consume budget")
+        assertEquals(1, staleResets)
+    }
+
+    /**
+     * A Legacy alternate that engages (e.g. SecureObserved primary fails after engagement → Legacy fallback) receives
+     * the full Legacy engaged budget of 3.
+     */
+    @Test
+    fun `Legacy alternate that engages receives full engaged Legacy budget`() = runTest {
+        val sessions = mutableListOf<LegacyDfuStreamProfile>()
+        val outcome =
+            runDfuRetryLoop(
+                protocol = DfuProtocolKind.LEGACY,
+                // Legacy alternate: pre=1 (probe), engaged=3 (full Legacy budget).
+                budget = DfuAttemptBudget(LIMITED_SESSION_ATTEMPTS, LEGACY_SESSION_ATTEMPTS),
+                maxStaleResets = MAX_LEGACY_STALE_RESETS,
+                runUploadSession = { profile ->
+                    sessions.add(profile)
+                    // Engages immediately, then fails until the budget is exhausted.
+                    DfuUploadResult.Failure(DfuException.TransferFailed("legacy alt failure"), true)
+                },
+                resetStaleBootloader = {},
+                interAttemptDelay = {},
+            )
+
+        assertIs<DfuUploadResult.Failure>(outcome)
+        assertEquals(3, sessions.size, "Legacy alternate that engages must receive full engaged budget")
+    }
+
+    /**
+     * Secure never receives a Legacy-style engaged promotion: its budget carries pre==engaged==1, so even when a Secure
+     * session reports protocolEngaged=true the active attempt cap stays at 1.
+     */
+    @Test
+    fun `Secure budget stays at 1 even when Secure session reports protocolEngaged true`() = runTest {
+        var sessions = 0
+        val outcome =
+            runDfuRetryLoop(
+                protocol = DfuProtocolKind.SECURE,
+                budget = DfuAttemptBudget(LIMITED_SESSION_ATTEMPTS, LIMITED_SESSION_ATTEMPTS),
+                maxStaleResets = MAX_LEGACY_STALE_RESETS,
+                runUploadSession = {
+                    sessions++
+                    // Even with protocolEngaged=true, the Secure budget cannot grow (pre==engaged==1).
+                    DfuUploadResult.Failure(DfuException.TransferFailed("secure engages then fails"), true)
+                },
+                resetStaleBootloader = {},
+                interAttemptDelay = {},
+            )
+
+        assertIs<DfuUploadResult.Failure>(outcome)
+        assertEquals(1, sessions, "Secure budget must stay at 1 even when protocolEngaged=true (no Legacy promotion)")
+    }
+
+    // -----------------------------------------------------------------------
+    // Cleanup policy (cleanupDfuSessionTransport)
+    // -----------------------------------------------------------------------
+
+    /** Fake transport for cleanup-policy tests — records abort/close calls and can inject failures. */
+    private class FakeCleanupTransport(val abortAction: () -> Unit = {}, val closeAction: () -> Unit = {}) :
+        DfuUploadTransport {
+        var abortCalled = false
+        var closeCalled = false
+
+        override val isLowSpeedTransfer: Boolean = false
+
+        override suspend fun connectToDfuMode(): Result<Unit> = Result.success(Unit)
+
+        override suspend fun transferInitPacket(initPacket: ByteArray): Result<Unit> = Result.success(Unit)
+
+        override suspend fun transferFirmware(firmware: ByteArray, onProgress: suspend (Float) -> Unit): Result<Unit> =
+            Result.success(Unit)
+
+        override suspend fun abort() {
+            abortCalled = true
+            abortAction()
+        }
+
+        override suspend fun close() {
+            closeCalled = true
+            closeAction()
+        }
+    }
+
+    @Test
+    fun `cleanup closes transport after abort throws Exception`() = runTest {
+        // Even if abort throws an operational Exception, close must still be called. The exception propagates after
+        // close because the helper's finally guarantees close.
+        val transport = FakeCleanupTransport(abortAction = { throw RuntimeException("abort failed") })
+        assertFailsWith<RuntimeException> {
+            cleanupDfuSessionTransport(transport, completed = false, sessionFailure = null)
+        }
+        assertTrue(transport.abortCalled)
+        assertTrue(transport.closeCalled)
+    }
+
+    @Test
+    fun `cleanup closes transport after abort throws Error`() = runTest {
+        // Error subtypes propagate through the cleanup helper's finally — close still runs.
+        val transport = FakeCleanupTransport(abortAction = { throw AssertionError("abort error") })
+        assertFailsWith<AssertionError> {
+            cleanupDfuSessionTransport(transport, completed = false, sessionFailure = null)
+        }
+        assertTrue(transport.abortCalled)
+        assertTrue(transport.closeCalled)
+    }
+
+    @Test
+    fun `cleanup skips abort after StaleSessionReset but always closes`() = runTest {
+        val transport = FakeCleanupTransport()
+        cleanupDfuSessionTransport(
+            transport,
+            completed = false,
+            sessionFailure = LegacyDfuException.StaleSessionReset(),
+        )
+        assertFalse(transport.abortCalled, "abort must be skipped after StaleSessionReset")
+        assertTrue(transport.closeCalled, "close must always run")
+    }
+
+    @Test
+    fun `cleanup skips abort after MidStreamDisconnect but always closes`() = runTest {
+        val transport = FakeCleanupTransport()
+        cleanupDfuSessionTransport(
+            transport,
+            completed = false,
+            sessionFailure = LegacyDfuException.MidStreamDisconnect(20, 200, "Disconnected", lastConfirmedBytes = -1),
+        )
+        assertFalse(transport.abortCalled, "abort must be skipped after MidStreamDisconnect")
+        assertTrue(transport.closeCalled, "close must always run")
+    }
+
+    @Test
+    fun `cleanup attempts abort after ambiguous ConnectionFailed and always closes`() = runTest {
+        val transport = FakeCleanupTransport()
+        cleanupDfuSessionTransport(
+            transport,
+            completed = false,
+            sessionFailure = DfuException.ConnectionFailed("connect failed"),
+        )
+        assertTrue(transport.abortCalled, "abort must be attempted after ConnectionFailed (link may still be usable)")
+        assertTrue(transport.closeCalled, "close must always run")
+    }
+
+    @Test
+    fun `cleanup skips abort after completed session but always closes`() = runTest {
+        val transport = FakeCleanupTransport()
+        cleanupDfuSessionTransport(transport, completed = true, sessionFailure = null)
+        assertFalse(transport.abortCalled, "abort must be skipped after a completed session")
+        assertTrue(transport.closeCalled, "close must always run")
+    }
+}
diff --git a/feature/firmware/src/commonTest/kotlin/org/meshtastic/feature/firmware/ota/dfu/LegacyDfuTransportTest.kt b/feature/firmware/src/commonTest/kotlin/org/meshtastic/feature/firmware/ota/dfu/LegacyDfuTransportTest.kt
index 01a8eb1a9..7ad73d3f7 100644
--- a/feature/firmware/src/commonTest/kotlin/org/meshtastic/feature/firmware/ota/dfu/LegacyDfuTransportTest.kt
+++ b/feature/firmware/src/commonTest/kotlin/org/meshtastic/feature/firmware/ota/dfu/LegacyDfuTransportTest.kt
@@ -18,11 +18,18 @@
 
 package org.meshtastic.feature.firmware.ota.dfu
 
+import kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.async
+import kotlinx.coroutines.awaitCancellation
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.StateFlow
+import kotlinx.coroutines.test.advanceUntilIdle
+import kotlinx.coroutines.test.runCurrent
 import kotlinx.coroutines.test.runTest
 import org.meshtastic.core.ble.BleCharacteristic
 import org.meshtastic.core.ble.BleConnection
@@ -40,6 +47,7 @@ import org.meshtastic.core.testing.FakeBleWrite
 import kotlin.test.Test
 import kotlin.test.assertContentEquals
 import kotlin.test.assertEquals
+import kotlin.test.assertFailsWith
 import kotlin.test.assertIs
 import kotlin.test.assertTrue
 import kotlin.time.Duration
@@ -308,7 +316,7 @@ class LegacyDfuTransportTest {
         env.transport.transferInitPacket(ByteArray(14)).getOrThrow()
         // Send (PRN_INTERVAL_PACKETS + 1) packets of 20 bytes — guarantees a PRN window fires
         // before the firmware completes, so the under-reported byte count surfaces as a mismatch.
-        val firmwareSize = (LegacyDfuTransport.PRN_INTERVAL_PACKETS + 1) * 20
+        val firmwareSize = (PRN_INTERVAL_PACKETS + 1) * 20
         val result = env.transport.transferFirmware(ByteArray(firmwareSize)) {}
 
         assertTrue(result.isFailure)
@@ -353,7 +361,11 @@ class LegacyDfuTransportTest {
     // Test infrastructure
     // -----------------------------------------------------------------------
 
-    private class TestEnv(val transport: LegacyDfuTransport, val service: AutoRespondingLegacyService) {
+    private class TestEnv(
+        val transport: LegacyDfuTransport,
+        val service: AutoRespondingLegacyService,
+        val connection: FakeBleConnection,
+    ) {
         val responder = service.responder
 
         fun controlPointWrites(): List<FakeBleWrite> =
@@ -363,7 +375,10 @@ class LegacyDfuTransportTest {
             service.delegate.writes.filter { it.characteristic.uuid == LEGACY_DFU_PACKET_UUID }
     }
 
-    private suspend fun createConnectedTransport(mtu: Int? = null): TestEnv {
+    private suspend fun createConnectedTransport(
+        mtu: Int? = null,
+        profile: LegacyDfuStreamProfile = LegacyDfuStreamProfile.NORMAL,
+    ): TestEnv {
         val scanner = FakeBleScanner()
         val fakeConnection = FakeBleConnection().apply { maxWriteValueLength = mtu }
         val autoService = AutoRespondingLegacyService(fakeConnection.service)
@@ -372,21 +387,41 @@ class LegacyDfuTransportTest {
             object : BleConnectionFactory {
                 override fun create(scope: CoroutineScope, tag: String): BleConnection = wrappedConnection
             }
-        val transport = LegacyDfuTransport(scanner, factory, address, Dispatchers.Unconfined)
+        val transport = LegacyDfuTransport(scanner, factory, address, Dispatchers.Unconfined, profile)
 
         scanner.emitDevice(FakeBleDevice(dfuAddress))
         transport.connectToDfuMode().getOrThrow()
-        return TestEnv(transport, autoService)
+        return TestEnv(transport, autoService, fakeConnection)
     }
 
     /**
      * Drives the simulated bootloader response stream. After each `write()` to control point or packet, this service
      * synthesises the appropriate notification(s) on the control-point characteristic so the transport's pending
-     * `awaitResponse` / `awaitPacketReceipt` calls unblock.
+     * `awaitResponse` / `awaitPacketReceiptDuringStream` calls unblock.
      */
     private class AutoRespondingLegacyService(val delegate: FakeBleService) : BleService {
         val responder = LegacyResponder()
 
+        /**
+         * When `true`, the next `write()` to the Legacy Control Point characteristic suspends forever (via
+         * [awaitCancellation]) instead of completing. Used to prove the bounded abort returns within
+         * [LegacyDfuTransport.RESET_WRITE_TIMEOUT] when the device never ACKs the RESET.
+         */
+        var hangOnControlPointWrites: Boolean = false
+
+        /**
+         * When `true`, the next `write()` to the Legacy Control Point characteristic throws an [AssertionError]. Used
+         * to prove [LegacyDfuTransport.abort] (which catches [Exception], not [Throwable]) does not swallow [Error]
+         * subtypes.
+         */
+        var throwErrorOnControlPointWrites: Boolean = false
+
+        /**
+         * When `true`, the next `write()` to the Legacy Control Point characteristic throws a [RuntimeException]. Used
+         * to prove [LegacyDfuTransport.abort] swallows operational [Exception] subtypes (best-effort / non-fatal).
+         */
+        var throwExceptionOnControlPointWrites: Boolean = false
+
         override fun hasCharacteristic(c: BleCharacteristic) = delegate.hasCharacteristic(c)
 
         override fun observe(c: BleCharacteristic): Flow<ByteArray> = delegate.observe(c)
@@ -396,6 +431,15 @@ class LegacyDfuTransportTest {
         override fun preferredWriteType(c: BleCharacteristic): BleWriteType = delegate.preferredWriteType(c)
 
         override suspend fun write(c: BleCharacteristic, data: ByteArray, writeType: BleWriteType) {
+            if (throwErrorOnControlPointWrites && c.uuid == LegacyDfuUuids.CONTROL_POINT) {
+                throw AssertionError("Simulated assertion failure during control point write")
+            }
+            if (throwExceptionOnControlPointWrites && c.uuid == LegacyDfuUuids.CONTROL_POINT) {
+                throw RuntimeException("Simulated link failure during control point write")
+            }
+            if (hangOnControlPointWrites && c.uuid == LegacyDfuUuids.CONTROL_POINT) {
+                awaitCancellation()
+            }
             delegate.write(c, data, writeType)
             val response = responder.onWrite(c.uuid, data) ?: return
             response.forEach { delegate.emitNotification(LegacyDfuUuids.CONTROL_POINT, it) }
@@ -426,13 +470,33 @@ class LegacyDfuTransportTest {
         var scheme: LegacyResponderScheme = LegacyResponderScheme.HappyPath
         var failOnActivateWrite: Boolean = false
 
+        /**
+         * Optional suspend hook invoked after each firmware-packet write with the cumulative byte count, while
+         * [AutoRespondingLegacyService.write] is still in progress. Tests use this to hold a write in flight (e.g. via
+         * [awaitCancellation]) while another coroutine changes connection state to simulate a mid-stream link drop at a
+         * deterministic offset.
+         */
+        var onFirmwarePacketWrite: (suspend (bytesReceived: Long) -> Unit)? = null
+
+        /**
+         * When `true`, packet-receipt notifications are NOT emitted by the simulated bootloader. Used to suspend the
+         * transport inside the PRN wait so a test can drive a disconnect while the wait is in flight.
+         */
+        var suppressPrnNotifications: Boolean = false
+
         private var packetBytesReceived = 0L
         private var packetsSinceLastPrn = 0
         private var firmwareTransferStarted = false
         private var imageSizesWritten = false
         private var expectedFirmwareSize: Int = 0
 
-        fun onWrite(uuid: kotlin.uuid.Uuid, data: ByteArray): List<ByteArray>? = when (uuid) {
+        /**
+         * Effective PRN cadence. Defaults to the NORMAL profile value; updated from the `PACKET_RECEIPT_NOTIF_REQ`
+         * control write so the responder matches whatever interval the transport requested (NORMAL=10, RECOVERY=5).
+         */
+        private var prnInterval: Int = PRN_INTERVAL_PACKETS
+
+        suspend fun onWrite(uuid: kotlin.uuid.Uuid, data: ByteArray): List<ByteArray>? = when (uuid) {
             LegacyDfuUuids.CONTROL_POINT -> handleControlWrite(data)
             LEGACY_DFU_PACKET_UUID -> handlePacketWrite(data)
             else -> null
@@ -454,7 +518,13 @@ class LegacyDfuTransportTest {
                     }
                 }
 
-                LegacyDfuOpcode.PACKET_RECEIPT_NOTIF_REQ -> null
+                LegacyDfuOpcode.PACKET_RECEIPT_NOTIF_REQ -> {
+                    if (data.size >= 3) {
+                        // Parse uint16-LE PRN value the transport just wrote and follow it for receipt cadence.
+                        prnInterval = (data[1].toInt() and 0xFF) or ((data[2].toInt() and 0xFF) shl 8)
+                    }
+                    null
+                }
 
                 LegacyDfuOpcode.RECEIVE_FIRMWARE_IMAGE -> {
                     firmwareTransferStarted = true
@@ -476,7 +546,7 @@ class LegacyDfuTransportTest {
             }
         }
 
-        private fun handlePacketWrite(data: ByteArray): List<ByteArray>? {
+        private suspend fun handlePacketWrite(data: ByteArray): List<ByteArray>? {
             // First packet write is the 12-byte image sizes payload (after START_DFU).
             if (!imageSizesWritten) {
                 imageSizesWritten = true
@@ -494,17 +564,22 @@ class LegacyDfuTransportTest {
             if (firmwareTransferStarted) {
                 packetBytesReceived += data.size
                 packetsSinceLastPrn++
+                onFirmwarePacketWrite?.invoke(packetBytesReceived)
                 val responses = mutableListOf<ByteArray>()
                 val firmwareDone = packetBytesReceived >= expectedFirmwareSize
-                if (packetsSinceLastPrn >= LegacyDfuTransport.PRN_INTERVAL_PACKETS && !firmwareDone) {
+                // Use the cadence the transport actually requested via PACKET_RECEIPT_NOTIF_REQ so a RECOVERY
+                // (prnInterval=5) upload receives receipts at the smaller interval.
+                if (packetsSinceLastPrn >= prnInterval && !firmwareDone) {
                     packetsSinceLastPrn = 0
-                    val reported =
-                        if (scheme == LegacyResponderScheme.PrnUnderReport) {
-                            packetBytesReceived - 1
-                        } else {
-                            packetBytesReceived
-                        }
-                    responses += packetReceipt(reported)
+                    if (!suppressPrnNotifications) {
+                        val reported =
+                            if (scheme == LegacyResponderScheme.PrnUnderReport) {
+                                packetBytesReceived - 1
+                            } else {
+                                packetBytesReceived
+                            }
+                        responses += packetReceipt(reported)
+                    }
                 }
                 if (firmwareDone) {
                     responses += success(LegacyDfuOpcode.RECEIVE_FIRMWARE_IMAGE)
@@ -585,4 +660,236 @@ class LegacyDfuTransportTest {
         override fun maximumWriteValueLength(writeType: BleWriteType): Int? =
             delegate.maximumWriteValueLength(writeType)
     }
+
+    // -----------------------------------------------------------------------
+    // Stream profile selection
+    // -----------------------------------------------------------------------
+
+    @Test
+    fun `NORMAL profile writes PRN request value 10`() = runTest {
+        val env = createConnectedTransport(profile = LegacyDfuStreamProfile.NORMAL)
+        env.responder.scheme = LegacyResponderScheme.HappyPath
+
+        env.transport.transferInitPacket(ByteArray(14)).getOrThrow()
+        env.transport.transferFirmware(ByteArray(40)) {}
+
+        val prnWrite = env.controlPointWrites().single { it.data[0] == LegacyDfuOpcode.PACKET_RECEIPT_NOTIF_REQ }
+        val requested = (prnWrite.data[1].toInt() and 0xFF) or ((prnWrite.data[2].toInt() and 0xFF) shl 8)
+        assertEquals(PRN_INTERVAL_PACKETS, requested)
+    }
+
+    @Test
+    fun `RECOVERY profile writes PRN request value 5`() = runTest {
+        val env = createConnectedTransport(profile = LegacyDfuStreamProfile.RECOVERY)
+        env.responder.scheme = LegacyResponderScheme.HappyPath
+
+        env.transport.transferInitPacket(ByteArray(14)).getOrThrow()
+        env.transport.transferFirmware(ByteArray(40)) {}
+
+        val prnWrite = env.controlPointWrites().single { it.data[0] == LegacyDfuOpcode.PACKET_RECEIPT_NOTIF_REQ }
+        val requested = (prnWrite.data[1].toInt() and 0xFF) or ((prnWrite.data[2].toInt() and 0xFF) shl 8)
+        assertEquals(RECOVERY_PRN_INTERVAL_PACKETS, requested)
+    }
+
+    @Test
+    fun `RECOVERY profile awaits packet receipts at the smaller interval`() = runTest {
+        // RECOVERY prnInterval = 5, so a 100-byte firmware at 20-byte packets fires a PRN exactly once at byte 100
+        // (5 packets × 20B) — but offset == firmware.size at that point means the `offset < firmware.size` gate
+        // suppresses the wait. Use 110 bytes (would need 6 packets at NORMAL=10 to hit the first PRN) so the
+        // first PRN at 100B fires under RECOVERY but never fires under NORMAL within the 110B upload.
+        val env = createConnectedTransport(profile = LegacyDfuStreamProfile.RECOVERY)
+        env.responder.scheme = LegacyResponderScheme.HappyPath
+
+        env.transport.transferInitPacket(ByteArray(14)).getOrThrow()
+        val progress = mutableListOf<Float>()
+        val result = env.transport.transferFirmware(ByteArray(110)) { progress += it }
+
+        assertTrue(result.isSuccess, "transferFirmware failed: ${result.exceptionOrNull()}")
+        // A PRN fires at offset 100 (5 packets × 20B under RECOVERY prnInterval=5); progress must record it during
+        // streaming, then the final 1f at completion. This proves the PRN was consumed mid-stream, not just drained
+        // by the final response wait.
+        assertEquals(listOf(100f / 110f, 1f), progress)
+    }
+
+    @Test
+    fun `NORMAL profile does not await a PRN for a sub-interval upload`() = runTest {
+        // 110-byte firmware at 20-byte packets = 6 packets. NORMAL prnInterval = 10, so no PRN is due before the
+        // upload completes — the auto-responder must NOT emit a PRN, and the transport must NOT call
+        // awaitPacketReceiptDuringStream.
+        val env = createConnectedTransport(profile = LegacyDfuStreamProfile.NORMAL)
+        env.responder.scheme = LegacyResponderScheme.HappyPath
+
+        env.transport.transferInitPacket(ByteArray(14)).getOrThrow()
+        val result = env.transport.transferFirmware(ByteArray(110)) {}
+
+        assertTrue(result.isSuccess, "transferFirmware failed: ${result.exceptionOrNull()}")
+    }
+
+    // -----------------------------------------------------------------------
+    // Mid-stream disconnect
+    // -----------------------------------------------------------------------
+
+    @Test
+    fun `mid-stream disconnect returns typed MidStreamDisconnect with correct byte counts`() = runTest {
+        val env = createConnectedTransport()
+        env.responder.scheme = LegacyResponderScheme.HappyPath
+
+        // Hold the first firmware-packet write in flight so the transfer cannot run through all 200 bytes before
+        // the test injects the disconnect. Production streamOffset has already published 20 before the write, so
+        // the tripwire observer will read bytesSent == 20 when the link drops.
+        val firstPacketWriteInFlight = CompletableDeferred<Unit>()
+        env.responder.onFirmwarePacketWrite = { bytes ->
+            if (bytes >= 20L) {
+                firstPacketWriteInFlight.complete(Unit)
+                awaitCancellation()
+            }
+        }
+
+        env.transport.transferInitPacket(ByteArray(14)).getOrThrow()
+
+        val transferDeferred = async { env.transport.transferFirmware(ByteArray(200)) {} }
+        firstPacketWriteInFlight.await()
+        runCurrent()
+
+        assertTrue(
+            !transferDeferred.isCompleted,
+            "transfer must remain suspended inside the first firmware-packet write",
+        )
+
+        env.connection.simulateRemoteDisconnect()
+        runCurrent()
+
+        val result = transferDeferred.await()
+        val error =
+            assertIs<LegacyDfuException.MidStreamDisconnect>(
+                result.exceptionOrNull(),
+                "expected the outer stream tripwire to classify the disconnect while the first write was in flight",
+            )
+        // First packet write was in flight (streamOffset published as 20 before service.write), so bytesSent == 20.
+        assertEquals(20, error.bytesSent)
+        assertEquals(200, error.totalBytes)
+        // No PRN was received before the drop (first PRN would be at byte 200), so lastConfirmedBytes is -1.
+        assertEquals(-1, error.lastConfirmedBytes)
+        assertTrue(error.connectionState.isNotBlank())
+    }
+
+    /**
+     * A drop during a stream PRN wait must be classified by the OUTER stream-level tripwire as
+     * [LegacyDfuException.MidStreamDisconnect] — NOT as a generic handshake-style [DfuException.ConnectionFailed]. The
+     * stream PRN wait uses [awaitPacketReceiptDuringStream], which deliberately does NOT install its own disconnect
+     * tripwire, so the outer watcher is the sole classifier.
+     */
+    @Test
+    fun `stream-level tripwire classifies a drop during a PRN wait as MidStreamDisconnect`() = runTest {
+        val env = createConnectedTransport()
+        env.responder.scheme = LegacyResponderScheme.HappyPath
+        // Suppress PRNs so the transport suspends inside awaitPacketReceiptDuringStream at the first PRN boundary.
+        env.responder.suppressPrnNotifications = true
+
+        env.transport.transferInitPacket(ByteArray(14)).getOrThrow()
+
+        // First NORMAL PRN boundary = prnInterval (10) × default packet size (20B, MTU not negotiated) = byte 200.
+        // The responder fires this hook synchronously after accounting the 10th packet write, so completing the
+        // deferred proves the stream has written bytes 0..200 and is about to enter awaitPacketReceiptDuringStream —
+        // long before virtual time can reach the 30s COMMAND_TIMEOUT that would otherwise mask the tripwire path.
+        val reachedPrnBoundary = CompletableDeferred<Unit>()
+        env.responder.onFirmwarePacketWrite = { bytes -> if (bytes >= 200L) reachedPrnBoundary.complete(Unit) }
+
+        // Run the transfer asynchronously so the test body can synchronize the disconnect to the suspended PRN wait.
+        val transferDeferred = async { env.transport.transferFirmware(ByteArray(220)) {} }
+        // Wait for the 10th packet write to land, then flush dispatched continuations so the stream coroutine reaches
+        // the suspended receive() inside awaitPacketReceiptDuringStream — without advancing virtual time.
+        reachedPrnBoundary.await()
+        runCurrent()
+
+        // Simulate remote disconnect while the transport is suspended in awaitPacketReceiptDuringStream.
+        env.connection.simulateRemoteDisconnect()
+        runCurrent()
+
+        val result = transferDeferred.await()
+        assertTrue(result.isFailure)
+        val ex = assertIs<LegacyDfuException.MidStreamDisconnect>(result.exceptionOrNull())
+        // Must be the typed MidStreamDisconnect from the outer stream-level tripwire — NOT the generic
+        // ConnectionFailed that an inner handshake-style tripwire would have produced, and NOT a local Timeout.
+        assertEquals(200, ex.bytesSent)
+        assertEquals(220, ex.totalBytes)
+    }
+
+    // -----------------------------------------------------------------------
+    // Bounded RESET (abort)
+    // -----------------------------------------------------------------------
+
+    @Test
+    fun `RESET is written WITH_RESPONSE`() = runTest {
+        val env = createConnectedTransport()
+
+        env.transport.abort()
+
+        val resetWrite = env.controlPointWrites().single { it.data.isNotEmpty() && it.data[0] == LegacyDfuOpcode.RESET }
+        assertEquals(BleWriteType.WITH_RESPONSE, resetWrite.writeType)
+    }
+
+    @Test
+    fun `acknowledged RESET completes and logs acknowledgement`() = runTest {
+        val env = createConnectedTransport()
+
+        env.transport.abort()
+
+        // An acknowledged RESET means the write completed within the timeout; abort must return normally.
+        val resetWrite = env.controlPointWrites().single { it.data.isNotEmpty() && it.data[0] == LegacyDfuOpcode.RESET }
+        assertTrue(resetWrite.data.size == 1, "RESET payload is a single opcode byte")
+    }
+
+    @Test
+    fun `abort returns within RESET_WRITE_TIMEOUT when the RESET write hangs`() = runTest {
+        val env = createConnectedTransport()
+        env.service.hangOnControlPointWrites = true
+
+        // Run abort on the test dispatcher so virtual time advances past RESET_WRITE_TIMEOUT. The hanging write
+        // never completes; withTimeoutOrNull inside abort must return null and abort must return normally.
+        val abortJob = async { env.transport.abort() }
+        advanceUntilIdle()
+        abortJob.await()
+
+        // No RESET write should have been recorded on the delegate — the hang prevented the write from reaching
+        // FakeBleService.write, so an "acknowledged" log would be a lie.
+        val resetWrites =
+            env.controlPointWrites().filter { it.data.isNotEmpty() && it.data[0] == LegacyDfuOpcode.RESET }
+        assertTrue(resetWrites.isEmpty(), "RESET write should not have been recorded when the link hung")
+    }
+
+    @Test
+    fun `abort propagates parent cancellation rather than reporting success`() = runTest {
+        val env = createConnectedTransport()
+        env.service.hangOnControlPointWrites = true
+
+        // parentJob stands in for the caller's scope; cancelling it must propagate through withTimeoutOrNull
+        // (which only swallows its own TimeoutCancellationException, not parent cancellation) and out of abort.
+        val parentJob = Job()
+        val abortDeferred = async(parentJob) { env.transport.abort() }
+
+        // Let abort reach the hanging RESET write.
+        runCurrent()
+        parentJob.cancel()
+        assertFailsWith<CancellationException> { abortDeferred.await() }
+    }
+
+    @Test
+    fun `abort does not swallow Error subtypes from the RESET write`() = runTest {
+        val env = createConnectedTransport()
+        env.service.throwErrorOnControlPointWrites = true
+
+        // abort catches Exception (operational link failures) but lets Error subtypes propagate.
+        assertFailsWith<AssertionError> { env.transport.abort() }
+    }
+
+    @Test
+    fun `abort swallows operational Exception from the RESET write`() = runTest {
+        val env = createConnectedTransport()
+        env.service.throwExceptionOnControlPointWrites = true
+
+        // A RuntimeException from the control-point write is an operational Exception — abort must catch it
+        // (best-effort, non-fatal) and return normally without rethrowing.
+        env.transport.abort()
+    }
 }