Preserve USB mOTA ownership across deadlines and logging handoffs

docs/full_companion_usb_switcher.md

@@ -82,8 +82,9 @@ switched; they remain binary.
 1. Companion initializes the normal USB Binary Companion interface, then
    gives its primary stream to the ASCII terminal before normal loop service
    begins.
-2. While the prompt has no buffered input, the terminal peeks at the next byte.
-   It does not remove that byte.
+2. While the prompt has no buffered input, the terminal skips blank CR/LF
+   delimiters and peeks at the next byte without removing it. A queued binary
+   frame after a blank line is detected before any ASCII banner is emitted.
 3. If the byte is `<`, the terminal temporarily releases the stream and enables
    the existing `ArduinoSerialInterface` frame parser.
 4. The parser consumes the original `<`, the two-byte length, and the payload.
@@ -94,11 +95,21 @@ switched; they remain binary.
 6. If no complete frame arrives within one second, the parser state is reset
    and the ASCII terminal prints a new banner and prompt.
 
-The switcher checks framing, not client identity. Any syntactically complete
-Binary Companion frame confirms binary mode; it does not require the first
+The switcher checks framing, not client identity. A complete Binary Companion
+frame with 1-176 payload bytes confirms binary mode; it does not require the first
 command to be `CMD_APP_START` or `CMD_DEVICE_QUERY`. Normal command validation
 still occurs after the frame parser returns the payload.
 
+Oversized frames are consumed and discarded, never truncated into executable
+commands or counted as proof of a Binary client. ASCII, framed Binary, and mOTA
+input handlers process at most the initial queued-byte count, so a continuously
+sending host cannot keep the radio loop trapped in one input call.
+
+ASCII and mOTA reject an entire line containing an embedded NUL or exceeding its
+line buffer, discarding through the next CR/LF. A valid-looking suffix is not
+executed. Binary payloads may still contain NUL bytes. The immediate manual
+terminal-stop token remains an exception to ordinary line-oriented handling.
+
 The empty-prompt requirement prevents a literal `<` in the middle of a command
 from silently changing modes. A literal `<` typed as the first character does
 start a probe, but the terminal returns after the one-second timeout if no

docs/ota_protocol.md

@@ -831,6 +831,35 @@ CDC/UART receive rings even though the host successfully wrote the complete
 reply. Chunking is internal to the transport and does not change `OP_READ` or
 the `MotaSource` random-access contract.
 
+Each response uses one deadline, starting after request transmission, for its
+magic, header, payload and checksum together. The deadline does not restart
+for each received byte. USB folder sessions use 600 ms; BLE/TCP folder sessions
+use 3000 ms. Each chunk of a larger logical read has its own deadline. A slow
+trickle can therefore fail sooner than before; a maximum 197-byte response
+takes about 17.1 ms on a 115200-baud UART, excluding host processing latency.
+
+A shared USB text/control session explicitly opts into retaining a timed-out
+response's sync/header/payload/checksum position. No subsequent USB request is
+sent until that old response reaches its known checksum boundary, or an actual
+host-session reset clears it. The request supplies the
+payload length: the response itself carries neither a length nor a request ID.
+Late binary bytes therefore cannot become USB `ota folder off` control text.
+A complete reply followed by that ordinary shutdown line still works. If a
+software handoff or failed attach leaves a response outstanding, USB input is
+quarantined until its boundary; commands typed during that wait are discarded.
+Leaving a log-only USB/network handoff preserves that response for the tracked
+consumer rather than discarding its tail as raw input. A deliberate USB mOTA
+entry cancels startup probe and deferred ASCII timers so they cannot reclaim
+the stream or emit terminal text into an active transfer.
+If the sender abandons a partial reply, reconnect the USB host session to
+recover. Bare shutdown text inside a partial payload is inherently ambiguous
+and is not interpreted as control. This assumes one ordered reply per request;
+unsolicited duplicate or reordered binary replies cannot be unambiguously
+separated by this legacy wire format.
+
+Dedicated UART, BLE and TCP sources retain their legacy retry/stale-drain
+behavior; their absolute response deadline still applies to every request.
+
 `OP_DEFLATE_BLOCK` lets a host-folder source perform the optional transport compression without linking an
 encoder into the embedded seeder. `len=0, off=0` queries the exact encoded length; subsequent chunks are at
 most 190 bytes, keeping the total response payload at 192 bytes. The host independently encodes each manifest

examples/companion_radio/main.cpp

@@ -9,6 +9,10 @@
 #include "CompanionBluetooth.h"
 #include "CompanionWireless.h"
 #include "CompanionWiFi.h"
+#if COMPANION_FEATURE_USB_MOTA_SOURCE || COMPANION_FEATURE_BLE_MOTA_SOURCE
+#include <helpers/ota/MotaSourceSerial.h>
+#include <helpers/ota/OtaContext.h>
+#endif
 #if defined(COMPANION_PAIRING_UI_HIL)
 void companionPairingUiHilProbe();
 #endif
