Fix Cascade power saving, USB Companion, and mOTA flows

test/README.md

@@ -1,16 +1,17 @@
 # Host unit tests
 
-Fast, hardware-free unit tests for the fork's pure logic, run on the host with
-GoogleTest via PlatformIO's `native` environment. They cover the extractable
-observer/WebConfig logic (validation, preset table, topic templates, key
-parsing) -- the parts that don't depend on the ESP32, radio, or network stack.
-Integration behavior (AsyncTCP transport, WiFi/MQTT, SoftAP) is exercised
-separately; see "Local testing without hardware" in `MQTT_IMPLEMENTATION.md`.
+Fast, hardware-free unit tests for pure and host-simulated logic, run with
+GoogleTest through PlatformIO. The `native` environment runs every suite except
+the KISS modem; `native_kiss_modem` builds that suite with its separate source
+filter. Hardware, real radio, AsyncTCP, Wi-Fi/MQTT, and SoftAP behavior still
+require integration or target testing; see "Local testing without hardware" in
+`MQTT_IMPLEMENTATION.md`.
 
 ## Running
 
 ```sh
-pio test -e native                      # all suites
+pio test -e native                      # all suites except KISS modem
+pio test -e native_kiss_modem           # KISS modem suite
 pio test -e native -f test_webconfig_keys   # a single suite
 ```
 
@@ -47,8 +48,28 @@ does not reflect the GoogleTest count -- run the built binary directly
 | `test_identity_generation` | `src/helpers/IdentityGeneration.h` | reserved-prefix rejection; bounded retries; final provisioned attempt; fail-closed exhaustion |
 | `test_remote_cli_reply_cache` | `src/helpers/RemoteCliReplyCache.h`, `src/helpers/RemoteCliRequest.h`, `src/helpers/RemoteCliTimeout.h` | authenticated logical-request matching; bounded recent-reply history; backward-compatible retry identity; 300% response timeout; empty-response completion; on-air truncation and clearing |
 | `test_companion_frame_queue` | `src/helpers/CompanionFrameQueue.h` | response/required/best-effort classification; reserved capacity; stable priority; safe eviction; message-waiting coalescing |
-| `test_serial_mode_switch` | `src/helpers/ArduinoSerialInterface.cpp`, `src/helpers/MultiSerialInterface.h` | independent terminal/seeder control-sequence recognition across reads and binary-frame boundaries; passthrough ownership of USB input and suppression of binary output; Bluetooth-only connection and pairing-request routing |
+| `test_serial_mode_switch` | `src/helpers/ArduinoSerialInterface.cpp`, `src/helpers/MultiSerialInterface.h` | terminal/seeder control-sequence recognition and passthrough ownership; queued/atomic USB output under backpressure and short writes; partial-frame busy state; requester-affine replies, locked contact streams, and Bluetooth-only pairing routing |
 | `test_ble_tx_stall_watchdog` | `src/helpers/BleTxStallWatchdog.h` | exact BLE fragment progress; blocked-reply timeout; rollover-safe elapsed time; disconnect recovery retry and completion |