@@ -321,9 +325,6 @@ void serviceWioE5HeadlessControls() {
 #endif
 
 #if COMPANION_FEATURE_BLE_MOTA_SOURCE
-#include <helpers/ota/MotaSourceSerial.h>
-#include <helpers/ota/OtaContext.h>
-
 class Nrf52BleMotaSourceControl : public mesh::companion::MotaSourceControl {
 public:
   Nrf52BleMotaSourceControl()
@@ -352,6 +353,7 @@ public:
     _packets_sent_at_start = context.manager.packetsSent();
     _last_packets_sent = 0;
     bluetooth_interface.setMotaStreamActive(true);
+    _source.resetSessionState();  // setActive(true) cleared the old ring generation.
     if (!context.attach_folder_source(
             &_source, mesh::ota::OtaContext::FOLDER_LINK_BLE, "ble",
             reply, reply_size)) {
@@ -636,12 +638,22 @@ static bool usb_logging_reply_state = false;
 #endif
 static mesh::UsbBinaryStartupProbe usb_binary_startup_probe;
 static mesh::UsbAsciiSessionDefault usb_ascii_session_default;
+static bool acceptUsbBinaryStartupFrame(uint32_t completed_at) {
+  // Radio/mesh work and the parser itself can cross the deadline after the
+  // main-loop precheck. Reject before command dispatch or Binary-client proof.
+  if (usb_binary_startup_probe.hasTimedOut(completed_at)) return false;
+  // This first valid frame establishes Binary ownership immediately, not at
+  // the later terminal-service poll. Later frames have no startup deadline.
+  usb_binary_startup_probe.cancel();
+  return true;
+}
 #if COMPANION_FEATURE_USB_MOTA_SOURCE
 static bool usb_mota_mode = false;
 static mesh::UsbMotaEntryOrigin usb_mota_entry_origin =
     mesh::UsbMotaEntryOrigin::BINARY;
 static char usb_mota_line[32];
 static size_t usb_mota_line_len = 0;
+static bool usb_mota_discard_line = false;
 static bool usb_mota_disconnect_armed = false;
 #endif
 
@@ -801,8 +813,12 @@ static void resetUsbMotaMode() {
   usb_mota_mode = false;
   usb_mota_line_len = 0;
   usb_mota_line[0] = 0;
+  usb_mota_discard_line = false;
   usb_mota_disconnect_armed = false;
-  usb_serial_interface.setPassthroughMode(false);
+  // A software owner change is not a transport reset. Retain any late binary
+  // reply and quarantine input until its known boundary is consumed.
+  usb_serial_interface.setPassthroughMode(
+      mesh::ota::OtaContext::serialFolderSource().hasPendingResponse());
 }
 
 static void leaveUsbMotaMode(bool acknowledge) {
@@ -826,13 +842,27 @@ static void leaveUsbMotaMode(bool acknowledge) {
 }
 
 static bool enterUsbMotaMode(mesh::UsbMotaEntryOrigin origin) {
+  // An explicit seeder command selects its own protocol owner, even if it
+  // arrived after a rejected startup frame. Neither a stale Binary deadline
+  // nor deferred default-ASCII restoration may reclaim a live mOTA stream.
+  usb_binary_startup_probe.cancel();
+  usb_ascii_session_default.cancel();
+  auto& source = mesh::ota::OtaContext::serialFolderSource();
+  if (source.hasPendingResponse()) {
+    queueUsbTerminalControlReply("\r\nERR waiting for previous mOTA response\r\n");
+    if (mesh::shouldRestoreAsciiAfterMotaFailure(origin)) enterUsbTerminalMode();
+    return false;
+  }
   cancelUsbSerialOperations();
   mesh::discardUsbTerminalOutput();
+  source.enableSharedTextControl();
+  source.resetSessionState();
   usb_serial_interface.setPassthroughMode(true);
   usb_mota_mode = true;
   usb_mota_entry_origin = origin;
   usb_mota_line_len = 0;
   usb_mota_line[0] = 0;
+  usb_mota_discard_line = false;
   usb_mota_disconnect_armed = isUsbTerminalDataConnected();
 #if defined(ESP32) && defined(ARDUINO_USB_MODE) && ARDUINO_USB_MODE == 1 \
     && defined(ARDUINO_USB_CDC_ON_BOOT) && ARDUINO_USB_CDC_ON_BOOT
@@ -876,14 +906,20 @@ static void serviceUsbMota() {
   }
 #endif
 
-  // SerialMotaSource consumes framed responses synchronously while serving a
-  // block. Bytes left here are host control text, notably motatool's automatic
-  // `ota folder off` on a clean shutdown.
-  Stream& usb_input = mesh::usbCompanionPort();
-  while (usb_input.available()) {
-    int value = usb_input.read();
+  // A failed transaction can leave a late binary response, not just host text.
+  // Retain its known body boundary before recognizing clean shutdown controls.
+  // The static source accessor never allocates/acquires an OTA context here.
+  auto& source = mesh::ota::OtaContext::serialFolderSource();
+  int pending = source.availableControlBytes();
+  while (pending-- > 0) {
+    int value = source.readControlByte();
+    if (value == mesh::ota::SerialMotaSource::RESPONSE_BYTE) continue;
     if (value < 0) break;
     char c = (char)value;
+    if (usb_mota_discard_line) {
+      if (c == '\r' || c == '\n') usb_mota_discard_line = false;
+      continue;
+    }
     if (c == '\r' || c == '\n') {
       if (usb_mota_line_len == 0) continue;
       usb_mota_line[usb_mota_line_len] = 0;
@@ -896,15 +932,41 @@ static void serviceUsbMota() {
       }
       continue;
     }
-    if (usb_mota_line_len < sizeof(usb_mota_line) - 1) {
+    if (c != 0 && usb_mota_line_len < sizeof(usb_mota_line) - 1) {
       usb_mota_line[usb_mota_line_len++] = c;
       usb_mota_line[usb_mota_line_len] = 0;
     } else {
       usb_mota_line_len = 0;
       usb_mota_line[0] = 0;
+      usb_mota_discard_line = true;
     }
   }
 }
+
+static bool serviceUsbMotaResponseBoundary() {
+  if (usb_mota_mode) return false;
+  auto& source = mesh::ota::OtaContext::serialFolderSource();
+  if (!source.hasPendingResponse()) return false;
+  // Failed attach or software handoff must not turn late binary data into
+  // terminal commands or Companion frames. Input typed while waiting is
+  // discarded; only an actual response boundary or host reset ends the wait.
+  if (!usb_serial_interface.isPassthroughMode()) {
+    usb_serial_interface.setPassthroughMode(true);
+  }
+  int pending = source.availableControlBytes();
+  while (pending-- > 0) {
+    if (source.readControlByte() == -1) break;
+    if (!source.hasPendingResponse()) {
+      source.resetSessionState();
+      usb_serial_interface.setPassthroughMode(
+          the_mesh.isTerminalMode() || usb_logging_network_parked
+          || (!mesh::hasDedicatedUsbLoggingPort()
+              && mesh::isUsbLoggingEnabled()));
+      break;
+    }
+  }
+  return true;
+}
 #endif
 
 // Consume an exact TinyUSB owner-task edge, or HWCDC's framework-debounced SOF
@@ -923,6 +985,7 @@ static void resetUsbTerminalHostSession() {
   if (usb_mota_mode) {
     leaveUsbMotaMode(false);
   }
+  mesh::ota::OtaContext::serialFolderSource().resetSessionState();
 #endif
   if (the_mesh.isTerminalMode()) leaveUsbTerminalMode(false);
   mesh::discardUsbTerminalOutput();
@@ -1034,6 +1097,7 @@ static void serviceUsbTerminal() {
   serviceUsbLoggingOwnership(mesh::isUsbLoggingEnabled());
   if (usb_logging_network_parked) return;
 #if COMPANION_FEATURE_USB_MOTA_SOURCE
+  if (serviceUsbMotaResponseBoundary()) return;
   if (usb_mota_mode) {
     serviceUsbMota();
     return;
@@ -1059,24 +1123,31 @@ static void serviceUsbTerminal() {
     return;
   }
 
-  // Companion boots as a useful ASCII terminal. MeshCLI's first framed
-  // command begins with '<'; hand that byte over untouched at an empty prompt.
-  // A malformed or accidental probe times out and restores the terminal.
-  if (!usb_logging_terminal_mode
-      && usb_binary_startup_probe.shouldStart(
-          usb_terminal_line_len == 0, usb_terminal_discard_line,
-          mesh::usbCompanionPort().peek())) {
-    const uint32_t frame_count = usb_serial_interface.getCompletedFrameCount();
-    leaveUsbTerminalMode(false);
-    usb_binary_startup_probe.start(millis(), frame_count);
-    return;
-  }
-
   Stream& usb_input = mesh::usbCompanionPort();
-  if (the_mesh.isTerminalWaitingForInput() && usb_input.available() > 0) {
-    enterUsbTerminalMode();
-  }
-  while (usb_input.available()) {
+  bool blank_input = false;
+  int pending = usb_input.available();
+  while (pending-- > 0) {
+    // A leftover CR/LF (for example after mOTA shutdown) must not hide a
+    // following frame or cause a banner to poison its first binary reply.
+    if (usb_terminal_line_len == 0 && !usb_terminal_discard_line) {
+      const int next_byte = usb_input.peek();
+      if (next_byte == '\r' || next_byte == '\n') {
+        usb_input.read();
+        blank_input = true;
+        continue;
+      }
+      // Re-check at each empty prompt, before consuming the '<' byte. A
+      // malformed or accidental probe still returns to ASCII after one second.
+      if (!usb_logging_terminal_mode
+          && usb_binary_startup_probe.shouldStart(true, false, next_byte)) {
+        const uint32_t frame_count = usb_serial_interface.getCompletedFrameCount();
+        leaveUsbTerminalMode(false);
+        usb_binary_startup_probe.start(millis(), frame_count);
+        return;
+      }
+    }
+    if (the_mesh.isTerminalWaitingForInput()) enterUsbTerminalMode();
+
     int value = usb_input.read();
     if (value < 0) break;
     char c = (char)value;
@@ -1089,6 +1160,15 @@ static void serviceUsbTerminal() {
       continue;
     }
 
+    if (c == 0) {
+      // The handlers consume C strings. Reject the whole malformed line, not
+      // an apparently valid mutating command before an embedded NUL.
+      clearUsbTerminalLine();
+      usb_terminal_discard_line = true;
+      usbTerminalOutput().print("\r\n  ERROR: invalid command\r\n");
+      continue;
+    }
+
     if (c == '\b' || c == 0x7F) {
       if (usb_terminal_line_len > 0) {
         usb_terminal_line_len = mesh::cli::eraseLastTerminalInput(
@@ -1122,9 +1202,9 @@ static void serviceUsbTerminal() {
         return;
       }
 #endif
-      // On dual CDC hardware only CDC1 is the logging-reader endpoint.
-      if (!mesh::hasDedicatedUsbLoggingPort()
-          && strlen(usb_terminal_line) == usb_terminal_line_len)
+      // NUL-containing lines were discarded before reaching this point. On
+      // dual CDC hardware only CDC1 is the logging-reader endpoint.
+      if (!mesh::hasDedicatedUsbLoggingPort())
         mesh::noteUsbLoggingStatsCommand(usb_terminal_line);
       the_mesh.handleTerminalCommand(usb_terminal_line);
       clearUsbTerminalLine();
@@ -1166,6 +1246,9 @@ static void serviceUsbTerminal() {
       return;
     }
   }
+  // Enter alone should still reveal the otherwise silent ASCII session.
+  if (blank_input && usb_input.available() == 0
+      && the_mesh.isTerminalWaitingForInput()) enterUsbTerminalMode();
 }
 
 static void serviceUsbLoggingOwnership(bool logging_enabled) {
@@ -1197,11 +1280,21 @@ static void serviceUsbLoggingOwnership(bool logging_enabled) {
       && logging_action != mesh::UsbLoggingTerminalAction::PARK_USB_FOR_NETWORK) {
     // Bytes typed while USB was log-only belong to neither the new ASCII CLI
     // nor a new Binary client. Bound the discard to the existing RX snapshot.
+#if COMPANION_FEATURE_USB_MOTA_SOURCE
+    const bool response_pending =
+        mesh::ota::OtaContext::serialFolderSource().hasPendingResponse();
+#else
+    const bool response_pending = false;
+#endif
+    // A timed-out mOTA transaction still owns its exact response boundary.
+    // Raw reads would consume its tail without advancing that tracker, leaving
+    // the stream quarantined forever. Let its bounded consumer drain instead.
     Stream& input = mesh::usbCompanionPort();
-    int pending = input.available();
+    int pending = response_pending ? 0 : input.available();
     while (pending-- > 0) input.read();
     usb_logging_network_parked = false;
-    usb_serial_interface.setPassthroughMode(the_mesh.isTerminalMode());
+    usb_serial_interface.setPassthroughMode(
+        the_mesh.isTerminalMode() || response_pending);
   }
   switch (logging_action) {
     case mesh::UsbLoggingTerminalAction::PARK_USB_FOR_NETWORK:
@@ -3347,6 +3440,7 @@ void setup() {
 #endif
   // keep frames intact and pace the contact stream when the host is slow
   usb_serial_interface.enableFlowControl(true);
+  usb_serial_interface.setReceiveFrameCheck(acceptUsbBinaryStartupFrame);
 #if defined(ESP32) && defined(ARDUINO_USB_MODE) && ARDUINO_USB_MODE == 1 \
     && defined(ARDUINO_USB_CDC_ON_BOOT) && ARDUINO_USB_CDC_ON_BOOT
   // The ESP32 USB-Serial-JTAG peripheral (HWCDC) has no DTR concept at all.
@@ -3485,6 +3579,9 @@ void loop() {
 #endif
 #if defined(ENABLE_USB_INTERFACE)
   expireUsbBinaryStartupProbeBeforeDispatch();
+#if COMPANION_FEATURE_USB_MOTA_SOURCE
+  serviceUsbMotaResponseBoundary();
+#endif
 #endif
   the_mesh.loop();
 #if defined(NRF52_PLATFORM) && defined(EXTRAFS) && !defined(QSPIFLASH)

src/helpers/esp32/WiFiOtaSeeder.cpp

@@ -115,6 +115,7 @@ void WiFiOtaSeeder::loop() {
   }
 
   seeder_client = incoming;
+  seeder_source.resetSessionState();  // The old socket cannot supply this session's reply.
   if (!context.attach_folder_source(&seeder_source, OtaContext::FOLDER_LINK_TCP,
                                     "tcp", attach_reply, sizeof(attach_reply))) {
     seeder_client.stop();

src/helpers/ota/MotaSourceSerial.cpp

@@ -1,23 +1,122 @@
 #include "MotaSourceSerial.h"
 #include "MotaSeederProto.h"
 #include "OtaByteIO.h"
+#include <limits.h>
 #include <string.h>
 
 namespace mesh {
 namespace ota {
 
-bool SerialMotaSource::readByteT(uint8_t& b) {
-  uint32_t t0 = millis();
-  while ((millis() - t0) < _to) {
-    int c = _io.read();
-    if (c >= 0) { b = (uint8_t)c; return true; }
+void SerialMotaSource::resetSessionState() {
+  _response_pending = false;
+  _response_stage = SYNC_FIRST;
+  _response_payload_len = _response_remaining = 0;
+  _control_offset = _control_len = 0;
+  _control_discard_line = false;
+}
+
+void SerialMotaSource::queueControlByte(uint8_t b) {
+  const bool newline = b == '\r' || b == '\n';
+  if (_control_discard_line) {
+    if (!newline) return;
+    _control_discard_line = false;
+  }
+  if (_control_len == sizeof(_control_queue)) {
+    // Invalidate any prefix already handed to the ASCII owner, then discard
+    // through the real line ending. Never expose an overlong line's suffix.
+    _control_offset = 0;
+    _control_len = 1;
+    _control_queue[0] = 0;
+    _control_discard_line = !newline;
+    if (!newline) return;
+  }
+  _control_queue[(_control_offset + _control_len) % sizeof(_control_queue)] = b;
+  ++_control_len;
+}
+
+int SerialMotaSource::takeControlByte() {
+  if (_control_len == 0) return -1;
+  const uint8_t value = _control_queue[_control_offset];
+  _control_offset = (_control_offset + 1) % sizeof(_control_queue);
+  --_control_len;
+  return value;
+}
+
+int SerialMotaSource::readTrackedByte() {
+  const int value = _io.read();
+  if (value < 0) return value;
+  if (!_shared_text_control) return value;
+  const uint8_t b = (uint8_t)value;
+  if (!_response_pending) {
+    queueControlByte(b);
+    return value;
+  }
+  switch (_response_stage) {
+    case SYNC_FIRST:
+      if (b == MOTA_SEEDER_RSP_MAGIC0) _response_stage = SYNC_SECOND;
+      else queueControlByte(b);
+      break;
+    case SYNC_SECOND:
+      if (b == MOTA_SEEDER_RSP_MAGIC1) {
+        _response_stage = RESPONSE_OP;
+      } else {
+        queueControlByte(MOTA_SEEDER_RSP_MAGIC0);
+        if (b != MOTA_SEEDER_RSP_MAGIC0) {
+          queueControlByte(b);
+          _response_stage = SYNC_FIRST;
+        }
+      }
+      break;
+    case RESPONSE_OP:
+      _response_stage = RESPONSE_STATUS;
+      break;
+    case RESPONSE_STATUS:
+      _response_remaining = b == MS_STATUS_OK ? _response_payload_len : 0;
+      _response_stage = _response_remaining ? RESPONSE_PAYLOAD : RESPONSE_CHECKSUM;
+      break;
+    case RESPONSE_PAYLOAD:
+      if (--_response_remaining == 0) _response_stage = RESPONSE_CHECKSUM;
+      break;
+    case RESPONSE_CHECKSUM:
+      // The request determines payload length, even when checksum/op fails.
+      // Only this full wire boundary, or an explicit owner reset, permits a
+      // new request; a late body tail can never become ASCII control text.
+      _response_pending = false;
+      _response_stage = SYNC_FIRST;
+      break;
+  }
+  return value;
+}
+
+int SerialMotaSource::availableControlBytes() {
+  const int available = _io.available();
+  if (available <= 0) return _control_len;
+  return available > INT_MAX - _control_len ? INT_MAX : available + _control_len;
+}
+
+int SerialMotaSource::readControlByte() {
+  if (!_shared_text_control) return _io.read();
+  if (_control_len != 0) return takeControlByte();
+  if (readTrackedByte() < 0) return -1;
+  return _control_len ? takeControlByte() : RESPONSE_BYTE;
+}
+
+bool SerialMotaSource::readByteT(uint8_t& b, uint32_t started) {
+  while ((uint32_t)(millis() - started) < _to) {
+    int c = readTrackedByte();
+    if (c >= 0) {
+      // A read or scheduler tick can cross the deadline after the precheck.
+      if ((uint32_t)(millis() - started) >= _to) return false;
+      b = (uint8_t)c;
+      return true;
+    }
     delay(1);  // let BLE/WiFi callbacks deliver the response
   }
   return false;
 }
 
-bool SerialMotaSource::readExact(uint8_t* b, uint32_t n) {
-  for (uint32_t i = 0; i < n; i++) if (!readByteT(b[i])) return false;
+bool SerialMotaSource::readExact(uint8_t* b, uint32_t n, uint32_t started) {
+  for (uint32_t i = 0; i < n; i++) if (!readByteT(b[i], started)) return false;
   return true;
 }
 
@@ -25,7 +124,13 @@ bool SerialMotaSource::readExact(uint8_t* b, uint32_t n) {
 // checksum, then scans for the response magic and validates op+status+checksum before delivering payload.
 bool SerialMotaSource::txn(uint8_t op, const uint8_t* args, uint8_t arglen,
                            uint8_t* payload, uint32_t payload_len) {
-  while (_io.read() >= 0) {}                       // drop any stale/partial bytes before a fresh request
+  // Drop the currently queued stale/partial input only. A continuously writing
+  // host cannot keep the primary mesh loop trapped before the request.
+  int pending = _io.available();
+  while (pending-- > 0) if (readTrackedByte() < 0) break;
+  // Shared text ownership must never overwrite an incomplete old response's
+  // body length. The wire has no request ID or response payload length.
+  if (_response_pending) return false;
   if (arglen > 7) return false;                    // largest source request is READ(idx,off,len)
   uint8_t xs = op;
   for (uint8_t i = 0; i < arglen; i++) xs ^= args[i];
@@ -37,13 +142,19 @@ bool SerialMotaSource::txn(uint8_t op, const uint8_t* args, uint8_t arglen,
   if (arglen) { memcpy(frame + frame_len, args, arglen); frame_len += arglen; }
   frame[frame_len++] = xs;
   if (_io.write(frame, frame_len) != frame_len) return false;
+  if (_shared_text_control) {
+    _response_pending = true;
+    _response_stage = SYNC_FIRST;
+    _response_payload_len = payload_len;
+    _response_remaining = 0;
+  }
   if (_write_policy == MotaStreamWritePolicy::FlushTransmit) _io.flush();
 
   // scan for response magic 'm' 's' (tolerate leading noise)
-  uint32_t t0 = millis(); bool got = false;
+  const uint32_t t0 = millis(); bool got = false;
   uint8_t prev = 0;
   while ((millis() - t0) < _to) {
-    int c = _io.read();
+    int c = readTrackedByte();
     if (c < 0) { delay(1); continue; }
     if (prev == MOTA_SEEDER_RSP_MAGIC0 && (uint8_t)c == MOTA_SEEDER_RSP_MAGIC1) { got = true; break; }
     prev = (uint8_t)c;
@@ -51,16 +162,16 @@ bool SerialMotaSource::txn(uint8_t op, const uint8_t* args, uint8_t arglen,
   if (!got) return false;
 
   uint8_t hdr[2];
-  if (!readExact(hdr, 2)) return false;            // op, status
+  if (!readExact(hdr, 2, t0)) return false;        // op, status
   if (hdr[0] != op) return false;
   uint8_t rxs = (uint8_t)(MOTA_SEEDER_RSP_MAGIC0 ^ MOTA_SEEDER_RSP_MAGIC1) ^ hdr[0] ^ hdr[1];
   bool ok = (hdr[1] == MS_STATUS_OK);
   if (ok && payload_len) {
-    if (!readExact(payload, payload_len)) return false;
+    if (!readExact(payload, payload_len, t0)) return false;
     for (uint32_t i = 0; i < payload_len; i++) rxs ^= payload[i];
   }
   uint8_t xsum;
-  if (!readByteT(xsum)) return false;
+  if (!readByteT(xsum, t0)) return false;
   if (xsum != rxs) return false;                   // corrupt frame -> caller retries
   return ok;
 }

src/helpers/ota/MotaSourceSerial.h

@@ -7,8 +7,10 @@
 // A MotaSource backed by a host "mota-seeder" daemon over a dedicated framed Stream, such as a spare
 // UART, USB CDC, or TCP client. The device pulls catalog + bytes on demand (MotaSeederProto.h); the folder
 // image is never held on the device - it streams through. Do not share the Stream with a text CLI because
-// command/log text would collide with the binary framing. Reads block up to `timeout_ms`; keep the host
-// endpoint responsive so a round-trip does not stall the primary transfer.
+// command/log text would collide with the binary framing. Each request's entire
+// response (sync, header, payload, checksum) has one `timeout_ms` deadline after
+// request transmission. Larger reads use multiple bounded requests. Keep the
+// host responsive so a round-trip does not stall the primary transfer.
 
 namespace mesh {
 namespace ota {
@@ -25,12 +27,40 @@ public:
   bool    read_deflated_block(uint8_t idx, uint16_t block, uint8_t* buf,
                               uint16_t cap, uint16_t* len) override;
 
+  // Only shared USB text/control ownership opts into retained boundaries.
+  // Dedicated UART/BLE/TCP keep their legacy retry and stale-drain behavior.
+  void enableSharedTextControl() { _shared_text_control = true; }
+  // A shared USB owner consumes only out-of-response control bytes. One call
+  // reads at most one underlying byte; RESPONSE_BYTE means binary was consumed.
+  static constexpr int RESPONSE_BYTE = -2;
+  int availableControlBytes();
+  int readControlByte();
+  bool hasPendingResponse() const { return _response_pending; }
+  void resetSessionState();
+
 private:
   // Send a request (op+args) and read its response header; on OK, `payload` (if non-null) receives
   // `payload_len` bytes. Returns true iff a well-formed OK response for `op` arrived in time.
   bool txn(uint8_t op, const uint8_t* args, uint8_t arglen, uint8_t* payload, uint32_t payload_len);
-  bool readByteT(uint8_t& b);          // one byte within the timeout
-  bool readExact(uint8_t* b, uint32_t n);
+  bool readByteT(uint8_t& b, uint32_t started); // within the shared response deadline
+  bool readExact(uint8_t* b, uint32_t n, uint32_t started);
+  int readTrackedByte();
+  void queueControlByte(uint8_t b);
+  int takeControlByte();
+
+  enum ResponseStage : uint8_t {
+    SYNC_FIRST, SYNC_SECOND, RESPONSE_OP, RESPONSE_STATUS,
+    RESPONSE_PAYLOAD, RESPONSE_CHECKSUM,
+  };
+  bool _shared_text_control = false;
+  bool _response_pending = false;
+  ResponseStage _response_stage = SYNC_FIRST;
+  uint32_t _response_payload_len = 0;
+  uint32_t _response_remaining = 0;
+  uint8_t _control_queue[64] = {};
+  uint8_t _control_offset = 0;
+  uint8_t _control_len = 0;
+  bool _control_discard_line = false;
 
   Stream&  _io;
   uint32_t _to;

src/helpers/ota/OtaContext.h

@@ -502,13 +502,16 @@ struct OtaContext {
   // Attach/detach an external folder of `.mota` served by a host daemon over the seeder UART (the node
   // then advertises + relays them alongside its own fw). Only built when OTA_FOLDER_SERIAL is configured.
 #if defined(OTA_FOLDER_SERIAL)
-  bool attach_folder(char* msg, size_t cap) {
+  static SerialMotaSource& serialFolderSource() {
     static SerialMotaSource src(OTA_FOLDER_SERIAL_STREAM,
                                 OTA_FOLDER_SERIAL_WRITE_POLICY, 600);
+    return src;
+  }
+  bool attach_folder(char* msg, size_t cap) {
 #ifdef OTA_FOLDER_SERIAL_BEGIN
     OTA_FOLDER_SERIAL_STREAM.begin(OTA_FOLDER_SERIAL_BAUD);     // dedicated UART; console is already up
 #endif
-    return attach_folder_source(&src, FOLDER_LINK_SERIAL, "serial", msg, cap);
+    return attach_folder_source(&serialFolderSource(), FOLDER_LINK_SERIAL, "serial", msg, cap);
   }
 #endif
   void detach_folder(bool preserve_capture = false) {

test/fixtures/shared_mota_queue/test_shared_mota_queue.cpp

@@ -85,7 +85,12 @@ Log& usbLoggingPort() { static Log log; return log; }
 namespace ota {
 class SerialMotaSource : public ::Folder {
 public:
+  inline static unsigned session_resets = 0;
   SerialMotaSource(int, MotaStreamWritePolicy, uint32_t) {}
+  void resetSessionState() {
+    assert(::bluetooth_interface.active);
+    ++session_resets;
+  }
 };
 } }
 #include "ble_control_under_test.h"
@@ -225,10 +230,12 @@ int main() {
 #else
   Nrf52BleMotaSourceControl ble;
   assert(!ble.start(reply, sizeof reply)); // no subscription, no storage loan
+  assert(SerialMotaSource::session_resets == 0);
   assert(!ota_context_if_active() && q.buffer.capacity() == 256);
   for (int disconnect = 0; disconnect < 2; ++disconnect) {
     bluetooth_interface.ready = true;
     assert(ble.start(reply, sizeof reply));
+    assert(SerialMotaSource::session_resets == static_cast<unsigned>(disconnect + 1));
     assert(ble.status().attached && q.buffer.capacity() == 128);
     assert(ble.status().offered == 1 && ble.status().advertised == 1);
     if (disconnect) {
@@ -255,6 +262,7 @@ int main() {
                                         "USB", reply, sizeof reply));
   bluetooth_interface.ready = true;
   assert(!ble.start(reply, sizeof reply));
+  assert(SerialMotaSource::session_resets == 2);
   assert(ble.status().another_link_active);
   assert(ble.stop(reply, sizeof reply));
   ota_release_context_if_idle(false);

test/fixtures/shared_mota_queue/test_wifi_shared_queue.h

@@ -115,6 +115,36 @@ static void test_wifi_shared_queue(Queue& q, CheckMessages check_messages) {
     packets.clear();
   }
 
+  // TCP keeps legacy retries on the same connection. A genuinely new socket
+  // starts its next catalog handshake without old transport state.
+  WiFi.connected = true;
+  auto stalled = connect_host();
+  WiFiOtaSeeder::loop();
+  assert(WiFiOtaSeeder::isAttached() && ota_ctx().manager.servedCount() == 1);
+  stalled->respond = [](const uint8_t* request, size_t) {
+    assert(request[2] == MS_OP_COUNT);
+    return std::vector<uint8_t>{'m', 's', MS_OP_COUNT, MS_STATUS_OK};
+  };
+  const unsigned before_timeout = stalled->requests;
+  ota_ctx().manager.refresh_sources();
+  assert(stalled->requests == before_timeout + 1);
+  assert(ota_ctx().manager.servedCount() == 0);
+  WiFiOtaSeeder::loop();
+  ota_ctx().manager.refresh_sources();
+  assert(stalled->requests == before_timeout + 2);
+  stalled->connected = false;
+  WiFiOtaSeeder::loop();
+  ota_release_context_if_idle(true);
+  auto replacement = connect_host();
+  WiFiOtaSeeder::loop();
+  assert(replacement->requests >= 2 && WiFiOtaSeeder::isAttached());
+  assert(ota_ctx().manager.servedCount() == 1);
+  WiFiOtaSeeder::stop();
+  ota_release_context_if_idle(true);
+  assert(!ota_context_if_active() && q.buffer.capacity() == 256);
+  check_messages();
+  packets.clear();
+
   // An incoming TCP client and listener stop must not detach a USB source.
   char reply[160];
   assert(ota_acquire_context(reply, sizeof reply));

test/test_companion_ota_includes.py

@@ -0,0 +1,94 @@
+#!/usr/bin/env python3
+"""Compile the production Companion include block without transitive OTA headers."""
+
+from pathlib import Path
+import subprocess
+import tempfile
+import unittest
+
+
+ROOT = Path(__file__).resolve().parents[1]
+MAIN = ROOT / "examples/companion_radio/main.cpp"
+
+
+class CompanionOtaIncludeTest(unittest.TestCase):
+    def compile_include_block(self, flags, usb, ble):
+        # Preserve the actual production includes and their feature guards.
+        # Unlike handler-extraction tests, this harness does not inject either
+        # OTA header. MyMesh intentionally lacks its optional queue-dependent
+        # OtaContext include, just like the Heltec V4 Full configuration.
+        prefix = MAIN.read_text().split(
+            "// Believe it or not, this std C function is busted on some platforms!",
+            1,
+        )[0]
+        self.assertIn('#include "MyMesh.h"', prefix)
+        with tempfile.TemporaryDirectory(prefix="meshcore-companion-includes-") as temp:
+            directory = Path(temp)
+            for header in (
+                "helpers/ui/StartupScreen.h", "helpers/ui/DisplayPowerSettings.h",
+                "helpers/UsbLoggingWatchdog.h", "helpers/UsbLoggingClientActivity.h",
+                "Mesh.h", "helpers/BluetoothMac.h", "CompanionBluetooth.h",
+                "CompanionWireless.h", "CompanionWiFi.h", "esp_bt.h", "esp_pm.h",
+                "esp_sleep.h", "esp_system.h", "esp_heap_caps.h",
+            ):
+                path = directory / header
+                path.parent.mkdir(parents=True, exist_ok=True)
+                path.write_text("#pragma once\n")
+            (directory / "MyMesh.h").write_text(
+                '#pragma once\n#include "CompanionFeatures.h"\n'
+            )
+            if not (usb or ble):
+                # Prove that small/non-OTA images do not gain these dependencies.
+                for header in ("MotaSourceSerial.h", "OtaContext.h"):
+                    path = directory / "helpers/ota" / header
+                    path.parent.mkdir(parents=True, exist_ok=True)
+                    path.write_text('#error "OTA headers must remain feature guarded"\n')
+            checks = (
+                "static_assert(COMPANION_FEATURE_USB_MOTA_SOURCE == %d, \"USB profile\");\n"
+                "static_assert(COMPANION_FEATURE_BLE_MOTA_SOURCE == %d, \"BLE profile\");\n"
+            ) % (usb, ble)
+            if usb or ble:
+                checks += (
+                    "static_assert(sizeof(mesh::ota::OtaContext) != 0, \"context declared\");\n"
+                    "static_assert(mesh::ota::SerialMotaSource::RESPONSE_BYTE == -2, "
+                    "\"source declared\");\n"
+                )
+            if usb:
+                checks += (
+                    "mesh::ota::SerialMotaSource& source_under_test() {\n"
+                    "  return mesh::ota::OtaContext::serialFolderSource();\n}\n"
+                )
+            source = directory / "includes.cpp"
+            source.write_text(
+                "class Stream;\nStream& include_test_serial_port();\n"
+                "#define OTA_FOLDER_SERIAL_STREAM include_test_serial_port()\n"
+                + prefix + checks
+            )
+            result = subprocess.run([
+                "c++", "-std=c++17", "-fsyntax-only", *flags,
+                "-I", temp, "-I", str(ROOT / "src"),
+                "-I", str(ROOT / "test/mocks"),
+                "-I", str(ROOT / "examples/companion_radio"), str(source),
+            ], text=True, capture_output=True)
+            self.assertEqual(result.returncode, 0, result.stderr)
+
+    def test_usb_only_full_esp32_declares_actual_ota_types(self):
+        self.compile_include_block([
+            "-DESP32_PLATFORM=1", "-DCOMPANION_RADIO_FULL=1", "-DBLE_PIN_CODE=123456",
+            "-DWIFI_SSID=\"\"", "-DENABLE_USB_INTERFACE=1", "-DENABLE_OTA=1",
+            "-DOTA_FOLDER_SERIAL=1", "-DOTA_SEEDER_ONLY=1",
+        ], usb=1, ble=0)
+
+    def test_ble_only_full_nrf52_declares_actual_ota_types(self):
+        self.compile_include_block([
+            "-DNRF52_PLATFORM=1", "-DCOMPANION_RADIO_FULL=1", "-DBLE_PIN_CODE=123456",
+            "-DENABLE_OTA=1", "-DOTA_SEEDER_ONLY=1",
+            "-DCOMPANION_FEATURE_BLE_MOTA_SOURCE=1",
+        ], usb=0, ble=1)
+
+    def test_non_ota_image_keeps_ota_headers_out(self):
+        self.compile_include_block(["-DSTM32_PLATFORM=1"], usb=0, ble=0)
+
+
+if __name__ == "__main__":
+    unittest.main()

test/test_companion_usb_input.py

@@ -0,0 +1,559 @@
+#!/usr/bin/env python3
+"""Run actual Companion ASCII/probe and mOTA input handlers on host streams."""
+
+from pathlib import Path
+import subprocess
+import tempfile
+import unittest
+
+from test_replay_reset_integration import extract_braced
+
+
+ROOT = Path(__file__).resolve().parents[1]
+MAIN = (ROOT / "examples/companion_radio/main.cpp").read_text()
+CONTEXT = (ROOT / "src/helpers/ota/OtaContext.h").read_text()
+
+
+HARNESS = r'''
+#include <cassert>
+#include <algorithm>
+#include <cstdio>
+#include <cstring>
+#include <deque>
+#include <string>
+#include <vector>
+#include "helpers/ArduinoSerialInterface.h"
+#include "helpers/ota/MotaSourceSerial.h"
+#include "helpers/UsbAsciiBinarySwitch.h"
+#include "helpers/CLICommandUtils.h"
+#define COMPANION_FEATURE_USB_MOTA_SOURCE 1
+#define COMPANION_FEATURE_NETWORK_TERMINAL 0
+#define MESH_USB_LOGGING_AVAILABLE 1
+struct Input : Stream {
+  std::deque<uint8_t> bytes;
+  std::string output;
+  std::deque<std::vector<uint8_t>> responses;
+  unsigned source_requests=0;
+  bool replenish=false;
+  uint32_t advance_per_read=0;
+  unsigned reads=0;
+  int available() override { return static_cast<int>(bytes.size()); }
+  int read() override {
+    if(bytes.empty()) return -1;
+    uint8_t result=bytes.front(); bytes.pop_front(); ++reads;
+    g_mock_millis+=advance_per_read;
+    if(replenish) bytes.push_back('x');
+    return result;
+  }
+  int peek() override { return bytes.empty()?-1:bytes.front(); }
+  int availableForWrite() override { return 4096; }
+  size_t write(uint8_t value) override { output+=char(value); return 1; }
+  size_t write(const uint8_t* p,size_t n) override {
+    if(n>=4 && p[0]=='M' && p[1]=='S') {
+      ++source_requests;
+      if(!responses.empty()) {
+        for(uint8_t value:responses.front()) bytes.push_back(value);
+        responses.pop_front();
+      }
+    }
+    output.append(reinterpret_cast<const char*>(p),n); return n;
+  }
+  void push(const char* p) { while(*p) bytes.push_back(uint8_t(*p++)); }
+  void frame(uint8_t tail=3) {
+    const uint8_t frame[]={'<',2,0,0x16,tail};
+    for(uint8_t c : frame) bytes.push_back(c);
+  }
+} input;
+#define OTA_FOLDER_SERIAL_STREAM input
+#define OTA_FOLDER_SERIAL_WRITE_POLICY mesh::ota::MotaStreamWritePolicy::NoFlush
+namespace mesh { namespace ota {
+struct OtaContext {
+@SOURCE_ACCESSOR@
+};
+} }
+ArduinoSerialInterface usb_serial_interface;
+static Stream& usbTerminalOutput() { return input; }
+static bool connected=true, logging=false, dedicated=false;
+static std::vector<std::string> stats_commands;
+namespace mesh {
+bool isUsbLoggingEnabled() { return logging; }
+bool hasDedicatedUsbLoggingPort() { return dedicated; }
+void noteUsbLoggingStatsCommand(const char* command) {
+  stats_commands.emplace_back(command);
+}
+Stream& usbCompanionPort() { return input; }
+void discardUsbTerminalOutput() {}
+}
+struct Mesh {
+  bool terminal=false, waiting=false, attach_succeeds=true;
+  bool attach_with_source=false;
+  unsigned banners=0, attaches=0, detaches=0;
+  std::vector<std::string> commands;
+  bool isTerminalMode() const { return terminal; }
+  bool isTerminalWaitingForInput() const { return waiting; }
+  void enterTerminalMode(bool banner) {
+    terminal=true; waiting=!banner;
+    if(banner) { ++banners; input.output+="BANNER"; }
+  }
+  void exitTerminalMode() { terminal=false; waiting=false; }
+  void resetUsbHostSessionInput() {}
+  void handleTerminalCommand(char* command) { commands.emplace_back(command); }
+  bool handleLocalControlCommand(const char* command,char* reply,size_t size) {
+    if(!strcmp(command,"ota folder on")) {
+      ++attaches;
+      if(attach_with_source) {
+        auto& source=mesh::ota::OtaContext::serialFolderSource();
+        mesh::ota::MotaDesc desc;
+        attach_succeeds=source.count()!=0 && source.describe(0,desc);
+      }
+      snprintf(reply,size,"%s",attach_succeeds?"OK attached":"ERR attach failed");
+    } else {
+      assert(!strcmp(command,"ota folder off")); ++detaches;
+      snprintf(reply,size,"OK detached");
+    }
+    return true;
+  }
+} the_mesh;
+static char usb_terminal_line[400]={};
+static size_t usb_terminal_line_len=0;
+static bool usb_terminal_discard_line=false;
+static bool usb_logging_terminal_mode=false,usb_logging_network_parked=false;
+static mesh::UsbBinaryStartupProbe usb_binary_startup_probe;
+static mesh::UsbAsciiSessionDefault usb_ascii_session_default;
+static bool usb_mota_mode=false,usb_mota_disconnect_armed=false;
+static mesh::UsbMotaEntryOrigin usb_mota_entry_origin=mesh::UsbMotaEntryOrigin::BINARY;
+static char usb_mota_line[32]={};
+static size_t usb_mota_line_len=0;
+static bool usb_mota_discard_line=false;
+static const char USB_TERMINAL_START_TOKEN[]="+++MESHCORE-TERM-START";
+static const char USB_TERMINAL_STOP_TOKEN[]="+++MESHCORE-TERM-STOP";
+static const char USB_MOTA_START_TOKEN[]="ota folder on";
+static bool isUsbTerminalDataConnected() { return connected; }
+static void redrawUsbTerminalInput() {}
+static void serviceUsbLoggingOwnership(bool) {}
+static void cancelUsbSerialOperations() {}
+static void queueUsbTerminalControlReply(const char* reply) { input.output+=reply; }
+static void drainUsbTerminalOutputBeforeProtocolSwitch() {}
+@FUNCTIONS@
+static void reset() {
+  input=Input(); the_mesh=Mesh(); connected=true; logging=dedicated=false;
+  stats_commands.clear();
+  mesh::ota::OtaContext::serialFolderSource().resetSessionState();
+  usb_serial_interface=ArduinoSerialInterface();
+  usb_serial_interface.begin(input,USB_TERMINAL_START_TOKEN,USB_MOTA_START_TOKEN);
+  usb_serial_interface.enable();
+  usb_serial_interface.setReceiveFrameCheck(acceptUsbBinaryStartupFrame);
+  usb_logging_terminal_mode=usb_logging_network_parked=false;
+  usb_mota_mode=usb_mota_disconnect_armed=usb_mota_discard_line=false;
+  usb_mota_line_len=0; usb_mota_line[0]=0;
+  clearUsbTerminalLine(); usb_terminal_discard_line=false;
+  g_mock_millis=100;
+  enterUsbTerminalMode(false);
+}
+static void finishProbe() {
+  uint8_t frame[MAX_FRAME_SIZE]={};
+  assert(usb_serial_interface.checkRecvFrame(frame)==2);
+  assert(frame[0]==0x16 && frame[1]==3);
+  serviceUsbTerminal();
+  assert(!the_mesh.terminal && !usb_serial_interface.isPassthroughMode());
+  assert(!usb_binary_startup_probe.isActive());
+}
+static std::vector<uint8_t> response(uint8_t op,const std::vector<uint8_t>& payload,
+                                     uint8_t status=0) {
+  std::vector<uint8_t> result={'m','s',op,status};
+  result.insert(result.end(),payload.begin(),payload.end());
+  uint8_t checksum=0;
+  for(uint8_t value:result) checksum^=value;
+  result.push_back(checksum);
+  return result;
+}
+int main() {
+  for(const char* leading : {"","\r","\n","\r\n","\n\r\n\r"}) {
+    reset(); input.push(leading); input.frame();
+    serviceUsbTerminal();
+    assert(!the_mesh.terminal && usb_binary_startup_probe.isActive());
+    assert(input.available()==5 && input.peek()=='<');
+    assert(input.output.empty() && the_mesh.banners==0);
+    finishProbe();
+  }
+  // Empty Enter reveals the silent terminal without fabricating a command.
+  for(const char* leading : {"\r","\n","\r\n"}) {
+    reset(); input.push(leading); serviceUsbTerminal();
+    assert(the_mesh.terminal && !the_mesh.waiting && the_mesh.banners==1);
+    assert(!usb_binary_startup_probe.isActive() && input.available()==0);
+    assert(the_mesh.commands.empty());
+    serviceUsbTerminal(); assert(the_mesh.banners==1);
+  }
+  reset(); input.push("\r\nget name\r\n"); serviceUsbTerminal();
+  assert(the_mesh.banners==1 && the_mesh.commands.size()==1);
+  assert(the_mesh.commands[0]=="get name" && the_mesh.terminal);
+  // Reaching an empty prompt through an edit also leaves the frame untouched.
+  reset(); input.push("x\b"); input.frame(); serviceUsbTerminal();
+  assert(usb_binary_startup_probe.isActive() && input.available()==5);
+  finishProbe();
+  // A typed prefix and an overlong-line discard cannot become binary probes.
+  reset(); input.push("x"); input.frame(); serviceUsbTerminal();
+  assert(the_mesh.terminal && !usb_binary_startup_probe.isActive());
+  reset(); the_mesh.waiting=false; usb_terminal_discard_line=true;
+  input.frame(); serviceUsbTerminal();
+  assert(the_mesh.terminal && !usb_binary_startup_probe.isActive());
+  assert(usb_terminal_discard_line && usb_terminal_line_len==0);
+  // Single-CDC logging forbids frames; a dedicated logger does not block CDC0.
+  reset(); logging=true; usb_logging_terminal_mode=true;
+  input.push("\r\n"); input.frame(); serviceUsbTerminal();
+  assert(the_mesh.terminal && !usb_binary_startup_probe.isActive());
+  reset(); logging=dedicated=true; input.push("\r\n"); input.frame();
+  serviceUsbTerminal(); assert(usb_binary_startup_probe.isActive()); finishProbe();
+  reset(); usb_logging_network_parked=true; input.frame(); serviceUsbTerminal();
+  assert(input.available()==5 && !usb_binary_startup_probe.isActive());
+  // NUL is legal inside a Binary frame, not a plaintext-line terminator.
+  reset(); input.push("\r\n"); input.frame(0); serviceUsbTerminal();
+  uint8_t nul_frame[MAX_FRAME_SIZE]={};
+  assert(usb_serial_interface.checkRecvFrame(nul_frame)==2);
+  assert(nul_frame[0]==0x16 && nul_frame[1]==0);
+  serviceUsbTerminal(); assert(!the_mesh.terminal && !usb_binary_startup_probe.isActive());
+
+  // The gate uses the actual completion timestamp, before returning a command
+  // or granting client proof, including delayed mesh work and millis rollover.
+  for(uint32_t start : {100u,UINT32_MAX-20}) {
+    for(uint32_t elapsed : {999u,1000u,1001u}) {
+      reset(); g_mock_millis=start; input.frame(); serviceUsbTerminal();
+      assert(usb_binary_startup_probe.isActive());
+      g_mock_millis=start+elapsed;
+      uint8_t command[MAX_FRAME_SIZE]={};
+      const bool accepted=elapsed<1000;
+      assert(usb_serial_interface.checkRecvFrame(command)==(accepted?2u:0u));
+      assert(usb_serial_interface.hasReceivedFrame()==accepted);
+      assert(usb_serial_interface.getCompletedFrameCount()==(accepted?1u:0u));
+      assert(usb_serial_interface.getLastFrameMillis()==(accepted?millis():0));
+      assert(!usb_serial_interface.isReadBusy());
+      serviceUsbTerminal();
+      assert(the_mesh.terminal==!accepted && !usb_binary_startup_probe.isActive());
+    }
+  }
+  reset(); input.frame(); serviceUsbTerminal();
+  g_mock_millis=1099; expireUsbBinaryStartupProbeBeforeDispatch();
+  assert(usb_binary_startup_probe.isActive());
+  // Time spent inside mesh.loop after its earlier deadline precheck.
+  g_mock_millis=1100;
+  uint8_t delayed_command[MAX_FRAME_SIZE]={};
+  assert(usb_serial_interface.checkRecvFrame(delayed_command)==0);
+  assert(!usb_serial_interface.hasReceivedFrame());
+  serviceUsbTerminal(); assert(the_mesh.terminal);
+  reset(); input.frame(); serviceUsbTerminal();
+  g_mock_millis=1098; input.advance_per_read=1;
+  assert(usb_serial_interface.checkRecvFrame(delayed_command)==0);
+  assert(millis()==1103 && !usb_serial_interface.hasReceivedFrame());
+  input.advance_per_read=0; serviceUsbTerminal(); assert(the_mesh.terminal);
+  reset(); input.frame(); serviceUsbTerminal(); g_mock_millis=1099;
+  assert(usb_serial_interface.checkRecvFrame(delayed_command)==2);
+  assert(!usb_binary_startup_probe.isActive());
+  // The first timely frame establishes ownership before the service poll.
+  g_mock_millis=1100; input.frame();
+  assert(usb_serial_interface.checkRecvFrame(delayed_command)==2);
+  assert(usb_serial_interface.getCompletedFrameCount()==2);
+  serviceUsbTerminal(); assert(!the_mesh.terminal);
+  // Once explicitly selected, newline-free STOP retains its legacy behavior.
+  reset(); input.push(USB_TERMINAL_STOP_TOKEN); serviceUsbTerminal();
+  assert(!the_mesh.terminal && !usb_binary_startup_probe.isActive());
+  g_mock_millis+=1001; input.frame();
+  assert(usb_serial_interface.checkRecvFrame(delayed_command)==2);
+
+  // Text-first attach, clean exit, and the remaining CRLF cannot corrupt a
+  // immediately following framed client command.
+  reset(); input.push("ota folder on\r\n"); serviceUsbTerminal();
+  assert(usb_mota_mode && usb_mota_entry_origin==mesh::UsbMotaEntryOrigin::ASCII);
+  assert(!the_mesh.terminal && the_mesh.attaches==1);
+  input.push("ota folder off\r\n"); input.frame(); serviceUsbTerminal();
+  assert(!usb_mota_mode && the_mesh.terminal && the_mesh.waiting);
+  assert(the_mesh.detaches==1);
+  input.output.clear(); serviceUsbTerminal();
+  assert(usb_binary_startup_probe.isActive() && input.available()==5);
+  assert(input.output.empty()); finishProbe();
+
+  // Failed ASCII entry stays ASCII; failed Binary entry stays Binary.
+  reset(); the_mesh.attach_succeeds=false;
+  input.push("ota folder on\r\n"); serviceUsbTerminal();
+  assert(!usb_mota_mode && the_mesh.terminal && the_mesh.banners==2);
+  assert(usb_serial_interface.isPassthroughMode() && !usb_mota_discard_line);
+  reset(); leaveUsbTerminalMode(false); the_mesh.attach_succeeds=false;
+  input.push("ota folder on\r\n"); uint8_t frame[MAX_FRAME_SIZE]={};
+  assert(usb_serial_interface.checkRecvFrame(frame)==0);
+  serviceUsbTerminal();
+  assert(!usb_mota_mode && !the_mesh.terminal && !usb_serial_interface.isPassthroughMode());
+  assert(!usb_mota_discard_line);
+
+  // Every NUL position rejects the whole ASCII owner-transition/mutating
+  // line, then the next real command is handled normally.
+  for(const char* command : {"ota folder on","ota folder off","set usb.logging on","get stats"}) {
+    for(size_t position=0;position<=strlen(command);++position) {
+      reset();
+      for(size_t i=0;i<position;++i) input.bytes.push_back(command[i]);
+      input.bytes.push_back(0);
+      input.push(command+position); input.push("\r\nget name\r\n");
+      serviceUsbTerminal();
+      assert(!usb_mota_mode && the_mesh.attaches==0 && the_mesh.detaches==0);
+      assert(the_mesh.commands.size()==1 && the_mesh.commands[0]=="get name");
+      assert(stats_commands==the_mesh.commands);
+      const auto error=input.output.find("ERROR: invalid command");
+      assert(error!=std::string::npos);
+      assert(input.output.find("ERROR: invalid command",error+1)==std::string::npos);
+    }
+  }
+  reset(); input.push("ota fol"); input.bytes.push_back(0); serviceUsbTerminal();
+  assert(usb_terminal_discard_line && usb_terminal_line_len==0 && the_mesh.attaches==0);
+  input.push("der on\r\nget name\r\n"); serviceUsbTerminal();
+  assert(the_mesh.commands.size()==1 && the_mesh.commands[0]=="get name");
+  assert(!usb_terminal_discard_line && !usb_mota_mode && the_mesh.attaches==0);
+
+  // Every accepted byte, UTF-8 erase, and invalid-line discard preserves the
+  // terminated-length invariant; only complete valid lines prove a reader.
+  reset();
+  const uint8_t edited[]={'g','e','t',' ','s','t','a','t','s',0xc3,0xa9,0x7f,'\r','\n'};
+  for(uint8_t byte : edited) {
+    input.bytes.push_back(byte); serviceUsbTerminal();
+    assert(strlen(usb_terminal_line)==usb_terminal_line_len);
+  }
+  assert(the_mesh.commands==std::vector<std::string>{"get stats"});
+  assert(stats_commands==the_mesh.commands);
+  reset(); dedicated=true; input.push("get stats\r\n"); serviceUsbTerminal();
+  assert(the_mesh.commands==std::vector<std::string>{"get stats"});
+  assert(stats_commands.empty());
+
+  // Overlong lines never accept their suffix as an independent shutdown.
+  for(unsigned length : {32u,40u,4096u}) {
+    reset(); leaveUsbTerminalMode(false);
+    assert(enterUsbMotaMode(mesh::UsbMotaEntryOrigin::ASCII));
+    for(unsigned i=0;i<length;++i) input.bytes.push_back('x');
+    input.push("ota folder off\r\n"); serviceUsbMota();
+    assert(usb_mota_mode && the_mesh.detaches==0);
+    assert(!usb_mota_discard_line && usb_mota_line_len==0);
+    input.push("ota folder off\r\n"); serviceUsbMota();
+    assert(!usb_mota_mode && the_mesh.detaches==1 && the_mesh.terminal);
+  }
+  // Discard survives arbitrary read boundaries, but reset/entry clear it.
+  reset(); leaveUsbTerminalMode(false);
+  assert(enterUsbMotaMode(mesh::UsbMotaEntryOrigin::BINARY));
+  input.push("xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"); serviceUsbMota();
+  assert(usb_mota_line_len==31 && !usb_mota_discard_line);
+  input.push("xota folder off"); serviceUsbMota();
+  assert(usb_mota_discard_line && usb_mota_mode && the_mesh.detaches==0);
+  input.push("\n"); serviceUsbMota(); assert(!usb_mota_discard_line);
+  input.push("ota folder off\r\n"); serviceUsbMota();
+  assert(!usb_mota_mode && !the_mesh.terminal && !usb_serial_interface.isPassthroughMode());
+  usb_mota_discard_line=true; resetUsbMotaMode(); assert(!usb_mota_discard_line);
+  usb_mota_discard_line=true;
+  assert(enterUsbMotaMode(mesh::UsbMotaEntryOrigin::BINARY));
+  assert(!usb_mota_discard_line);
+  // mOTA NUL-containing shutdown strings must not truncate into an exact
+  // owner command. Rejecting them is silent and preserves the source.
+  for(const char* command : {"ota folder on","ota folder off"}) {
+    for(size_t position=0;position<=strlen(command);++position) {
+      reset(); leaveUsbTerminalMode(false);
+      assert(enterUsbMotaMode(mesh::UsbMotaEntryOrigin::ASCII));
+      input.output.clear();
+      for(size_t i=0;i<position;++i) input.bytes.push_back(command[i]);
+      input.bytes.push_back(0);
+      input.push(command+position); input.push("\r\n"); serviceUsbMota();
+      assert(usb_mota_mode && the_mesh.detaches==0 && input.output.empty());
+      input.push("ota folder off\r\n"); serviceUsbMota();
+      assert(!usb_mota_mode && the_mesh.detaches==1 && the_mesh.terminal);
+    }
+  }
+  reset(); leaveUsbTerminalMode(false);
+  assert(enterUsbMotaMode(mesh::UsbMotaEntryOrigin::BINARY));
+  input.push("ota folder off"); input.bytes.push_back(0); serviceUsbMota();
+  assert(usb_mota_discard_line && usb_mota_line_len==0 && the_mesh.detaches==0);
+  input.push("suffix\n"); serviceUsbMota();
+  assert(!usb_mota_discard_line && usb_mota_mode && the_mesh.detaches==0);
+  input.push("ota folder off\r\n"); serviceUsbMota();
+  assert(!usb_mota_mode && the_mesh.detaches==1 && !the_mesh.terminal);
+  // Continually arriving data cannot make a single mOTA control poll unbounded.
+  reset(); leaveUsbTerminalMode(false);
+  assert(enterUsbMotaMode(mesh::UsbMotaEntryOrigin::BINARY));
+  input.bytes={'x','x','x','x'}; input.replenish=true;
+  const unsigned before=input.reads; serviceUsbMota();
+  assert(input.reads==before+4 && input.available()==4);
+  // The ordinary ASCII handler yields too, including while discarding an
+  // overlong command from a host that never stops writing.
+  reset(); input.bytes={'x','x','x','x'}; input.replenish=true;
+  for(unsigned pass=0;pass<128;++pass) {
+    const unsigned before=input.reads; serviceUsbTerminal();
+    assert(input.reads==before+4 && input.available()==4);
+  }
+  assert(usb_terminal_discard_line && the_mesh.commands.empty());
+  input.replenish=false; input.push("\nget name\r\n"); serviceUsbTerminal();
+  assert(!usb_terminal_discard_line && the_mesh.commands.size()==1);
+  assert(the_mesh.commands[0]=="get name");
+
+  // A real timed-out READ reply can contain shutdown text as arbitrary image
+  // bytes. No prefix, header, payload or checksum boundary may expose it.
+  const char dangerous[]="x\nota folder off\n";
+  const auto late_read=response(3,std::vector<uint8_t>(dangerous,dangerous+sizeof(dangerous)-1));
+  for(size_t prefix=0;prefix<late_read.size();++prefix) {
+    reset(); leaveUsbTerminalMode(false);
+    assert(enterUsbMotaMode(mesh::UsbMotaEntryOrigin::ASCII));
+    auto& source=mesh::ota::OtaContext::serialFolderSource();
+    input.responses.emplace_back(late_read.begin(),late_read.begin()+prefix);
+    uint8_t data[sizeof(dangerous)-1]={};
+    assert(!source.read(0,0,data,sizeof(data)) && source.hasPendingResponse());
+    const unsigned requests=input.source_requests;
+    for(unsigned retry=0;retry<3;++retry) {
+      assert(!source.read(0,0,data,1));
+      assert(input.source_requests==requests && source.hasPendingResponse());
+    }
+    // Drain fragments through both txn's stale pass and the control consumer.
+    for(size_t i=prefix;i<late_read.size();++i) {
+      input.bytes.push_back(late_read[i]);
+      if(i+1<late_read.size() && i%2==0) {
+        assert(!source.read(0,0,data,1));
+        assert(input.source_requests==requests);
+      } else serviceUsbMota();
+      assert(usb_mota_mode && the_mesh.detaches==0);
+    }
+    assert(!source.hasPendingResponse());
+    input.push("ota folder off\r\n"); serviceUsbMota();
+    assert(!usb_mota_mode && the_mesh.detaches==1 && the_mesh.terminal);
+  }
+  // Successful response followed immediately by an ordinary shutdown retains
+  // both its payload and the exact control bytes at the proven wire boundary.
+  reset(); leaveUsbTerminalMode(false);
+  assert(enterUsbMotaMode(mesh::UsbMotaEntryOrigin::ASCII));
+  auto good_read=response(3,{42});
+  const char off[]="ota folder off\r\n";
+  good_read.insert(good_read.end(),off,off+sizeof(off)-1);
+  input.responses.push_back(good_read);
+  uint8_t good_byte=0;
+  assert(mesh::ota::OtaContext::serialFolderSource().read(0,0,&good_byte,1));
+  assert(good_byte==42); serviceUsbMota();
+  assert(!usb_mota_mode && the_mesh.detaches==1 && the_mesh.terminal);
+
+  // Failed ASCII and Binary attach must quarantine a late COUNT/DESCRIBE
+  // frame through every partial boundary, not execute its text/frame payload.
+  std::vector<uint8_t> desc(38,'x');
+  const char mutation[]="\nset usb.logging on\n";
+  std::copy(mutation,mutation+sizeof(mutation)-1,desc.begin()+2);
+  const uint8_t embedded_frame[]={'<',2,0,0x16,3};
+  std::copy(embedded_frame,embedded_frame+sizeof(embedded_frame),desc.begin()+25);
+  for(bool ascii : {false,true}) {
+    for(bool describe : {false,true}) {
+      const auto delayed=describe?response(2,desc):response(1,{1});
+      for(size_t prefix=0;prefix<delayed.size();++prefix) {
+        reset(); the_mesh.attach_with_source=true;
+        if(describe) input.responses.push_back(response(1,{1}));
+        input.responses.emplace_back(delayed.begin(),delayed.begin()+prefix);
+        if(!ascii) {
+          leaveUsbTerminalMode(false);
+          assert(!enterUsbMotaMode(mesh::UsbMotaEntryOrigin::BINARY));
+        } else {
+          input.push("ota folder on\r\n"); serviceUsbTerminal();
+        }
+        auto& source=mesh::ota::OtaContext::serialFolderSource();
+        assert(!usb_mota_mode && source.hasPendingResponse());
+        assert(the_mesh.terminal==ascii && usb_serial_interface.isPassthroughMode());
+        const unsigned requests=input.source_requests;
+        assert(!enterUsbMotaMode(mesh::UsbMotaEntryOrigin::BINARY));
+        assert(source.hasPendingResponse() && input.source_requests==requests);
+        uint8_t command[MAX_FRAME_SIZE]={};
+        for(size_t i=prefix;i<delayed.size();++i) {
+          input.bytes.push_back(delayed[i]);
+          assert(usb_serial_interface.checkRecvFrame(command)==0);
+          serviceUsbTerminal();
+          assert(the_mesh.commands.empty() && !usb_serial_interface.hasReceivedFrame());
+        }
+        assert(!source.hasPendingResponse());
+        assert(usb_serial_interface.isPassthroughMode()==ascii);
+        if(ascii) {
+          input.push("get name\r\n"); serviceUsbTerminal();
+          assert(the_mesh.commands.size()==1 && the_mesh.commands[0]=="get name");
+        } else {
+          input.frame(); assert(usb_serial_interface.checkRecvFrame(command)==2);
+        }
+      }
+    }
+  }
+  // A partial payload is deliberately ambiguous: ordinary input cannot end
+  // quarantine. A physical session reset does clear the retained transaction.
+  reset(); the_mesh.attach_with_source=true;
+  input.responses.push_back(response(1,{1}));
+  const auto delayed_desc=response(2,desc);
+  input.responses.emplace_back(delayed_desc.begin(),delayed_desc.begin()+5);
+  input.push("ota folder on\r\n"); serviceUsbTerminal();
+  auto& waiting_source=mesh::ota::OtaContext::serialFolderSource();
+  assert(waiting_source.hasPendingResponse());
+  input.push("get name\r\n"); serviceUsbTerminal();
+  assert(the_mesh.commands.empty() && waiting_source.hasPendingResponse());
+  input.bytes.clear(); // The transport purges the old session's bytes.
+  resetUsbTerminalHostSession();
+  assert(!waiting_source.hasPendingResponse() && !usb_serial_interface.hasReceivedFrame());
+  // A real software handoff preserves the outstanding frame and restores the
+  // chosen terminal/logging/network ownership only after its final checksum.
+  for(bool terminal : {false,true}) {
+    for(bool shared_logging : {false,true}) {
+      for(bool parked : {false,true}) {
+        reset(); leaveUsbTerminalMode(false);
+        assert(enterUsbMotaMode(mesh::UsbMotaEntryOrigin::BINARY));
+        auto& source=mesh::ota::OtaContext::serialFolderSource();
+        input.responses.emplace_back(late_read.begin(),late_read.begin()+5);
+        uint8_t data[sizeof(dangerous)-1]={};
+        assert(!source.read(0,0,data,sizeof(data)));
+        leaveUsbMotaMode(false);
+        assert(source.hasPendingResponse() && usb_serial_interface.isPassthroughMode());
+        if(terminal) enterUsbTerminalMode(false);
+        logging=true; dedicated=!shared_logging;
+        usb_logging_network_parked=parked;
+        for(size_t i=5;i<late_read.size();++i) input.bytes.push_back(late_read[i]);
+        assert(serviceUsbMotaResponseBoundary());
+        assert(!source.hasPendingResponse() && the_mesh.commands.empty());
+        assert(the_mesh.terminal==terminal);
+        assert(usb_serial_interface.isPassthroughMode()==(terminal || shared_logging || parked));
+      }
+    }
+  }
+}
+'''
+
+
+class CompanionUsbInputTests(unittest.TestCase):
+    def test_actual_usb_input_state_machines(self):
+        functions = "\n".join(extract_braced(MAIN, signature)
+                              for signature in (
+            "static void clearUsbTerminalLine()",
+            "static bool acceptUsbBinaryStartupFrame(uint32_t completed_at)",
+            "static void enterUsbTerminalMode(bool show_banner",
+            "static void leaveUsbTerminalMode(bool acknowledge)",
+            "static void resetUsbMotaMode()",
+            "static void leaveUsbMotaMode(bool acknowledge)",
+            "static bool enterUsbMotaMode(mesh::UsbMotaEntryOrigin origin)",
+            "static void serviceUsbMota()",
+            "static bool serviceUsbMotaResponseBoundary()",
+            "static void resetUsbTerminalHostSession()",
+            "static void serviceUsbTerminal()",
+            "static void expireUsbBinaryStartupProbeBeforeDispatch()",
+        ))
+        # The default is in the forward declaration in production.
+        functions = functions.replace("static void enterUsbTerminalMode(bool show_banner)",
+                                      "static void enterUsbTerminalMode(bool show_banner = true)")
+        accessor = extract_braced(CONTEXT, "static SerialMotaSource& serialFolderSource()")
+        with tempfile.TemporaryDirectory() as directory:
+            cpp = Path(directory) / "usb-input.cpp"
+            cpp.write_text(HARNESS.replace("@FUNCTIONS@", functions)
+                           .replace("@SOURCE_ACCESSOR@", accessor))
+            for sanitizer in (False, True):
+                with self.subTest(sanitizer=sanitizer):
+                    binary = Path(directory) / ("usb-input-" + str(sanitizer))
+                    flags = (["-fsanitize=address,undefined", "-fno-sanitize-recover=all",
+                              "-fno-pie", "-no-pie"] if sanitizer else [])
+                    build = subprocess.run([
+                        "g++", "-std=c++17", *flags,
+                        "-I", str(ROOT / "test/mocks"), "-I", str(ROOT / "src"),
+                        str(cpp), str(ROOT / "src/helpers/ArduinoSerialInterface.cpp"),
+                        str(ROOT / "src/helpers/ota/MotaSourceSerial.cpp"),
+                        "-o", str(binary)], capture_output=True, text=True)
+                    self.assertEqual(build.returncode, 0, build.stdout + build.stderr)
+                    run = subprocess.run([str(binary)], capture_output=True, text=True)
+                    self.assertEqual(run.returncode, 0, run.stdout + run.stderr)
+
+
+if __name__ == "__main__":
+    unittest.main()

test/test_companion_usb_logging_handoff.py

@@ -0,0 +1,223 @@
+#!/usr/bin/env python3
+"""Execute real USB logging/mOTA handoffs across retained response boundaries."""
+
+from pathlib import Path
+import subprocess
+import tempfile
+import unittest
+
+from test_companion_usb_input import CONTEXT, HARNESS, MAIN, ROOT
+from test_replay_reset_integration import extract_braced
+
+
+CASES = r'''
+static void selectAscii() {
+  if (!the_mesh.terminal) {
+    input.push(USB_TERMINAL_START_TOKEN); input.push("\n");
+    uint8_t frame[MAX_FRAME_SIZE]={};
+    assert(usb_serial_interface.checkRecvFrame(frame)==0);
+    serviceUsbTerminal();
+    assert(the_mesh.terminal);
+  }
+}
+
+static void verifyNextClients() {
+  selectAscii();
+  input.push("get name\r\n"); serviceUsbTerminal();
+  assert(the_mesh.commands.size()==1 && the_mesh.commands[0]=="get name");
+  logging=false; serviceUsbLoggingOwnership(false);
+  input.push(USB_TERMINAL_STOP_TOKEN); serviceUsbTerminal();
+  assert(!the_mesh.terminal && !usb_serial_interface.isPassthroughMode());
+  input.frame();
+  uint8_t frame[MAX_FRAME_SIZE]={};
+  assert(usb_serial_interface.checkRecvFrame(frame)==2);
+  assert(frame[0]==0x16 && frame[1]==3);
+  assert(usb_serial_interface.getCompletedFrameCount()==1);
+}
+
+static void responseHandoffs() {
+  // Every partial stage is exercised, with image bytes that resemble both
+  // ASCII ownership commands and Companion framing (including embedded NUL).
+  const char controls[]="\nget name\r\nota folder off\n+++MESHCORE-TERM-STOP\n";
+  std::vector<uint8_t> payload(controls,controls+sizeof(controls)-1);
+  payload.insert(payload.end(),{'<',2,0,0x16,3});
+  for (uint8_t status : {uint8_t(0),uint8_t(1)}) {
+    const auto late=response(3,status==0?payload:std::vector<uint8_t>{},status);
+    for (size_t prefix=0;prefix<late.size();++prefix) {
+      for (bool logging_after_close : {false,true}) {
+        for (bool logging_off_before_close : {false,true}) {
+          for (auto origin : {mesh::UsbMotaEntryOrigin::ASCII,
+                              mesh::UsbMotaEntryOrigin::BINARY}) {
+            reset(); network=false; leaveUsbTerminalMode(false);
+            assert(enterUsbMotaMode(origin));
+            auto& source=mesh::ota::OtaContext::serialFolderSource();
+            input.responses.emplace_back(late.begin(),late.begin()+prefix);
+            std::vector<uint8_t> data(payload.size());
+            assert(!source.read(0,0,data.data(),data.size()));
+            assert(source.hasPendingResponse());
+            logging=network=true; serviceUsbLoggingOwnership(true);
+            assert(!usb_mota_mode && usb_logging_network_parked);
+            assert(usb_serial_interface.isPassthroughMode());
+            assert(the_mesh.attaches==1 && the_mesh.detaches==1);
+            if (logging_off_before_close) {
+              logging=false; serviceUsbLoggingOwnership(false);
+              assert(!usb_logging_network_parked && source.hasPendingResponse());
+              assert(usb_serial_interface.isPassthroughMode());
+            }
+            network=false; logging=logging_after_close;
+            uint8_t command[MAX_FRAME_SIZE]={};
+            for (size_t i=prefix;i<late.size();++i) {
+              input.bytes.push_back(late[i]);
+              serviceUsbLoggingOwnership(logging);
+              // Unparking cannot raw-drain the retained response, even if
+              // diagnostics are being disabled at the same time.
+              assert(input.available()==1);
+              assert(usb_serial_interface.isPassthroughMode());
+              assert(usb_serial_interface.checkRecvFrame(command)==0);
+              serviceUsbTerminal();
+              assert(source.hasPendingResponse()==(i+1<late.size()));
+              assert(the_mesh.commands.empty() && the_mesh.detaches==1);
+              assert(!usb_serial_interface.hasReceivedFrame());
+            }
+            assert(!usb_logging_network_parked && !source.hasPendingResponse());
+            verifyNextClients();
+          }
+        }
+      }
+    }
+  }
+
+  // The boundary consumer stops at the checksum, leaving a subsequent real
+  // ASCII command intact rather than discarding the entire queued snapshot.
+  reset(); network=false; leaveUsbTerminalMode(false);
+  assert(enterUsbMotaMode(mesh::UsbMotaEntryOrigin::BINARY));
+  auto& source=mesh::ota::OtaContext::serialFolderSource();
+  const auto late=response(3,{42,43});
+  input.responses.emplace_back(late.begin(),late.begin()+5);
+  uint8_t data[2]={}; assert(!source.read(0,0,data,2));
+  logging=network=true; serviceUsbLoggingOwnership(true);
+  for (size_t i=5;i<late.size();++i) input.bytes.push_back(late[i]);
+  input.push("get name\n"); network=false;
+  serviceUsbTerminal();
+  assert(!source.hasPendingResponse() && input.available()==9);
+  assert(the_mesh.commands.empty());
+  serviceUsbTerminal();
+  assert(the_mesh.commands.size()==1 && the_mesh.commands[0]=="get name");
+
+  // Without a retained binary response, the original bounded discard still
+  // removes log-only input. A refilling source cannot trap this handoff.
+  reset(); network=true; logging=true; leaveUsbTerminalMode(false);
+  serviceUsbLoggingOwnership(true); assert(usb_logging_network_parked);
+  input.push("xxxx"); input.replenish=true;
+  const unsigned reads=input.reads; network=false;
+  serviceUsbLoggingOwnership(true);
+  assert(input.reads==reads+4 && input.available()==4);
+  assert(!usb_logging_network_parked && the_mesh.terminal);
+  input.replenish=false;
+}
+
+static void probeHandoffs() {
+  // Empty and oversize startup frames grant no Binary proof; a deliberate
+  // mOTA token after either must supersede the abandoned probe's deadline.
+  for (uint16_t length : {uint16_t(0),uint16_t(177),uint16_t(65535)}) {
+    for (bool attach_succeeds : {false,true}) {
+      reset(); network=false; the_mesh.attach_succeeds=attach_succeeds;
+      input.bytes={'<',uint8_t(length),uint8_t(length>>8)};
+      for (uint32_t i=0;i<length;++i) input.bytes.push_back('x');
+      input.push("ota folder on\n"); serviceUsbTerminal();
+      assert(usb_binary_startup_probe.isActive() && !the_mesh.terminal);
+      uint8_t frame[MAX_FRAME_SIZE]={};
+      assert(usb_serial_interface.checkRecvFrame(frame)==0);
+      assert(!usb_serial_interface.hasReceivedFrame());
+      serviceUsbTerminal();
+      assert(usb_mota_mode==attach_succeeds && !the_mesh.terminal);
+      assert(!usb_binary_startup_probe.isActive());
+      assert(usb_serial_interface.isPassthroughMode()==attach_succeeds);
+      const unsigned banners=the_mesh.banners;
+      input.bytes.push_back('x'); g_mock_millis+=1001;
+      expireUsbBinaryStartupProbeBeforeDispatch();
+      assert(usb_mota_mode==attach_succeeds && !the_mesh.terminal);
+      assert(the_mesh.banners==banners && input.available()==1);
+    }
+  }
+
+  // A deferred ASCII default from a prior network handoff must also lose to
+  // deliberate mOTA ownership, including Binary-origin attach failure.
+  for (bool attach_succeeds : {false,true}) {
+    for (auto origin : {mesh::UsbMotaEntryOrigin::ASCII,
+                        mesh::UsbMotaEntryOrigin::BINARY}) {
+      reset(); leaveUsbTerminalMode(false); network=true;
+      usb_ascii_session_default.request(usb_serial_interface.getCompletedFrameCount());
+      usb_binary_startup_probe.start(millis(),usb_serial_interface.getCompletedFrameCount());
+      the_mesh.attach_succeeds=attach_succeeds;
+      assert(enterUsbMotaMode(origin)==attach_succeeds);
+      assert(!usb_binary_startup_probe.isActive());
+      network=false; serviceUsbAsciiSessionDefault();
+      assert(!usb_ascii_session_default.shouldRestore(
+          true,false,usb_serial_interface.getCompletedFrameCount()));
+      assert(usb_mota_mode==attach_succeeds);
+      assert(the_mesh.terminal==(!attach_succeeds && origin==mesh::UsbMotaEntryOrigin::ASCII));
+    }
+  }
+}
+
+int main() { responseHandoffs(); probeHandoffs(); }
+'''
+
+
+class CompanionUsbLoggingHandoffTests(unittest.TestCase):
+    def test_actual_logging_and_mota_ownership(self):
+        functions = "\n".join(extract_braced(MAIN, signature) for signature in (
+            "static void clearUsbTerminalLine()",
+            "static bool acceptUsbBinaryStartupFrame(uint32_t completed_at)",
+            "static void enterUsbTerminalMode(bool show_banner",
+            "static void enterUsbLoggingTerminalMode()",
+            "static void leaveUsbTerminalMode(bool acknowledge)",
+            "static void resetUsbMotaMode()",
+            "static void leaveUsbMotaMode(bool acknowledge)",
+            "static bool enterUsbMotaMode(mesh::UsbMotaEntryOrigin origin)",
+            "static void serviceUsbMota()",
+            "static bool serviceUsbMotaResponseBoundary()",
+            "static void resetUsbTerminalHostSession()",
+            "static void serviceUsbAsciiSessionDefault()",
+            "static void serviceUsbLoggingOwnership(bool logging_enabled) {",
+            "static void serviceUsbTerminal()",
+            "static void expireUsbBinaryStartupProbeBeforeDispatch()",
+        ))
+        functions = functions.replace(
+            "static void enterUsbTerminalMode(bool show_banner)",
+            "static void enterUsbTerminalMode(bool show_banner = true)")
+        harness = HARNESS[:HARNESS.index("int main()")]
+        harness = harness.replace("#define COMPANION_FEATURE_NETWORK_TERMINAL 0",
+                                  "#define COMPANION_FEATURE_NETWORK_TERMINAL 1")
+        harness = harness.replace(
+            "static void serviceUsbLoggingOwnership(bool) {}",
+            "static bool network=false;\n"
+            "static bool usb_terminal_host_reset_completion_pending=false;\n"
+            "static bool isNetworkTerminalActive() { return network; }\n"
+            "static void serviceUsbLoggingOwnership(bool);")
+        source = harness.replace("@FUNCTIONS@", functions).replace(
+            "@SOURCE_ACCESSOR@",
+            extract_braced(CONTEXT, "static SerialMotaSource& serialFolderSource()"))
+        with tempfile.TemporaryDirectory() as directory:
+            cpp = Path(directory) / "usb-handoff.cpp"
+            cpp.write_text(source + CASES, encoding="ascii")
+            for sanitizer in (False, True):
+                with self.subTest(sanitizer=sanitizer):
+                    binary = Path(directory) / ("usb-handoff-" + str(sanitizer))
+                    flags = (["-fsanitize=address,undefined", "-fno-sanitize-recover=all",
+                              "-fno-pie", "-no-pie"] if sanitizer else [])
+                    build = subprocess.run([
+                        "g++", "-std=c++17", *flags,
+                        "-I", str(ROOT / "test/mocks"), "-I", str(ROOT / "src"),
+                        str(cpp), str(ROOT / "src/helpers/ArduinoSerialInterface.cpp"),
+                        str(ROOT / "src/helpers/ota/MotaSourceSerial.cpp"),
+                        "-o", str(binary)], capture_output=True, text=True, timeout=60)
+                    self.assertEqual(build.returncode, 0, build.stdout + build.stderr)
+                    run = subprocess.run([str(binary)], capture_output=True,
+                                         text=True, timeout=20)
+                    self.assertEqual(run.returncode, 0, run.stdout + run.stderr)
+
+
+if __name__ == "__main__":
+    unittest.main()

test/test_companion_usb_mota_exit.py

@@ -27,11 +27,21 @@ HARNESS = r'''
 namespace mesh {
 enum class UsbMotaEntryOrigin { BINARY, ASCII };
 static void discardUsbTerminalOutput() {}
+namespace ota {
+struct Source {
+  bool hasPendingResponse() const { return false; }
+  void resetSessionState() {}
+};
+struct OtaContext {
+  static Source& serialFolderSource() { static Source source; return source; }
+};
+}
 }
 static bool usb_mota_mode = true, usb_mota_disconnect_armed = true;
 static mesh::UsbMotaEntryOrigin usb_mota_entry_origin = mesh::UsbMotaEntryOrigin::BINARY;
 static char usb_mota_line[32] = "old";
 static size_t usb_mota_line_len = 3;
+static bool usb_mota_discard_line = false;
 static bool ascii = false;
 static int drained = 0, folder_off = 0;
 static char control_reply[224];
@@ -61,6 +71,7 @@ static void prepare(mesh::UsbMotaEntryOrigin origin) {
   usb_mota_mode = usb_mota_disconnect_armed = true;
   usb_mota_entry_origin = origin;
   usb_mota_line_len = 3;
+  usb_mota_discard_line = true;
   strcpy(usb_mota_line, "old");
   usb_serial_interface.passthrough = true;
   ascii = false;
@@ -72,6 +83,7 @@ int main() {
   leaveUsbMotaMode(true);
   assert(ascii && usb_serial_interface.passthrough);
   assert(!usb_mota_mode && !usb_mota_disconnect_armed && usb_mota_line_len == 0);
+  assert(!usb_mota_discard_line);
   assert(folder_off == 1 && drained == 1);
   assert(strstr(control_reply, "OK - Terminal mode") != nullptr);
 

test/test_mota_serial_deadline.py

@@ -0,0 +1,65 @@
+#!/usr/bin/env python3
+"""Run real native mOTA serial tests with actual production code and sanitizers."""
+
+from pathlib import Path
+import subprocess
+import tempfile
+import unittest
+
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+class MotaSerialDeadlineTests(unittest.TestCase):
+    def test_actual_native_serial_cases(self):
+        native = (ROOT / "test/test_ota/test_ota_core.cpp").read_text()
+        start = native.index("class FakeMotaSeederStream")
+        end = native.index("class FakeFolderMotaSeederStream", start)
+        source = r'''
+#include <gtest/gtest.h>
+#include <algorithm>
+#include <array>
+#include <vector>
+#include <cstring>
+#include <type_traits>
+#include <utility>
+#include "helpers/ota/MotaSourceSerial.h"
+#include "helpers/BleMotaStream.h"
+#include "helpers/ota/MotaSeederProto.h"
+#include "helpers/ota/OtaByteIO.h"
+#include "helpers/ota/FolderMotaStore.h"
+using namespace mesh::ota;
+'''
+        source += native[start:end]
+        source += r'''
+int main(int argc, char** argv) {
+  ::testing::InitGoogleTest(&argc, argv);
+  return RUN_ALL_TESTS();
+}
+'''
+        googletest = ROOT / ".pio/libdeps/native/googletest/googletest"
+        self.assertTrue(googletest.is_dir(), "Native googletest dependency must be present")
+        with tempfile.TemporaryDirectory() as directory:
+            cpp = Path(directory) / "serial-deadline.cpp"
+            cpp.write_text(source)
+            for sanitizer in (False, True):
+                with self.subTest(sanitizer=sanitizer):
+                    binary = Path(directory) / ("serial-deadline-" + str(sanitizer))
+                    flags = (["-fsanitize=address,undefined", "-fno-sanitize-recover=all",
+                              "-fno-pie", "-no-pie"] if sanitizer else [])
+                    build = subprocess.run([
+                        "g++", "-std=c++17", "-pthread", *flags,
+                        "-I", str(ROOT / "test/mocks"), "-I", str(ROOT / "src"),
+                        "-I", str(googletest / "include"), "-I", str(googletest),
+                        str(cpp), str(ROOT / "src/helpers/ota/MotaSourceSerial.cpp"),
+                        str(ROOT / "src/helpers/ota/FolderMotaStore.cpp"),
+                        str(googletest / "src/gtest-all.cc"), "-o", str(binary)],
+                        capture_output=True, text=True)
+                    self.assertEqual(build.returncode, 0, build.stdout + build.stderr)
+                    run = subprocess.run([str(binary), "--gtest_brief=1"],
+                                         capture_output=True, text=True, timeout=30)
+                    self.assertEqual(run.returncode, 0, run.stdout + run.stderr)
+
+
+if __name__ == "__main__":
+    unittest.main()

test/test_ota/test_ota_core.cpp

@@ -4,6 +4,7 @@
 #include <vector>
 #include <cstring>
 #include <type_traits>
+#include <utility>
 
 #include "helpers/ota/MotaContainer.h"
 #include "helpers/ota/MerkleTree.h"
@@ -12,6 +13,7 @@
 #include "helpers/ota/FirmwareInfo.h"
 #include "helpers/ota/MotaSeederProto.h"
 #include "helpers/ota/MotaSourceSerial.h"
+#include "helpers/BleMotaStream.h"
 #include "helpers/ota/FolderMotaStore.h"
 #include "helpers/ota/SignerAllowlist.h"
 #include "helpers/ota/OtaStore.h"
@@ -886,6 +888,10 @@ public:
     return -1;
   }
 
+  int available() override {
+    return static_cast<int>(response.size() - response_pos);
+  }
+
   void flush() override {
     ++flush_calls;
     if (flush_discards_receive) {
@@ -900,7 +906,7 @@ public:
   bool flush_discards_receive = false;
   size_t flush_calls = 0;
 
-private:
+protected:
   std::vector<uint8_t> response;
   size_t response_pos = 0;
 };
@@ -955,6 +961,10 @@ public:
     return -1;
   }
 
+  int available() override {
+    return static_cast<int>(response.size() - response_pos);
+  }
+
   void flush() override { ++flush_calls; }
 
   std::vector<uint8_t> encoded;
@@ -1065,6 +1075,403 @@ TEST(MotaSourceSerial, DeflateExtensionFailureLeavesRawFallbackAvailable) {
   EXPECT_EQ(stream.lengths, (std::vector<uint16_t>{0}));
 }
 
+class TimedMotaSeederStream : public FakeMotaSeederStream {
+public:
+  std::vector<uint32_t> byte_at;
+  uint32_t started = 0;
+  uint32_t delay_per_byte = 0;
+  bool wrong_op = false, corrupt_checksum = false, error_status = false;
+  unsigned stale_remaining = 0, stale_refills = 0, stale_reads = 0;
+  unsigned stale_reads_at_request = 0;
+
+  size_t write(const uint8_t* request, size_t len) override {
+    stale_reads_at_request = stale_reads;
+    stale_remaining = stale_refills = 0;
+    started = millis();
+    const size_t written = FakeMotaSeederStream::write(request, len);
+    if (wrong_op) ++response[2];
+    if (error_status) {
+      response.resize(5);
+      response[3] = MS_STATUS_ERR;
+    }
+    uint8_t checksum = 0;
+    for (size_t i = 0; i + 1 < response.size(); ++i) checksum ^= response[i];
+    response.back() = checksum ^ (corrupt_checksum ? 1 : 0);
+    return written;
+  }
+
+  uint32_t byteTime(size_t index) const {
+    return index < byte_at.size() ? byte_at[index]
+                                  : uint32_t(index) * delay_per_byte;
+  }
+
+  int available() override {
+    if (stale_remaining) return static_cast<int>(stale_remaining);
+    const uint32_t elapsed = millis() - started;
+    size_t count = 0;
+    while (response_pos + count < response.size()
+           && byteTime(response_pos + count) <= elapsed) ++count;
+    return static_cast<int>(count);
+  }
+
+  int read() override {
+    if (stale_remaining) {
+      --stale_remaining;
+      ++stale_reads;
+      if (stale_refills) { --stale_refills; ++stale_remaining; }
+      return 'x';
+    }
+    if (response_pos >= response.size()
+        || uint32_t(millis() - started) < byteTime(response_pos)) return -1;
+    return FakeMotaSeederStream::read();
+  }
+
+  size_t unreadResponse() const { return response.size() - response_pos; }
+};
+
+TEST(MotaSourceSerial, SharedDeadlineBoundsTricklingPayload) {
+  resetArduinoMock();
+  TimedMotaSeederStream stream;
+  stream.byte_at.resize(MOTA_SEEDER_READ_MAX + 5);
+  for (size_t i = 4; i < stream.byte_at.size(); ++i) {
+    stream.byte_at[i] = uint32_t(i - 3) * 590;
+  }
+  SerialMotaSource source(stream, MotaStreamWritePolicy::NoFlush, 600);
+  std::array<uint8_t, MOTA_SEEDER_READ_MAX> data{};
+  EXPECT_FALSE(source.read(0, 0, data.data(), data.size()));
+  EXPECT_EQ(millis(), 600u);
+  EXPECT_EQ(data[0], 0u);
+  EXPECT_EQ(data[1], 0u);
+  EXPECT_GT(stream.unreadResponse(), 1u);
+}
+
+TEST(MotaSourceSerial, SyncHeaderPayloadAndChecksumShareOneDeadline) {
+  resetArduinoMock();
+  TimedMotaSeederStream stream;
+  stream.byte_at = {0, 300, 350, 450, 550, 575, 600};
+  SerialMotaSource source(stream, MotaStreamWritePolicy::NoFlush, 600);
+  uint8_t data[2] = {};
+  EXPECT_FALSE(source.read(0, 0, data, sizeof(data)));
+  EXPECT_EQ(millis(), 600u);
+  EXPECT_EQ(stream.unreadResponse(), 1u);
+}
+
+TEST(MotaSourceSerial, DeadlineIsStrictAndHandlesMillisRollover) {
+  for (uint32_t started : {0u, UINT32_MAX - 20}) {
+    for (uint32_t checksum_at : {599u, 600u}) {
+      SCOPED_TRACE(started);
+      SCOPED_TRACE(checksum_at);
+      resetArduinoMock();
+      g_mock_millis = started;
+      TimedMotaSeederStream stream;
+      stream.byte_at = {0, 0, 0, 0, 0, checksum_at};
+      SerialMotaSource source(stream, MotaStreamWritePolicy::NoFlush, 600);
+      uint8_t data = 0;
+      EXPECT_EQ(source.read(0, 0, &data, 1), checksum_at < 600);
+      EXPECT_EQ(uint32_t(millis() - started), checksum_at);
+      EXPECT_EQ(stream.unreadResponse(), checksum_at < 600 ? 0u : 1u);
+    }
+  }
+}
+
+TEST(MotaSourceSerial, ActualReadCompletionMustPrecedeDeadline) {
+  class AdvancingReadStream : public FakeMotaSeederStream {
+  public:
+    int read() override {
+      const int value = FakeMotaSeederStream::read();
+      if (value >= 0) ++g_mock_millis;
+      return value;
+    }
+  };
+  for (uint32_t start : {0u, UINT32_MAX - 3}) {
+    for (uint32_t timeout : {5u, 6u, 7u}) {
+      for (bool shared : {false, true}) {
+        SCOPED_TRACE(start);
+        SCOPED_TRACE(timeout);
+        SCOPED_TRACE(shared);
+        resetArduinoMock();
+        g_mock_millis = start;
+        AdvancingReadStream stream;
+        SerialMotaSource source(stream, MotaStreamWritePolicy::NoFlush, timeout);
+        if (shared) source.enableSharedTextControl();
+        uint8_t data = 0;
+        EXPECT_EQ(source.read(0, 0, &data, 1), timeout > 6);
+        EXPECT_EQ(uint32_t(millis() - start), std::min(timeout, 6u));
+        EXPECT_EQ(source.hasPendingResponse(), shared && timeout < 6);
+      }
+    }
+  }
+}
+
+TEST(MotaSourceSerial, LargestRepliesFitExistingUsbUartAndWirelessDeadlines) {
+  for (const auto profile : {std::pair<uint32_t, uint32_t>{400, 2},
+                             std::pair<uint32_t, uint32_t>{600, 3},
+                             std::pair<uint32_t, uint32_t>{3000, 15}}) {
+    SCOPED_TRACE(profile.first);
+    resetArduinoMock();
+    TimedMotaSeederStream stream;
+    stream.delay_per_byte = profile.second;
+    SerialMotaSource source(stream, MotaStreamWritePolicy::NoFlush, profile.first);
+    std::array<uint8_t, MOTA_SEEDER_READ_MAX> data{};
+    ASSERT_TRUE(source.read(0, 0x1000, data.data(), data.size()));
+    EXPECT_EQ(millis(), uint32_t(MOTA_SEEDER_READ_MAX + 4) * profile.second);
+    EXPECT_LT(millis(), profile.first);
+    for (size_t i = 0; i < data.size(); ++i) EXPECT_EQ(data[i], uint8_t(i));
+  }
+}
+
+TEST(MotaSourceSerial, StaleDrainDoesNotChaseNewlyArrivingInput) {
+  resetArduinoMock();
+  TimedMotaSeederStream stream;
+  stream.stale_remaining = 4;
+  stream.stale_refills = 512;
+  SerialMotaSource source(stream, MotaStreamWritePolicy::NoFlush, 600);
+  uint8_t data = 0;
+  ASSERT_TRUE(source.read(0, 0x5A, &data, 1));
+  EXPECT_EQ(stream.stale_reads_at_request, 4u);
+  EXPECT_EQ(data, 0x5A);
+}
+
+TEST(MotaSourceSerial, ErrorChecksumAndWrongOpStayFailClosedAndRetryResynchronizes) {
+  for (unsigned failure = 0; failure < 3; ++failure) {
+    SCOPED_TRACE(failure);
+    resetArduinoMock();
+    TimedMotaSeederStream stream;
+    stream.error_status = failure == 0;
+    stream.corrupt_checksum = failure == 1;
+    stream.wrong_op = failure == 2;
+    SerialMotaSource source(stream, MotaStreamWritePolicy::NoFlush, 600);
+    uint8_t data[8] = {};
+    EXPECT_FALSE(source.read(0, 0x20, data, sizeof(data)));
+    stream.error_status = stream.corrupt_checksum = stream.wrong_op = false;
+    ASSERT_TRUE(source.read(0, 0x30, data, sizeof(data)));
+    EXPECT_EQ(stream.unreadResponse(), 0u);
+    for (size_t i = 0; i < sizeof(data); ++i) EXPECT_EQ(data[i], uint8_t(0x30 + i));
+  }
+}
+
+class ManualMotaSeederStream : public Stream {
+public:
+  using Stream::write;
+  std::vector<std::vector<uint8_t>> replies;
+  std::vector<uint8_t> input;
+  size_t position = 0, writes = 0;
+  size_t write(const uint8_t*, size_t len) override {
+    if (writes < replies.size()) push(replies[writes]);
+    ++writes;
+    return len;
+  }
+  void push(const std::vector<uint8_t>& bytes) {
+    input.insert(input.end(), bytes.begin(), bytes.end());
+  }
+  int available() override { return static_cast<int>(input.size() - position); }
+  int read() override { return position < input.size() ? input[position++] : -1; }
+};
+
+static std::vector<uint8_t> serialResponse(uint8_t op,
+                                         const std::vector<uint8_t>& payload,
+                                         uint8_t status = MS_STATUS_OK) {
+  std::vector<uint8_t> response = {'m', 's', op, status};
+  response.insert(response.end(), payload.begin(), payload.end());
+  uint8_t checksum = 0;
+  for (uint8_t value : response) checksum ^= value;
+  response.push_back(checksum);
+  return response;
+}
+
+static std::vector<uint8_t> serialControlBytes(SerialMotaSource& source) {
+  std::vector<uint8_t> result;
+  int pending = source.availableControlBytes();
+  while (pending-- > 0) {
+    const int value = source.readControlByte();
+    if (value == -1) break;
+    if (value >= 0) result.push_back(static_cast<uint8_t>(value));
+  }
+  return result;
+}
+
+TEST(MotaSourceSerial, LateReplyBoundariesCoverAllSourceOperationsAndErrorStatus) {
+  const char stop[] = "ota folder off\r\n";
+  const std::vector<uint8_t> controls(stop, stop + sizeof(stop) - 1);
+  for (uint8_t op : {MS_OP_COUNT, MS_OP_DESCRIBE, MS_OP_READ, MS_OP_DEFLATE_BLOCK}) {
+    for (bool error : {false, true}) {
+      std::vector<uint8_t> payload;
+      if (!error) {
+        if (op == MS_OP_COUNT) payload = {1};
+        else if (op == MS_OP_DESCRIBE) payload.assign(MOTA_DESC_WIRE, 'x');
+        else if (op == MS_OP_DEFLATE_BLOCK) payload = {17, 0};
+        else {
+          const char body[] = "x\nota folder off\n";
+          payload.assign(body, body + sizeof(body) - 1);
+        }
+      }
+      const auto response = serialResponse(op, payload, error ? MS_STATUS_ERR : MS_STATUS_OK);
+      for (size_t prefix = 0; prefix < response.size(); ++prefix) {
+        SCOPED_TRACE(op);
+        SCOPED_TRACE(error);
+        SCOPED_TRACE(prefix);
+        resetArduinoMock();
+        ManualMotaSeederStream stream;
+        stream.replies.emplace_back(response.begin(), response.begin() + prefix);
+        SerialMotaSource source(stream, MotaStreamWritePolicy::NoFlush, 4);
+        source.enableSharedTextControl();
+        uint8_t data[64] = {};
+        uint16_t encoded_len = 123;
+        MotaDesc desc;
+        if (op == MS_OP_COUNT) EXPECT_EQ(source.count(), 0);
+        else if (op == MS_OP_DESCRIBE) EXPECT_FALSE(source.describe(0, desc));
+        else if (op == MS_OP_DEFLATE_BLOCK) {
+          EXPECT_FALSE(source.read_deflated_block(0, 0, data, sizeof(data), &encoded_len));
+          EXPECT_EQ(encoded_len, 0u);
+        } else EXPECT_FALSE(source.read(0, 0, data, payload.size() ? payload.size() : 17));
+        EXPECT_TRUE(source.hasPendingResponse());
+        for (unsigned retry = 0; retry < 3; ++retry) {
+          EXPECT_FALSE(source.read(0, 0, data, 7));
+          EXPECT_EQ(stream.writes, 1u);
+          EXPECT_TRUE(source.hasPendingResponse());
+        }
+        stream.push(std::vector<uint8_t>(response.begin() + prefix, response.end()));
+        stream.push(controls);
+        EXPECT_EQ(serialControlBytes(source), controls);
+        EXPECT_FALSE(source.hasPendingResponse());
+      }
+    }
+  }
+}
+
+TEST(MotaSourceSerial, StaleDrainRetainsOldBodyLengthAcrossDifferentSizeRetries) {
+  resetArduinoMock();
+  ManualMotaSeederStream stream;
+  const auto old_response = serialResponse(MS_OP_READ, {'x', '\n', 'o', 't', 'a', ' ', 'f', 'o',
+                                                      'l', 'd', 'e', 'r', ' ', 'o', 'f', 'f', '\n'});
+  stream.replies.emplace_back(old_response.begin(), old_response.begin() + 6);
+  stream.replies.push_back(serialResponse(MS_OP_READ, {42, 43}));
+  SerialMotaSource source(stream, MotaStreamWritePolicy::NoFlush, 4);
+  source.enableSharedTextControl();
+  uint8_t old_data[17] = {};
+  EXPECT_FALSE(source.read(0, 0, old_data, sizeof(old_data)));
+  uint8_t new_data[2] = {};
+  for (size_t index = 6; index + 1 < old_response.size(); ++index) {
+    stream.push({old_response[index]});
+    EXPECT_FALSE(source.read(0, 99, new_data, sizeof(new_data)));
+    EXPECT_EQ(stream.writes, 1u);
+    EXPECT_TRUE(source.hasPendingResponse());
+    EXPECT_TRUE(serialControlBytes(source).empty());
+  }
+  stream.push({old_response.back()});
+  const char stop[] = "ota folder off\r\n";
+  const std::vector<uint8_t> controls(stop, stop + sizeof(stop) - 1);
+  stream.push(controls);
+  EXPECT_TRUE(source.read(0, 99, new_data, sizeof(new_data)));
+  EXPECT_EQ(stream.writes, 2u);
+  EXPECT_EQ(new_data[0], 42u);
+  EXPECT_EQ(new_data[1], 43u);
+  EXPECT_EQ(serialControlBytes(source), controls);
+  EXPECT_FALSE(source.hasPendingResponse());
+}
+
+TEST(MotaSourceSerial, LateDeflateChunkLengthIncludesItsTwoByteSizePrefix) {
+  const char body[] = "x\nota folder off\n";
+  std::vector<uint8_t> payload = {sizeof(body) - 1, 0};
+  payload.insert(payload.end(), body, body + sizeof(body) - 1);
+  const auto delayed = serialResponse(MS_OP_DEFLATE_BLOCK, payload);
+  const char stop[] = "ota folder off\r\n";
+  const std::vector<uint8_t> controls(stop, stop + sizeof(stop) - 1);
+  for (size_t prefix = 0; prefix < delayed.size(); ++prefix) {
+    SCOPED_TRACE(prefix);
+    resetArduinoMock();
+    ManualMotaSeederStream stream;
+    stream.replies.push_back(serialResponse(MS_OP_DEFLATE_BLOCK, {sizeof(body) - 1, 0}));
+    stream.replies.emplace_back(delayed.begin(), delayed.begin() + prefix);
+    SerialMotaSource source(stream, MotaStreamWritePolicy::NoFlush, 4);
+    source.enableSharedTextControl();
+    uint8_t data[64] = {};
+    uint16_t encoded_len = 123;
+    EXPECT_FALSE(source.read_deflated_block(0, 0, data, sizeof(data), &encoded_len));
+    EXPECT_EQ(encoded_len, 0u);
+    EXPECT_EQ(stream.writes, 2u);
+    EXPECT_TRUE(source.hasPendingResponse());
+    stream.push(std::vector<uint8_t>(delayed.begin() + prefix, delayed.end()));
+    stream.push(controls);
+    EXPECT_EQ(serialControlBytes(source), controls);
+    EXPECT_FALSE(source.hasPendingResponse());
+  }
+}
+
+TEST(MotaSourceSerial, ControlQueueOverflowRejectsWholeLineAndResetClearsQuarantine) {
+  resetArduinoMock();
+  ManualMotaSeederStream stream;
+  std::vector<uint8_t> malformed(100, 'x');
+  const char suffix[] = "ota folder off\n";
+  malformed.insert(malformed.end(), suffix, suffix + sizeof(suffix) - 1);
+  stream.push(malformed);
+  SerialMotaSource source(stream, MotaStreamWritePolicy::NoFlush, 4);
+  source.enableSharedTextControl();
+  uint8_t data = 0;
+  EXPECT_FALSE(source.read(0, 0, &data, 1));
+  EXPECT_TRUE(source.hasPendingResponse());
+  EXPECT_EQ(serialControlBytes(source), (std::vector<uint8_t>{0, '\n'}));
+  source.resetSessionState();
+  EXPECT_FALSE(source.hasPendingResponse());
+  EXPECT_EQ(source.availableControlBytes(), 0);
+  stream.replies.resize(2);
+  stream.replies[1] = serialResponse(MS_OP_READ, {42});
+  EXPECT_TRUE(source.read(0, 0, &data, 1));
+  EXPECT_EQ(data, 42u);
+}
+
+TEST(MotaSourceSerial, DefaultBleRetryAndNewRingSessionPreserveLegacyBehavior) {
+  resetArduinoMock();
+  BleMotaStream stream;
+  struct Host {
+    BleMotaStream* stream;
+    unsigned requests = 0;
+    std::vector<uint8_t> reply;
+  } host{&stream, 0, {'m', 's', MS_OP_READ, MS_STATUS_OK, 'x'}};
+  stream.setSender([](void* context, const uint8_t*, size_t len) {
+    auto& host = *static_cast<Host*>(context);
+    ++host.requests;
+    if (!host.reply.empty()) EXPECT_TRUE(host.stream->pushRx(host.reply.data(), host.reply.size()));
+    return len;
+  }, &host);
+  stream.setActive(true);
+  SerialMotaSource source(stream, MotaStreamWritePolicy::NoFlush, 4);
+  uint8_t data[17] = {};
+  EXPECT_FALSE(source.read(0, 0, data, sizeof(data)));
+  EXPECT_FALSE(source.hasPendingResponse());
+  host.reply = serialResponse(MS_OP_COUNT, {1});
+  EXPECT_EQ(source.count(), 1);
+  EXPECT_EQ(host.requests, 2u); // Default non-text transports retain ordinary retries.
+  stream.setActive(false);
+  stream.setActive(true); // New generation drops the old ring contents/writers.
+  EXPECT_EQ(stream.available(), 0);
+  EXPECT_FALSE(source.hasPendingResponse());
+  source.resetSessionState(); // The production BLE start pairs these two resets.
+  host.reply = serialResponse(MS_OP_COUNT, {1});
+  EXPECT_EQ(source.count(), 1);
+  EXPECT_EQ(host.requests, 3u);
+  EXPECT_FALSE(source.hasPendingResponse());
+}
+
+TEST(MotaSourceSerial, DefaultRetryCanShareTcpWithLegacyFolderStoreReads) {
+  resetArduinoMock();
+  ManualMotaSeederStream stream;
+  stream.replies.push_back({'m', 's', MS_OP_COUNT, MS_STATUS_OK});
+  stream.replies.push_back(serialResponse(MS_OP_STAT, {1, 13, 0, 0, 0}));
+  stream.replies.push_back(serialResponse(MS_OP_COUNT, {1}));
+  SerialMotaSource source(stream, MotaStreamWritePolicy::NoFlush, 4);
+  FolderMotaStore store(stream, MotaStreamWritePolicy::NoFlush, 4);
+  EXPECT_EQ(source.count(), 0);
+  EXPECT_FALSE(source.hasPendingResponse());
+  const auto old_response = serialResponse(MS_OP_COUNT, {1});
+  stream.push(std::vector<uint8_t>(old_response.begin() + 4, old_response.end()));
+  EXPECT_TRUE(store.reopen()); // Its independent legacy drain consumes the old tail.
+  EXPECT_EQ(store.staged_size(), 13u);
+  EXPECT_EQ(source.count(), 1); // No USB-only retained state poisons this source.
+  EXPECT_EQ(stream.writes, 3u);
+  EXPECT_FALSE(source.hasPendingResponse());
+}
+
 class FakeFolderMotaSeederStream : public Stream {
 public:
   using Stream::write;