+| `test_atomic_file_writer` | `src/helpers/AtomicFileWriter.h` | verified temporary-file commit; short-write, readback, validation, and rename failures; preservation of the live file and stale-temp cleanup |
+| `test_cad_timing` | `src/helpers/radiolib/CadTiming.h`, `LR2021SideDetectorConfig.h`, `RadioAirtime.h` | Cascade and slow-profile CAD deadlines; invalid airtime handling; bounded LR2021 side-detector parsing and LDRO recomputation |
+| `test_companion_node_prefs` | `examples/companion_radio/NodePrefs.h` | independent device power saving, RXPS, Wi-Fi, and FEM preferences; one-time migration of the regressed power-saving default |
+| `test_config_serializer` | `src/helpers/ConfigSerializer.cpp`, Companion `NodePrefs` | escaped config save/load, whitespace and malformed input, unknown fields, and FEM preference round trips |
+| `test_deferred_cli_command` | `src/helpers/DeferredCliCommand.h` | copying authenticated command context, single-pending-command enforcement, clearing, and length rejection |
+| `test_kiss_modem` | `examples/kiss_modem/KissModem.cpp` | KISS escaping/framing and packet metadata under partial writes, host TX backpressure, queue saturation, and radio completion; run with `native_kiss_modem` |
+| `test_mesh_tables` | `src/helpers/SimpleMeshTables.h` | packet and ACK/multipart deduplication, scope-independent identity, route-prefix matching, and deterministic recent-repeater expiry/eviction |
+| `test_mqtt_lifecycle` | `src/helpers/MQTTLifecycle.h` | idempotent start/stop, initialization rollback, cooperative stop acknowledgment and timeout, callback ownership, restart, and the OTA flash barrier |
+| `test_mqtt_reply_format` | `src/helpers/MQTTReplyFormat.h` | bounded formatted appends, exact-fit and one-byte buffers, truncation, NUL termination, and invalid starting positions |
+| `test_packet_manager` | `src/Packet.cpp`, `src/Dispatcher.cpp`, `src/helpers/StaticPoolPacketManager.cpp` | truncated-packet rejection, unavailable-radio behavior, scoped RX-delay replacement, queue/CAD scheduling, and staged radio/TX recovery |
+| `test_persistent_store_format` | `src/helpers/PersistentStoreFormat.h` | contact-page headers and CRCs, dirty-page state, stable slot allocation, and bounded resumable legacy migration across power loss |
+| `test_power_management` | `src/helpers/PowerManagementUtils.h` | median filtering of a brownout outlier and valid-reading requirements for the boot lock |
+| `test_region_names` | `src/helpers/RegionNameUtils.h` | canonical public-region markers while preserving distinct private and differently named regions |
+| `test_routing_policy` | `src/helpers/RoutingPolicy.h` | scoped/unscoped flood hop limits and selection of direct, path-return, mirrored-scope, default-scope, or unscoped replies |
+| `test_rs232_uart` | `src/helpers/bridges/RS232UartUtils.h` | stopping the active UART peripheral before reassigning its pins |
+| `test_security_session_timer` | `src/helpers/nrf52/SecuritySessionTimer.h` | two-minute security-session expiry, cancellation, restart, and `millis()` rollover |
+| `test_trace_path_helpers` | `src/helpers/TracePathHelpers.h` | round-trip route construction, hash-width conversion, raw path parsing/limits, and terminal trace timeout bounds |
+| `test_user_gpio` | `src/helpers/UserGpio.cpp`, `UserGpioReplyTracker.h` | board-approved pins, get/set/reset, timed nonblocking transitions, duplicate suppression, rollover, and completion-reply routing |
+| `test_utf8_helpers` | `src/helpers/UTF8Helpers.h` | byte-limit truncation at complete code-point boundaries and rejection of malformed or truncated UTF-8 |
+| `test_wifi_ota_seeder_policy` | `src/helpers/WiFiOtaSeederPolicy.h`, `WiFiOtaSeederStatus.h` | listener state versus network availability, serial/TCP folder ownership, detach detection, and bounded status formatting |
 | `test_ota` | `src/helpers/ota/` | container and EndF integrity; protocol codecs; transfer, resume, and apply safety; adaptive 2-to-4 block-request window growth and stall contraction; active-transfer priority classification |
 | `test_trace_retry` | `src/Mesh.cpp` retry and relay policy | opaque OTA relay behavior during TempRadio; background discovery priority; immediate primary transfer relay, receive-delay bypass, fast CAD retry, and no generic flood retry; trace and non-OTA flood retry timing |
 | `test_utils` | `src/Utils.cpp` | `Utils::toHex` (upstream) |

test/test_ota/test_ota_core.cpp

@@ -1428,6 +1428,20 @@ TEST(OtaCatalog, RejectsAdvertisedSourceWithOversizedBlocks) {
   EXPECT_EQ(server.servedCount(), 0);
 }
 
+TEST(OtaFolder, ReportsEntriesOmittedByServeRegistryCapacity) {
+  OtaManager server;
+  server.begin(0, nullptr, nullptr);
+  ASSERT_TRUE(server.serve(SIM_MOTA, SIM_MOTA_LEN));
+  SyntheticCatalogSource source((uint8_t)(OTA_MAX_SERVE + 3));
+  ASSERT_TRUE(server.add_source(&source));
+
+  uint16_t offered = 0, advertised = 0;
+  ASSERT_TRUE(server.sourceStats(&source, offered, advertised));
+  EXPECT_EQ(offered, (uint16_t)(OTA_MAX_SERVE + 3));
+  EXPECT_EQ(advertised, (uint16_t)(OTA_MAX_SERVE - 1)); // primary image occupies slot zero
+  EXPECT_EQ(server.servedCount(), OTA_MAX_SERVE);
+}
+
 // Fetch-resume across a reboot: a client commits some blocks, "reboots" (a fresh OtaManager on the SAME
 // persisted store), and resumeStaged() re-adopts the partial container and finishes the remaining blocks -
 // without re-fetching the manifest or the blocks already present.
@@ -1633,6 +1647,49 @@ TEST(OtaTransfer, RejectsIncompatibleCodec) {
   g_q.clear();
 }
 
+TEST(OtaTransfer, ManualPullReportsIncompatibleManifestInsteadOfGoingIdle) {
+  g_q.clear();
+  OtaManager server, client;
+  OtaStoreRam<4096> store;
+  SendTo to_client{&client}, to_server{&server};
+  server.begin(0, sim_send, &to_client);
+  client.begin(SIM_TARGET_ID, sim_send, &to_server);
+  client.set_fetch_store(&store);
+  client.set_apply_codec(CODEC_DETOOLS_INPLACE);
+  client.set_accept_full(false);
+  ASSERT_TRUE(server.serve(SIM_MOTA, SIM_MOTA_LEN));
+  MotaManifest manifest;
+  ASSERT_TRUE(mota_parse(SIM_MOTA, SIM_MOTA_LEN, manifest));
+
+  EXPECT_EQ(client.pull(manifest.merkle_root, manifest.target_id), OtaManager::PULL_STARTED);
+  pump(client, &server);
+
+  EXPECT_EQ(client.fetchState(), OtaManager::FAILED);
+  EXPECT_EQ(client.fetchError(), OtaManager::FETCH_ERROR_CODEC);
+  EXPECT_EQ(client.wanted(), 0u);                    // terminal rejection no longer leaves discovery armed
+  EXPECT_EQ(store.staged_size(), 0u);                // compatibility is rejected before the store is begun
+}
+
+TEST(OtaTransfer, PullAdmissionReportsNoStoreAndBusyWithoutReplacingIntent) {
+  g_q.clear();
+  OtaManager client;
+  SendTo to_none{&client};
+  client.begin(SIM_TARGET_ID, sim_send, &to_none);
+  uint8_t first[4] = {1, 2, 3, 4}, second[4] = {5, 6, 7, 8};
+  EXPECT_EQ(client.pull(first, SIM_TARGET_ID), OtaManager::PULL_NO_STORE);
+  EXPECT_EQ(client.fetchState(), OtaManager::IDLE);
+  EXPECT_EQ(client.wanted(), 0u);
+
+  OtaStoreRam<4096> store;
+  client.set_fetch_store(&store);
+  EXPECT_EQ(client.pull(first, SIM_TARGET_ID), OtaManager::PULL_STARTED);
+  EXPECT_EQ(client.pull(second, SIM_TARGET_ID ^ 0x55AAu), OtaManager::PULL_BUSY);
+  EXPECT_EQ(client.fetchState(), OtaManager::WANT_MANIFEST);
+  EXPECT_EQ(client.wanted(), SIM_TARGET_ID);
+  EXPECT_EQ(0, memcmp(client.fetchManifestId(), first, sizeof(first)));
+  g_q.clear();
+}
+
 // An archive capture is not an install. It must retain cross-target and otherwise unsupported containers
 // byte-for-byte so this node can relay them to hardware that does understand their codec.
 TEST(OtaTransfer, ArchivePullAcceptsCrossTargetUnsupportedCodec) {
@@ -1660,14 +1717,16 @@ TEST(OtaTransfer, ArchivePullAcceptsCrossTargetUnsupportedCodec) {
 // Encode a 1-row OTA_HAVE from a specific seeder, carrying have_count (Phase-2 awareness).
 static uint16_t make_have_row(uint8_t* buf, uint16_t cap, const uint8_t mid[4], uint32_t target,
                               uint32_t fwver, uint8_t codec, uint8_t flags,
-                              const uint8_t seeder[4], uint16_t have_count) {
+                              const uint8_t seeder[4], uint16_t have_count,
+                              const uint8_t digest[4] = nullptr) {
   uint8_t row[OTA_HAVE_ROW_BYTES];
   memcpy(row, mid, 4);
   row[4]=target; row[5]=target>>8; row[6]=target>>16; row[7]=target>>24;
   row[8]=fwver; row[9]=fwver>>8; row[10]=fwver>>16; row[11]=fwver>>24;
   row[12]=codec; row[13]=flags;
   row[14]=(uint8_t)(have_count & 0xFF); row[15]=(uint8_t)(have_count >> 8);
-  HaveMsg hv; memcpy(hv.seeder_id, seeder, 4); memset(hv.set_digest, 0, 4);
+  HaveMsg hv; memcpy(hv.seeder_id, seeder, 4);
+  if (digest) memcpy(hv.set_digest, digest, 4); else memset(hv.set_digest, 0, 4);
   hv.frag_idx=0; hv.frag_total=1; hv.n_rows=1; hv.rows=row;
   return encode_have(buf, cap, hv);
 }
@@ -1689,6 +1748,49 @@ TEST(OtaCatalog, DistinctSeederCountAndHaveCount) {
   g_q.clear();
 }
 
+TEST(OtaCatalog, DigestChangePurgesOnlyThatSeedersRowsAndProgress) {
+  OtaManager m; SendTo none{&m}; m.begin(SIM_TARGET_ID, sim_send, &none);
+  m.set_archive_interest(true);
+  uint8_t wire[64], mid[4] = {9, 8, 7, 6};
+  uint8_t s1[4] = {1, 0, 0, 0}, s2[4] = {2, 0, 0, 0};
+  uint8_t d1[4] = {0x11, 0, 0, 0}, d2[4] = {0x22, 0, 0, 0}, changed[4] = {0x33, 0, 0, 0};
+
+  auto advertise = [&](const uint8_t sid[4], const uint8_t digest[4], uint8_t count = 1) {
+    AdvMsg adv{}; memcpy(adv.seeder_id, sid, 4); memcpy(adv.set_digest, digest, 4); adv.n_motas = count;
+    uint16_t n = encode_adv(wire, sizeof(wire), adv);
+    ASSERT_GT(n, 0);
+    m.on_message(wire, n);
+  };
+  advertise(s1, d1);
+  advertise(s2, d2);
+  m.set_clock(100);
+  m.on_message(wire, make_have_row(wire, sizeof(wire), mid, SIM_TARGET_ID, 0x01020300,
+                                   CODEC_FULL, MFLAG_FULL, s1, 7, d1));
+  m.set_clock(200);
+  m.on_message(wire, make_have_row(wire, sizeof(wire), mid, SIM_TARGET_ID, 0x01020300,
+                                   CODEC_FULL, MFLAG_FULL, s2, 3, d2));
+  ASSERT_EQ(m.catalogCount(), 1);
+  ASSERT_EQ(m.catalogRow(0)->n_seeders, 2);
+  ASSERT_EQ(m.catalogRow(0)->have_max, 7u);
+
+  advertise(s1, changed);
+  ASSERT_EQ(m.catalogCount(), 1);
+  EXPECT_EQ(m.catalogRow(0)->n_seeders, 1);
+  EXPECT_EQ(m.catalogRow(0)->have_max, 3u);
+  EXPECT_EQ(m.catalogRow(0)->last_ms, 200u);
+
+  m.set_clock(300);
+  m.on_message(wire, make_have_row(wire, sizeof(wire), mid, SIM_TARGET_ID, 0x01020300,
+                                   CODEC_FULL, MFLAG_FULL, s1, 9, d1)); // delayed row from the old digest
+  ASSERT_EQ(m.catalogCount(), 1);
+  EXPECT_EQ(m.catalogRow(0)->n_seeders, 1);
+  EXPECT_EQ(m.catalogRow(0)->have_max, 3u);
+
+  advertise(s2, changed, 0);                         // an explicit empty advert withdraws the source
+  EXPECT_EQ(m.catalogCount(), 0);                    // nobody still advertises the old set
+  g_q.clear();
+}
+
 // An unanswered GET_MANIFEST must not pin the fetch slot forever: after OTA_MANIFEST_MAX_RETRY ticks with
 // no manifest, the session gives up (FAILED) so a new pull can take the slot. (Bounded primary operation.)
 TEST(OtaTransfer, ManifestGiveUpAfterRetries) {
@@ -1701,6 +1803,8 @@ TEST(OtaTransfer, ManifestGiveUpAfterRetries) {
   EXPECT_EQ(client.fetchState(), OtaManager::WANT_MANIFEST);
   for (int i = 0; i < OTA_MANIFEST_MAX_RETRY + 2; i++) { g_clk += 5000; client.set_clock(g_clk); client.loop(); g_q.clear(); }
   EXPECT_EQ(client.fetchState(), OtaManager::FAILED);
+  EXPECT_EQ(client.fetchError(), OtaManager::FETCH_ERROR_MANIFEST_TIMEOUT);
+  EXPECT_EQ(client.wanted(), 0u);
 }
 
 // A receiver never becomes a source, either while fetching or after completion.

test/test_serial_mode_switch/test_serial_mode_switch.cpp

@@ -1,6 +1,8 @@
 #include <gtest/gtest.h>
 
+#include <algorithm>
 #include <deque>
+#include <limits>
 #include <vector>
 
 #include "helpers/ArduinoSerialInterface.h"
@@ -10,6 +12,8 @@ class BufferStream : public Stream {
 public:
   std::deque<uint8_t> input;
   std::vector<uint8_t> output;
+  int write_capacity = 4096;
+  size_t max_write = std::numeric_limits<size_t>::max();
 
   void push(const char* data) {
     while (*data) input.push_back((uint8_t)*data++);
@@ -20,6 +24,7 @@ public:
   }
 
   int available() override { return (int)input.size(); }
+  int availableForWrite() override { return write_capacity; }
 
   int read() override {
     if (input.empty()) return -1;
@@ -29,13 +34,15 @@ public:
   }
 
   size_t write(uint8_t value) override {
-    output.push_back(value);
-    return 1;
+    return write(&value, 1);
   }
 
   size_t write(const uint8_t* data, size_t len) override {
-    output.insert(output.end(), data, data + len);
-    return len;
+    size_t accepted = std::min(len, max_write);
+    accepted = std::min(accepted, (size_t)std::max(write_capacity, 0));
+    output.insert(output.end(), data, data + accepted);
+    write_capacity -= (int)accepted;
+    return accepted;
   }
 };
 
@@ -47,20 +54,32 @@ public:
   bool enabled = false;
   bool connected = false;
   bool pairing_request = false;
+  bool write_busy = false;
+  std::deque<std::vector<uint8_t>> received_frames;
+  std::vector<std::vector<uint8_t>> sent_frames;
 
   void enable() override { enabled = true; }
   void disable() override { enabled = false; }
   bool isEnabled() const override { return enabled; }
   bool isConnected() const override { return connected; }
   bool isReadBusy() const override { return false; }
-  bool isWriteBusy() const override { return false; }
+  bool isWriteBusy() const override { return write_busy; }
   bool takePairingRequest() override {
     bool pending = pairing_request;
     pairing_request = false;
     return pending;
   }
-  size_t writeFrame(const uint8_t[], size_t len) override { return len; }
-  size_t checkRecvFrame(uint8_t[]) override { return 0; }
+  size_t writeFrame(const uint8_t src[], size_t len) override {
+    sent_frames.emplace_back(src, src + len);
+    return len;
+  }
+  size_t checkRecvFrame(uint8_t dest[]) override {
+    if (received_frames.empty()) return 0;
+    const std::vector<uint8_t> frame = received_frames.front();
+    received_frames.pop_front();
+    memcpy(dest, frame.data(), frame.size());
+    return frame.size();
+  }
 };
 
 TEST(MultiSerialInterface, TracksBluetoothConnectionSeparately) {
@@ -99,6 +118,120 @@ TEST(MultiSerialInterface, PairingRequestsComeOnlyFromBluetooth) {
   EXPECT_FALSE(manager.takePairingRequest());
 }
 
+TEST(MultiSerialInterface, RoutesRequiredRepliesToTheirRequestingInterface) {
+  MultiSerialInterface manager;
+  FakeSerialInterface usb;
+  FakeSerialInterface wifi;
+  usb.connected = true;
+  wifi.connected = true;
+  ASSERT_TRUE(manager.addInterface(InterfaceType::USB, &usb));
+  ASSERT_TRUE(manager.addInterface(InterfaceType::WiFi, &wifi));
+  manager.enable();
+
+  usb.received_frames.push_back({0x01});
+  uint8_t command[MAX_FRAME_SIZE] = {};
+  ASSERT_EQ(manager.checkRecvFrame(command), 1u);
+
+  const uint8_t response[] = {0x05, 0xAA};
+  EXPECT_EQ(manager.writeFrame(response, sizeof(response)), sizeof(response));
+  ASSERT_EQ(usb.sent_frames.size(), 1u);
+  EXPECT_TRUE(wifi.sent_frames.empty());
+
+  // Login/status-style pushes complete a client operation and follow the same
+  // requester route rather than exposing the result on another transport.
+  const uint8_t required_push[] = {0x85, 0xBB};
+  EXPECT_EQ(manager.writeFrame(required_push, sizeof(required_push)),
+            sizeof(required_push));
+  ASSERT_EQ(usb.sent_frames.size(), 2u);
+  EXPECT_TRUE(wifi.sent_frames.empty());
+
+  // Passive observations remain visible to every enabled client.
+  const uint8_t best_effort_push[] = {0x80, 0xCC};
+  EXPECT_EQ(manager.writeFrame(best_effort_push, sizeof(best_effort_push)),
+            sizeof(best_effort_push));
+  ASSERT_EQ(usb.sent_frames.size(), 3u);
+  ASSERT_EQ(wifi.sent_frames.size(), 1u);
+}
+
+TEST(MultiSerialInterface, LocksMultiFrameRepliesToOneRequester) {
+  MultiSerialInterface manager;
+  FakeSerialInterface usb;
+  FakeSerialInterface wifi;
+  usb.connected = true;
+  wifi.connected = true;
+  ASSERT_TRUE(manager.addInterface(InterfaceType::USB, &usb));
+  ASSERT_TRUE(manager.addInterface(InterfaceType::WiFi, &wifi));
+  manager.enable();
+
+  usb.received_frames.push_back({0x04});
+  uint8_t command[MAX_FRAME_SIZE] = {};
+  ASSERT_EQ(manager.checkRecvFrame(command), 1u);
+  manager.lockReplyRoute();
+
+  wifi.received_frames.push_back({0x16});
+  EXPECT_EQ(manager.checkRecvFrame(command), 0u);
+  EXPECT_EQ(wifi.received_frames.size(), 1u);
+
+  const uint8_t contact[] = {0x03, 0x42};
+  EXPECT_EQ(manager.writeFrame(contact, sizeof(contact)), sizeof(contact));
+  ASSERT_EQ(usb.sent_frames.size(), 1u);
+  EXPECT_TRUE(wifi.sent_frames.empty());
+
+  manager.unlockReplyRoute();
+  ASSERT_EQ(manager.checkRecvFrame(command), 1u);
+  EXPECT_EQ(command[0], 0x16);
+  const uint8_t device_info[] = {0x0D, 0x43};
+  EXPECT_EQ(manager.writeFrame(device_info, sizeof(device_info)),
+            sizeof(device_info));
+  ASSERT_EQ(wifi.sent_frames.size(), 1u);
+}
+
+TEST(MultiSerialInterface, LosingLockedRequesterCannotFallBackToBroadcast) {
+  MultiSerialInterface manager;
+  FakeSerialInterface usb;
+  FakeSerialInterface bluetooth;
+  usb.connected = true;
+  bluetooth.connected = true;
+  ASSERT_TRUE(manager.addInterface(InterfaceType::USB, &usb));
+  ASSERT_TRUE(manager.addInterface(InterfaceType::Bluetooth, &bluetooth));
+  manager.enable();
+
+  bluetooth.received_frames.push_back({0x04});
+  uint8_t command[MAX_FRAME_SIZE] = {};
+  ASSERT_EQ(manager.checkRecvFrame(command), 1u);
+  manager.lockReplyRoute();
+  manager.disableBluetooth();
+
+  EXPECT_FALSE(manager.isConnected());
+  const uint8_t contact[] = {0x03, 0x42};
+  EXPECT_EQ(manager.writeFrame(contact, sizeof(contact)), 0u);
+  EXPECT_TRUE(usb.sent_frames.empty());
+
+  manager.unlockReplyRoute();
+  EXPECT_TRUE(manager.isConnected());
+}
+
+TEST(MultiSerialInterface, PacesOnlyTheActiveReplyTransport) {
+  MultiSerialInterface manager;
+  FakeSerialInterface usb;
+  FakeSerialInterface wifi;
+  usb.connected = true;
+  wifi.connected = true;
+  wifi.write_busy = true;
+  ASSERT_TRUE(manager.addInterface(InterfaceType::USB, &usb));
+  ASSERT_TRUE(manager.addInterface(InterfaceType::WiFi, &wifi));
+  manager.enable();
+
+  usb.received_frames.push_back({0x04});
+  uint8_t command[MAX_FRAME_SIZE] = {};
+  ASSERT_EQ(manager.checkRecvFrame(command), 1u);
+  EXPECT_FALSE(manager.isWriteBusy());
+
+  wifi.received_frames.push_back({0x04});
+  ASSERT_EQ(manager.checkRecvFrame(command), 1u);
+  EXPECT_TRUE(manager.isWriteBusy());
+}
+
 TEST(SerialModeSwitch, RecognizesControlSequenceAcrossReads) {
   BufferStream stream;
   ArduinoSerialInterface interface;
@@ -230,6 +363,123 @@ TEST(SerialModeSwitch, PassthroughLeavesInputAndSuppressesBinaryOutput) {
   EXPECT_EQ(stream.output[0], '>');
 }
 
+TEST(SerialFlowControl, KeepsAFrameQueuedUntilUsbHasSpace) {
+  BufferStream stream;
+  stream.write_capacity = 0;
+  ArduinoSerialInterface interface;
+  interface.begin(stream);
+  interface.enableFlowControl(true);
+  interface.enable();
+
+  const uint8_t payload[] = {0x05, 0xA5, 0x5A};
+  EXPECT_EQ(interface.writeFrame(payload, sizeof(payload)), sizeof(payload));
+  EXPECT_TRUE(stream.output.empty());
+  EXPECT_TRUE(interface.hasPendingIO());
+  EXPECT_TRUE(interface.isWriteBusy());
+
+  stream.write_capacity = MAX_FRAME_SIZE + 3;
+  interface.loop();
+  const std::vector<uint8_t> expected = {'>', 3, 0, 0x05, 0xA5, 0x5A};
+  EXPECT_EQ(stream.output, expected);
+  EXPECT_FALSE(interface.hasPendingIO());
+}
+
+TEST(SerialFlowControl, FinishesAnUnexpectedShortWriteBeforeNextFrame) {
+  BufferStream stream;
+  stream.write_capacity = MAX_FRAME_SIZE + 3;
+  stream.max_write = 2;
+  ArduinoSerialInterface interface;
+  interface.begin(stream);
+  interface.enableFlowControl(true);
+  interface.enable();
+
+  const uint8_t first[] = {0x05, 0x11, 0x22};
+  const uint8_t second[] = {0x00};
+  EXPECT_EQ(interface.writeFrame(first, sizeof(first)), sizeof(first));
+  ASSERT_EQ(stream.output.size(), 2u);
+  EXPECT_EQ(interface.writeFrame(second, sizeof(second)), sizeof(second));
+
+  stream.max_write = std::numeric_limits<size_t>::max();
+  interface.loop();
+  const std::vector<uint8_t> expected = {
+      '>', 3, 0, 0x05, 0x11, 0x22,
+      '>', 1, 0, 0x00};
+  EXPECT_EQ(stream.output, expected);
+  EXPECT_FALSE(interface.hasPendingIO());
+}
+
+TEST(SerialFlowControl, DrainsFramesThroughAFifoSmallerThanTheFrame) {
+  BufferStream stream;
+  stream.write_capacity = 4;
+  ArduinoSerialInterface interface;
+  interface.begin(stream);
+  interface.enableFlowControl(true);
+  interface.enable();
+
+  const uint8_t payload[] = {0x05, 1, 2, 3, 4, 5};
+  EXPECT_EQ(interface.writeFrame(payload, sizeof(payload)), sizeof(payload));
+  ASSERT_EQ(stream.output.size(), 4u);
+  EXPECT_TRUE(interface.hasPendingIO());
+
+  stream.write_capacity = 4;
+  interface.loop();
+  ASSERT_EQ(stream.output.size(), 8u);
+  EXPECT_TRUE(interface.hasPendingIO());
+
+  stream.write_capacity = 4;
+  interface.loop();
+  const std::vector<uint8_t> expected = {'>', 6, 0, 0x05, 1, 2, 3, 4, 5};
+  EXPECT_EQ(stream.output, expected);
+  EXPECT_FALSE(interface.hasPendingIO());
+}
+
+TEST(SerialFlowControl, PartialInboundFrameKeepsTransportBusy) {
+  BufferStream stream;
+  ArduinoSerialInterface interface;
+  interface.begin(stream);
+  interface.enable();
+  uint8_t frame[MAX_FRAME_SIZE] = {};
+
+  const uint8_t partial[] = {'<', 2};
+  stream.push(partial, sizeof(partial));
+  EXPECT_EQ(interface.checkRecvFrame(frame), 0u);
+  EXPECT_TRUE(interface.isReadBusy());
+  EXPECT_TRUE(interface.hasPendingIO());
+
+  const uint8_t remainder[] = {0, 0xA5, 0x5A};
+  stream.push(remainder, sizeof(remainder));
+  EXPECT_EQ(interface.checkRecvFrame(frame), 2u);
+  EXPECT_FALSE(interface.isReadBusy());
+  EXPECT_EQ(frame[0], 0xA5);
+  EXPECT_EQ(frame[1], 0x5A);
+}
+
+TEST(SerialFlowControl, AbandonsATruncatedInboundFrameAfterTimeout) {
+  resetArduinoMock();
+  BufferStream stream;
+  ArduinoSerialInterface interface;
+  interface.begin(stream);
+  interface.enable();
+  uint8_t frame[MAX_FRAME_SIZE] = {};
+
+  const uint8_t partial[] = {'<', 2, 0, 0xA5};
+  stream.push(partial, sizeof(partial));
+  EXPECT_EQ(interface.checkRecvFrame(frame), 0u);
+  EXPECT_TRUE(interface.isReadBusy());
+
+  delay(999);
+  interface.loop();
+  EXPECT_TRUE(interface.isReadBusy());
+  delay(1);
+  interface.loop();
+  EXPECT_FALSE(interface.isReadBusy());
+
+  const uint8_t complete[] = {'<', 1, 0, 0x5A};
+  stream.push(complete, sizeof(complete));
+  EXPECT_EQ(interface.checkRecvFrame(frame), 1u);
+  EXPECT_EQ(frame[0], 0x5A);
+}
+
 int main(int argc, char** argv) {
   ::testing::InitGoogleTest(&argc, argv);
   return RUN_ALL_TESTS();