Preserve Wi-Fi reconnect backlogs by client IP

Retain same-IP replies and park one displaced IP backlog for 30 seconds using existing fixed buffers. Cancel old route-owned operations on takeover, preserve partial TCP frames, and cover reconnect isolation and queue limits with regression tests.

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

@@ -155,6 +155,8 @@ jobs:
       - name: Verify WiFi, ESP-NOW and combined wireless controls
         run: |
           python3 -B test/test_wireless_control.py -v
+          python3 -B test/test_serial_wifi_sessions.py -v
+          python3 -B test/test_companion_wifi_session.py -v
           python3 -B test/test_espnow_radio_lifecycle.py -v
 
       - name: Verify message reader buttons, touch targets, and footer layouts

docs/companion_radio_full.md

@@ -386,6 +386,23 @@ their views. Companion session state is device-wide, so use one active
 Companion application at a time. On nRF52, BLE remains available while USB is
 in terminal or mOTA mode.
 
+On WiFi TCP port 5000, reconnecting from the same IP retains queued outgoing
+frames. A different IP takes over immediately, but never receives the previous
+IP's replies: up to four queued frames are parked separately for 30 seconds
+from takeover. The old IP can reconnect within that window to retrieve them.
+Only one displaced IP's backlog is retained; if another client also needs that
+slot, the older backlog is discarded so the new connection is not delayed.
+This reuses an existing fixed buffer instead of allocating additional frame
+storage. Pending commands, signing sessions, and contact streams belonging to
+the displaced WiFi session are canceled immediately; already accepted radio
+message transmissions are not canceled. Queued frames are retained, not a
+resumable command session, and IP matching is not authentication. Turning WiFi
+off clears both queues. Stored message history is unaffected.
+
+Partial TCP writes retain the unsent bytes without interleaving later frames.
+After a reconnect, an incomplete frame restarts from its header; the new TCP
+connection must not receive only the old connection's trailing bytes.
+
 USB Binary output is queued as complete length-prefixed frames. Temporary CDC
 or UART backpressure pauses the contact stream; a frame may drain through a
 smaller hardware FIFO in ordered chunks, but its remainder is retained and no

examples/companion_radio/main.cpp

@@ -1896,6 +1896,14 @@ void halt() {
     WIFI_DEBUG_PRINTLN("WiFi enabled by BOOT/GPIO 0 control");
   }
 
+  static void cancelCompanionWiFiSession(void*) {
+    if (interface_manager.isReplyRouteFor(&wifi_interface)) {
+      the_mesh.cancelSerialResponseStream();
+    }
+    the_mesh.cancelSerialOperationsForRoute(&wifi_interface);
+    interface_manager.forgetReplyRouteForDisconnected(&wifi_interface);
+  }
+
   static void stopCompanionWiFiServices() {
     if (companion_wifi_services_stopped) return;
     wifi_interface.end();
@@ -2778,6 +2786,7 @@ void setup() {
 
 // add wifi interface
 #ifdef WIFI_SSID
+  wifi_interface.setSessionChangedCallback(cancelCompanionWiFiSession, nullptr);
 #if defined(COMPANION_EXCLUSIVE_WIFI_BLE)
   if (companionTransportWiFiActiveAtBoot()) {
 #endif

src/helpers/MultiSerialInterface.h

@@ -344,8 +344,12 @@ public:
     // Other interfaces retain their input until the producer unlocks the route.
     if (_lockedReplyInterface != nullptr) {
       if (!isAvailableReplyTarget(_lockedReplyInterface)) return 0;
-      size_t frameSize = _lockedReplyInterface->checkRecvFrame(dest);
-      if (frameSize > 0) _lastRxInterface = _lockedReplyInterface;
+      // A backend may cancel the previous host's stream inside this call
+      // (e.g. WiFi IP takeover), clearing the lock before returning a new
+      // host's first command. Preserve the actual source of that command.
+      BaseSerialInterface* target = _lockedReplyInterface;
+      size_t frameSize = target->checkRecvFrame(dest);
+      if (frameSize > 0) _lastRxInterface = target;
       return frameSize;
     }
 

src/helpers/esp32/SerialWifiInterface.cpp

@@ -2,6 +2,62 @@
 #include "../CompanionFrameQueue.h"
 #include <WiFi.h>
 
+void SerialWifiInterface::clearBuffers() {
+  held_queue_len = send_queue_len = 0;
+  send_offset = 0;
+  queue_has_ip = false;
+  held_since = 0;
+  memset(held_queue, 0, sizeof(held_queue));
+  memset(send_queue, 0, sizeof(send_queue));
+}
+
+void SerialWifiInterface::expireHeldQueue(uint32_t now) {
+  if (held_queue_len > 0 && (uint32_t)(now - held_since) >= 30000U) {
+    held_queue_len = 0;
+    memset(held_queue, 0, sizeof(held_queue));
+  }
+}
+
+void SerialWifiInterface::loop() {
+  expireHeldQueue((uint32_t)millis());
+}
+
+void SerialWifiInterface::selectClient(const IPAddress& ip, uint32_t now) {
+  expireHeldQueue(now);
+  // TCP framing always starts afresh on a new connection, even for the same
+  // IP. An incomplete outgoing frame is retained whole, not just its suffix.
+  send_offset = 0;
+  if (queue_has_ip && queue_ip != ip) {
+    if (session_changed != nullptr) session_changed(session_context);
+
+    if (held_queue_len > 0 && held_ip == ip) {
+      // The previous owner returned: restore its queue and park the displaced
+      // active queue in the same fixed storage. Only one old IP is retained.
+      for (int i = 0; i < FRAME_QUEUE_SIZE; ++i) {
+        Frame swap = send_queue[i];
+        send_queue[i] = held_queue[i];
+        held_queue[i] = swap;
+      }
+      const int restored_len = held_queue_len;
+      held_queue_len = send_queue_len;
+      send_queue_len = restored_len;
+      held_ip = queue_ip;
+      held_since = now;
+    } else if (send_queue_len > 0) {
+      // Bounded fallback: a third owner's backlog replaces the older parked
+      // queue instead of allocating RAM or delaying the new connection.
+      memcpy(held_queue, send_queue, sizeof(held_queue));
+      held_queue_len = send_queue_len;
+      held_ip = queue_ip;
+      held_since = now;
+      send_queue_len = 0;
+      memset(send_queue, 0, sizeof(send_queue));
+    }
+  }
+  queue_ip = ip;
+  queue_has_ip = true;
+}
+
 void SerialWifiInterface::begin(int port) {
   // wifi setup is handled outside of this class, only starts the server
   server.begin(port);
@@ -22,10 +78,16 @@ void SerialWifiInterface::enable() {
 
   _isEnabled = true;
   clearBuffers();
+  resetReceivedFrameHeader();
 }
 
 void SerialWifiInterface::disable() {
   _isEnabled = false;
+  deviceConnected = false;
+  if (client) client.stop();
+  if (queue_has_ip && session_changed != nullptr) session_changed(session_context);
+  resetReceivedFrameHeader();
+  clearBuffers();
 }
 
 size_t SerialWifiInterface::writeFrame(const uint8_t src[], size_t len) {
@@ -34,12 +96,17 @@ size_t SerialWifiInterface::writeFrame(const uint8_t src[], size_t len) {
     return 0;
   }
 
-  if (deviceConnected && len > 0) {
-    if (!mesh::enqueueCompanionFrame(send_queue, send_queue_len, FRAME_QUEUE_SIZE,
-                                     src, len)) {
+  if (_isEnabled && deviceConnected && src != nullptr && len > 0) {
+    // Once any bytes of the front frame have reached TCP, priority insertion
+    // must not move or evict it. New frames may reorder only the unsent tail.
+    const int pinned = send_offset != 0 ? 1 : 0;
+    int tail_len = send_queue_len - pinned;
+    if (!mesh::enqueueCompanionFrame(send_queue + pinned, tail_len,
+                                     FRAME_QUEUE_SIZE - pinned, src, len)) {
       WIFI_DEBUG_PRINTLN("writeFrame(), send_queue is full!");
       return 0;
     }
+    send_queue_len = tail_len + pinned;
     return len;
   }
   return 0;
@@ -50,7 +117,7 @@ bool SerialWifiInterface::isReadBusy() const {
 }
 
 bool SerialWifiInterface::isWriteBusy() const {
-  return false;
+  return send_queue_len >= FRAME_QUEUE_SIZE;
 }
 
 bool SerialWifiInterface::hasReceivedFrameHeader() {
@@ -63,14 +130,21 @@ void SerialWifiInterface::resetReceivedFrameHeader() {
 }
 
 size_t SerialWifiInterface::checkRecvFrame(uint8_t dest[]) {
+  expireHeldQueue((uint32_t)millis());
+  if (!_isEnabled) return 0;
   // check if new client connected
   auto newClient = server.available();
   if (newClient) {
+    const IPAddress new_ip = newClient.remoteIP();
 
     // disconnect existing client
     deviceConnected = false;
     client.stop();
 
+    // Partition queues and cancel the displaced session before exposing the
+    // new connection to response producers or consuming its first command.
+    selectClient(new_ip, (uint32_t)millis());
+
     // switch active connection to new client
     client = newClient;
 
@@ -97,16 +171,22 @@ size_t SerialWifiInterface::checkRecvFrame(uint8_t dest[]) {
       _last_write = millis();
       int len = send_queue[0].len;
 
-      uint8_t pkt[3+len]; // use same header as serial interface so client can delimit frames
+      uint8_t pkt[3 + MAX_FRAME_SIZE]; // serial-compatible framing
       pkt[0] = '>';
       pkt[1] = (len & 0xFF);  // LSB
       pkt[2] = (len >> 8);    // MSB
       memcpy(&pkt[3], send_queue[0].buf, send_queue[0].len);
-      client.write(pkt, 3 + len);
+      const size_t total = 3U + len;
+      const size_t remaining = total - send_offset;
+      const size_t written = client.write(pkt + send_offset, remaining);
+      send_offset += written < remaining ? written : remaining;
+      if (send_offset < total) return 0;
+      send_offset = 0;
       send_queue_len--;
       for (int i = 0; i < send_queue_len; i++) {   // delete top item from queue
         send_queue[i] = send_queue[i + 1];
       }
+      memset(&send_queue[send_queue_len], 0, sizeof(send_queue[send_queue_len]));
     } else {
 
       // check if we are waiting for a frame header

src/helpers/esp32/SerialWifiInterface.h

@@ -8,7 +8,6 @@ class SerialWifiInterface : public BaseSerialInterface {
   bool deviceConnected;
   bool _isEnabled;
   unsigned long _last_write;
-  unsigned long adv_restart_time;
 
   WiFiServer server;
   WiFiClient client;
@@ -26,12 +25,23 @@ class SerialWifiInterface : public BaseSerialInterface {
   FrameHeader received_frame_header;
 
   #define FRAME_QUEUE_SIZE  4
-  int recv_queue_len;
-  Frame recv_queue[FRAME_QUEUE_SIZE];
+  // Reuse the former (unused) receive queue for one displaced client's replies.
+  // No heap allocation or additional frame buffer is needed on reconnect.
+  int held_queue_len;
+  Frame held_queue[FRAME_QUEUE_SIZE];
   int send_queue_len;
   Frame send_queue[FRAME_QUEUE_SIZE];
+  IPAddress queue_ip;
+  IPAddress held_ip;
+  bool queue_has_ip = false;
+  uint32_t held_since = 0;
+  size_t send_offset = 0;
+  void (*session_changed)(void*) = nullptr;
+  void* session_context = nullptr;
 
-  void clearBuffers() { recv_queue_len = 0; send_queue_len = 0; }
+  void clearBuffers();
+  void expireHeldQueue(uint32_t now);
+  void selectClient(const IPAddress& ip, uint32_t now);
 
 protected:
 
@@ -40,23 +50,31 @@ public:
     deviceConnected = false;
     _isEnabled = false;
     _last_write = 0;
-    send_queue_len = recv_queue_len = 0;
+    clearBuffers();
     received_frame_header.type = 0;
     received_frame_header.length = 0;
   }
 
   void begin(int port);
   void end();
+  // Called synchronously while disconnected, before a different IP can send
+  // commands or receive replies. The owner can cancel route-bound operations.
+  void setSessionChangedCallback(void (*callback)(void*), void* context) {
+    session_changed = callback;
+    session_context = context;
+  }
 
   // BaseSerialInterface methods
   void enable() override;
   void disable() override;
   bool isEnabled() const override { return _isEnabled; }
+  void loop() override;
 
   bool isConnected() const override;
   bool isReadBusy() const override;
   bool isWriteBusy() const override;
-  bool hasPendingIO() const override { return recv_queue_len > 0 || send_queue_len > 0; }
+  // A parked backlog is not runnable work and must not keep the MCU awake.
+  bool hasPendingIO() const override { return _isEnabled && deviceConnected && send_queue_len > 0; }
 
   size_t writeFrame(const uint8_t src[], size_t len) override;
   size_t checkRecvFrame(uint8_t dest[]) override;

test/fixtures/serial_wifi_sessions/mocks/Arduino.h

@@ -0,0 +1,8 @@
+#pragma once
+
+#include <cstddef>
+#include <cstdint>
+#include <cstring>
+
+extern uint32_t mock_millis;
+inline unsigned long millis() { return mock_millis; }

test/fixtures/serial_wifi_sessions/mocks/WiFi.h

@@ -0,0 +1,72 @@
+#pragma once
+
+#include <algorithm>
+#include <cstddef>
+#include <cstdint>
+#include <deque>
+#include <limits>
+#include <memory>
+#include <vector>
+
+class IPAddress {
+  uint32_t _address = 0;
+public:
+  IPAddress() = default;
+  IPAddress(uint32_t address) : _address(address) {}
+  IPAddress(uint8_t a, uint8_t b, uint8_t c, uint8_t d)
+      : _address(uint32_t(a) << 24 | uint32_t(b) << 16 | uint32_t(c) << 8 | d) {}
+  operator uint32_t() const { return _address; }
+  bool operator==(const IPAddress& other) const { return _address == other._address; }
+  bool operator!=(const IPAddress& other) const { return !(*this == other); }
+};
+
+struct MockSocket {
+  IPAddress address;
+  bool connected = true;
+  size_t write_limit = 0;
+  size_t read_count = 0;
+  std::vector<uint8_t> sent;
+  std::deque<uint8_t> received;
+  explicit MockSocket(IPAddress value) : address(value) {}
+};
+
+class WiFiClient {
+  std::shared_ptr<MockSocket> _socket;
+public:
+  WiFiClient() = default;
+  explicit WiFiClient(std::shared_ptr<MockSocket> socket) : _socket(socket) {}
+  explicit operator bool() const { return _socket && _socket->connected; }
+  bool connected() const { return _socket && _socket->connected; }
+  void stop() { if (_socket) _socket->connected = false; }
+  IPAddress remoteIP() const { return _socket ? _socket->address : IPAddress(); }
+  int available() const { return connected() ? int(_socket->received.size()) : 0; }
+  size_t read(uint8_t* dest, size_t size) {
+    size_t read = 0;
+    while (connected() && read < size && !_socket->received.empty()) {
+      dest[read++] = _socket->received.front();
+      _socket->received.pop_front();
+      ++_socket->read_count;
+    }
+    return read;
+  }
+  size_t readBytes(uint8_t* dest, size_t size) { return read(dest, size); }
+  size_t write(const uint8_t* src, size_t size) {
+    if (!connected()) return 0;
+    const size_t written = std::min(size, _socket->write_limit);
+    _socket->sent.insert(_socket->sent.end(), src, src + written);
+    return written;
+  }
+};
+
+class WiFiServer {
+public:
+  static std::deque<WiFiClient> incoming;
+  void begin(int) {}
+  void end() { incoming.clear(); }
+  WiFiClient available() {
+    if (incoming.empty()) return WiFiClient();
+    WiFiClient client = incoming.front();
+    incoming.pop_front();
+    return client;
+  }
+};

test/fixtures/serial_wifi_sessions/test.cpp

@@ -0,0 +1,344 @@
+#include <helpers/esp32/SerialWifiInterface.h>
+#include <cassert>
+#include <cstdio>
+#include <cstdlib>
+#include <initializer_list>
+#include <limits>
+#include <vector>
+
+uint32_t mock_millis = 0;
+std::deque<WiFiClient> WiFiServer::incoming;
+
+using Bytes = std::vector<uint8_t>;
+using Socket = std::shared_ptr<MockSocket>;
+
+static const IPAddress A(192, 168, 1, 20);
+static const IPAddress B(192, 168, 1, 30);
+static const IPAddress C(192, 168, 1, 40);
+
+static Bytes framed(std::initializer_list<Bytes> frames) {
+  Bytes result;
+  for (const auto& frame : frames) {
+    result.push_back('>');
+    result.push_back(uint8_t(frame.size()));
+    result.push_back(uint8_t(frame.size() >> 8));
+    result.insert(result.end(), frame.begin(), frame.end());
+  }
+  return result;
+}
+
+struct Fixture {
+  SerialWifiInterface transport;
+  uint8_t input[MAX_FRAME_SIZE] = {};
+  int callbacks = 0;
+  bool pending_operation = false;
+  Socket next_client;
+
+  Fixture() {
+    mock_millis = 100;
+    WiFiServer::incoming.clear();
+    transport.begin(5000);
+    transport.enable();
+    transport.setSessionChangedCallback([](void* context) {
+      Fixture& fixture = *static_cast<Fixture*>(context);
+      ++fixture.callbacks;
+      fixture.pending_operation = false;
+      // Cancellation must happen before a new socket sees an old response or
+      // has any of its own command bytes consumed.
+      if (fixture.next_client) {
+        assert(fixture.next_client->sent.empty());
+        assert(fixture.next_client->read_count == 0);
+      }
+    }, this);
+  }
+
+  ~Fixture() { transport.end(); }
+
+  size_t tick() { return transport.checkRecvFrame(input); }
+
+  Socket connect(IPAddress address, const Bytes& command = {}) {
+    Socket socket = std::make_shared<MockSocket>(address);
+    socket->received.insert(socket->received.end(), command.begin(), command.end());
+    next_client = socket;
+    WiFiServer::incoming.push_back(WiFiClient(socket));
+    tick();
+    next_client.reset();
+    assert(transport.isConnected());
+    return socket;
+  }
+
+  void enqueue(const Bytes& frame) {
+    assert(transport.writeFrame(frame.data(), frame.size()) == frame.size());
+  }
+
+  void drain(const Socket& socket) {
+    socket->write_limit = std::numeric_limits<size_t>::max();
+    for (int i = 0; i < 12; ++i) tick();
+  }
+};
+
+int main(int argc, char** argv) {
+  assert(argc == 2);
+  const int scenario = std::atoi(argv[1]);
+  Fixture f;
+  const Bytes reply_a{0x01, 0x21, 0x22, 0x23};
+  const Bytes reply_b{0x02, 0x31, 0x32};
+  const Bytes reply_c{0x03, 0x41};
+
+  if (scenario == 0) {
+    auto old = f.connect(A);
+    f.enqueue(reply_a);
+    auto replacement = f.connect(A);
+    assert(!old->connected && f.callbacks == 0);
+    f.drain(replacement);
+    assert(replacement->sent == framed({reply_a}));
+  } else if (scenario == 1) {
+    auto old = f.connect(A);
+    f.enqueue(reply_a);
+    old->connected = false;
+    f.tick();
+    assert(!f.transport.isConnected());
+    mock_millis += 60000; // Same-IP retention is not the different-IP grace timer.
+    auto replacement = f.connect(A);
+    f.drain(replacement);
+    assert(replacement->sent == framed({reply_a}) && f.callbacks == 0);
+  } else if (scenario == 2) {
+    f.connect(A);
+    f.enqueue(reply_a);
+    auto other = f.connect(B);
+    f.enqueue(reply_b);
+    f.drain(other);
+    assert(other->sent == framed({reply_b}) && f.callbacks == 1);
+  } else if (scenario == 3) {
+    f.connect(A);
+    f.enqueue(reply_a);
+    auto other = f.connect(B);
+    f.enqueue(reply_b);
+    mock_millis += 29999;
+    auto returned = f.connect(A);
+    f.drain(returned);
+    assert(returned->sent == framed({reply_a}));
+    assert(other->sent.empty());
+    auto other_returned = f.connect(B);
+    f.drain(other_returned);
+    assert(other_returned->sent == framed({reply_b}));
+    assert(f.callbacks == 3);
+  } else if (scenario == 4) {
+    f.connect(A);
+    f.enqueue(reply_a);
+    f.connect(B);
+    mock_millis += 30000;
+    auto expired = f.connect(A);
+    f.drain(expired);
+    assert(expired->sent.empty());
+  } else if (scenario == 5) {
+    f.connect(A);
+    f.enqueue(reply_a);
+    f.connect(B);
+    mock_millis += 20000;
+    f.connect(B); // This must not extend A's held deadline.
+    assert(f.callbacks == 1);
+    mock_millis += 10000;
+    auto expired = f.connect(A);
+    f.drain(expired);
+    assert(expired->sent.empty());
+  } else if (scenario == 6 || scenario == 7) {
+    mock_millis = UINT32_MAX - 10000;
+    f.connect(A);
+    f.enqueue(reply_a);
+    f.connect(B);
+    mock_millis += scenario == 6 ? 29999 : 30000;
+    auto returned = f.connect(A);
+    f.drain(returned);
+    assert(returned->sent == (scenario == 6 ? framed({reply_a}) : Bytes{}));
+  } else if (scenario == 8) {
+    f.connect(A);
+    f.enqueue(reply_a);
+    f.connect(B);
+    f.enqueue(reply_b);
+    f.connect(C); // Park B and evict A: only one spare fixed-size queue exists.
+    f.enqueue(reply_c);
+    auto evicted = f.connect(A);
+    f.drain(evicted);
+    assert(evicted->sent.empty());
+    auto newest = f.connect(C);
+    f.drain(newest);
+    assert(newest->sent == framed({reply_c}));
+  } else if (scenario == 9) {
+    auto socket = f.connect(A);
+    f.enqueue(reply_a);
+    socket->write_limit = 2;
+    for (int i = 0; i < 8; ++i) f.tick();
+    assert(socket->sent == framed({reply_a}));
+    assert(!f.transport.hasPendingIO());
+  } else if (scenario == 10) {
+    auto socket = f.connect(A);
+    const Bytes push{0x80, 0xA1, 0xA2, 0xA3};
+    f.enqueue(push);
+    socket->write_limit = 2;
+    f.tick();
+    assert(socket->sent.size() == 2);
+    f.enqueue(reply_a); // Responses have higher queue priority than this push.
+    f.drain(socket);
+    assert(socket->sent == framed({push, reply_a}));
+  } else if (scenario == 11) {
+    auto old = f.connect(A);
+    f.enqueue(reply_a);
+    old->write_limit = 5; // Header and part of payload already sent.
+    f.tick();
+    assert(old->sent.size() == 5);
+    auto replacement = f.connect(A);
+    f.drain(replacement);
+    assert(replacement->sent == framed({reply_a}));
+  } else if (scenario == 12) {
+    auto old = f.connect(A);
+    f.enqueue(reply_a);
+    old->write_limit = 2; // Partial header must not be a suffix on reconnection.
+    f.tick();
+    auto other = f.connect(B);
+    f.drain(other);
+    assert(other->sent.empty());
+    auto returned = f.connect(A);
+    f.drain(returned);
+    assert(returned->sent == framed({reply_a}));
+  } else if (scenario == 13 || scenario == 14) {
+    f.connect(A);
+    f.enqueue(reply_a);
+    f.connect(B);
+    f.enqueue(reply_b);
+    if (scenario == 13) f.transport.disable();
+    else f.transport.end();
+    assert(!f.transport.isConnected() && !f.transport.hasPendingIO());
+    f.transport.begin(5000);
+    f.transport.enable();
+    auto previous = f.connect(A);
+    f.drain(previous);
+    assert(previous->sent.empty());
+    auto other = f.connect(B);
+    f.drain(other);
+    assert(other->sent.empty());
+  } else if (scenario == 15) {
+    auto old = f.connect(A, Bytes{'<', 4, 0, 0x11}); // Incomplete old command.
+    assert(old->read_count == 3);
+    f.pending_operation = true;
+    f.enqueue(reply_a);
+    auto other = f.connect(B, Bytes{'<', 2, 0, 0x71, 0x72});
+    assert(f.callbacks == 1 && !f.pending_operation);
+    // The new header is parsed from its start, not attached to the old header.
+    if (other->read_count == 0) assert(f.tick() == 2);
+    assert(other->read_count == 5);
+    assert(f.input[0] == 0x71 && f.input[1] == 0x72);
+    assert(other->sent.empty());
+  } else if (scenario == 16) {
+    auto old = f.connect(A, Bytes{'<', 4, 0, 0x11});
+    assert(old->read_count == 3);
+    f.pending_operation = true;
+    auto replacement = f.connect(A, Bytes{'<', 2, 0, 0x61, 0x62});
+    assert(f.callbacks == 0 && f.pending_operation);
+    assert(replacement->read_count == 5);
+    assert(f.input[0] == 0x61 && f.input[1] == 0x62);
+  } else if (scenario == 17) {
+    auto socket = f.connect(A);
+    f.enqueue(reply_a);
+    f.tick(); // A zero-byte socket write is backpressure, not frame completion.
+    assert(socket->sent.empty() && f.transport.hasPendingIO());
+    f.drain(socket);
+    assert(socket->sent == framed({reply_a}));
+  } else if (scenario == 18) {
+    f.connect(A);
+    for (int i = 0; i < 4; ++i) f.enqueue(Bytes{0x01, uint8_t(i)});
+    assert(f.transport.writeFrame(reply_a.data(), reply_a.size()) == 0);
+    f.connect(B);
+    for (int i = 0; i < 4; ++i) f.enqueue(Bytes{0x02, uint8_t(i)});
+    auto returned = f.connect(A);
+    f.drain(returned);
+    assert(returned->sent == framed({{0x01, 0}, {0x01, 1}, {0x01, 2}, {0x01, 3}}));
+    auto other = f.connect(B);
+    f.drain(other);
+    assert(other->sent == framed({{0x02, 0}, {0x02, 1}, {0x02, 2}, {0x02, 3}}));
+  } else if (scenario == 19) {
+    f.connect(A);
+    f.enqueue(reply_a);
+    f.connect(B);
+    mock_millis += 20000;
+    f.connect(C); // B has no backlog: preserve A without extending its deadline.
+    mock_millis += 9999;
+    auto returned = f.connect(A);
+    f.drain(returned);
+    assert(returned->sent == framed({reply_a}));
+  } else if (scenario == 20) {
+    f.connect(A);
+    f.enqueue(reply_a);
+    f.connect(B);
+    mock_millis += 20000;
+    f.connect(C);
+    assert(!f.transport.hasPendingIO()); // Parked-only data must not busy-loop.
+    mock_millis += 10000;
+    f.transport.loop();
+    auto expired = f.connect(A);
+    f.drain(expired);
+    assert(expired->sent.empty());
+  } else if (scenario == 21) {
+    auto socket = f.connect(A);
+    const Bytes push{0x80, 0xA1, 0xA2, 0xA3};
+    const Bytes keep{0x81, 0xB1};
+    const Bytes evict{0x88, 0xC1};
+    f.enqueue(push);
+    socket->write_limit = 2;
+    f.tick();
+    f.enqueue(keep);
+    f.enqueue(evict);
+    f.enqueue(reply_a);
+    f.enqueue(reply_b); // Full queue can evict a waiting push, never its sent head.
+    f.drain(socket);
+    assert(socket->sent == framed({push, reply_a, reply_b, keep}));
+  } else if (scenario == 22) {
+    f.connect(A, Bytes{'<', 4, 0, 0x11});
+    f.pending_operation = true;
+    f.transport.disable();
+    assert(f.callbacks == 1 && !f.pending_operation);
+    f.transport.end(); // Repeated teardown must not cancel the same owner twice.
+    assert(f.callbacks == 1);
+    f.transport.begin(5000);
+    f.transport.enable();
+    auto replacement = f.connect(A, Bytes{'<', 2, 0, 0x61, 0x62});
+    assert(replacement->read_count == 5);
+    assert(f.input[0] == 0x61 && f.input[1] == 0x62);
+  } else if (scenario == 23) {
+    auto socket = f.connect(A);
+    Bytes maximum(MAX_FRAME_SIZE);
+    for (size_t i = 0; i < maximum.size(); ++i) maximum[i] = uint8_t(i);
+    maximum[0] = 0x80; // Low-priority push, pinned once any bytes are sent.
+    f.enqueue(maximum);
+    socket->write_limit = MAX_FRAME_SIZE + 2; // All but the final payload byte.
+    f.tick();
+    assert(socket->sent.size() == MAX_FRAME_SIZE + 2);
+    assert(f.transport.hasPendingIO());
+    f.enqueue(reply_a);
+    f.drain(socket);
+    assert(socket->sent == framed({maximum, reply_a}));
+    assert(!f.transport.hasPendingIO());
+  } else if (scenario == 24) {
+    auto old = f.connect(A);
+    Bytes maximum(MAX_FRAME_SIZE);
+    for (size_t i = 0; i < maximum.size(); ++i) maximum[i] = uint8_t(i ^ 0x55);
+    maximum[0] = 0x01;
+    f.enqueue(maximum);
+    old->write_limit = MAX_FRAME_SIZE + 2;
+    f.tick();
+    assert(old->sent.size() == MAX_FRAME_SIZE + 2);
+    f.connect(B);
+    Bytes other_maximum(MAX_FRAME_SIZE, 0xFE);
+    other_maximum[0] = 0x02;
+    f.enqueue(other_maximum);
+    auto returned = f.connect(A);
+    f.drain(returned);
+    assert(returned->sent == framed({maximum}));
+    auto other_returned = f.connect(B);
+    f.drain(other_returned);
+    assert(other_returned->sent == framed({other_maximum}));
+  } else {
+    assert(false && "unknown scenario");
+  }
+  std::printf("PASS: Wi-Fi session scenario %d\n", scenario);
+}

test/test_companion_wifi_session.py

@@ -0,0 +1,154 @@
+"""Exercise WiFi session cancellation against production route ownership code."""
+from pathlib import Path
+import os
+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"
+MESH = ROOT / "examples/companion_radio/MyMesh.cpp"
+
+
+HARNESS = r'''
+#include <cassert>
+#include <cstdint>
+#include <cstring>
+#include <deque>
+#include <vector>
+#include <helpers/MultiSerialInterface.h>
+
+struct FakeInterface : BaseSerialInterface {
+  bool enabled = false;
+  bool connected = true;
+  std::deque<std::vector<uint8_t>> input;
+  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; }
+  size_t writeFrame(const uint8_t[], size_t len) override { return len; }
+  size_t checkRecvFrame(uint8_t dest[]) override {
+    if (input.empty()) return 0;
+    auto frame = input.front(); input.pop_front();
+    memcpy(dest, frame.data(), frame.size());
+    return frame.size();
+  }
+} wifi_interface, usb_interface, bluetooth_interface;
+MultiSerialInterface interface_manager;
+
+constexpr int EXPECTED_ACK_TABLE_SIZE = 6;
+struct MyMesh {
+  bool streaming = false;
+  int stream_cancels = 0, pending_cancels = 0, radio_cancels = 0;
+  int trace_cancels = 0, signing_cancels = 0, expirations = 0;
+  BaseSerialInterface* pending_serial_reply_route = nullptr;
+  BaseSerialInterface* command_radio_reply_route = nullptr;
+  BaseSerialInterface* binary_trace_reply_route = nullptr;
+  BaseSerialInterface* sign_data_reply_route = nullptr;
+  struct Ack { BaseSerialInterface* reply_route; bool radio_retry; };
+  Ack expected_ack_table[EXPECTED_ACK_TABLE_SIZE] = {};
+  void cancelSerialResponseStream() {
+    ++stream_cancels;
+    if (streaming) { streaming = false; interface_manager.unlockReplyRoute(); }
+  }
+  void clearPendingReqs() { ++pending_cancels; pending_serial_reply_route = nullptr; }
+  void cancelPendingRadioParamApply() { ++radio_cancels; command_radio_reply_route = nullptr; }
+  void clearBinaryTraceReply() { ++trace_cancels; binary_trace_reply_route = nullptr; }
+  void cancelSigningSession() { ++signing_cancels; sign_data_reply_route = nullptr; }
+  void expireExpectedAcks() { ++expirations; }
+  void cancelSerialOperationsForRoute(BaseSerialInterface*);
+} the_mesh;
+
+@CANCEL_OPERATIONS@
+@CANCEL_SESSION@
+
+int main() {
+  assert(interface_manager.addInterface(InterfaceType::WiFi, &wifi_interface));
+  assert(interface_manager.addInterface(InterfaceType::USB, &usb_interface));
+  assert(interface_manager.addInterface(InterfaceType::Bluetooth, &bluetooth_interface));
+  FakeInterface* routes[] = {&wifi_interface, &usb_interface, &bluetooth_interface};
+  for (auto owner : routes) {
+    interface_manager.enable();
+    the_mesh = MyMesh();
+    owner->input.push_back({0x04});
+    uint8_t command[MAX_FRAME_SIZE] = {};
+    assert(interface_manager.checkRecvFrame(command) == 1);
+    interface_manager.lockReplyRoute();
+    the_mesh.streaming = true;
+    the_mesh.pending_serial_reply_route = owner;
+    the_mesh.command_radio_reply_route = owner;
+    the_mesh.binary_trace_reply_route = owner;
+    the_mesh.sign_data_reply_route = owner;
+    for (unsigned i = 0; i < EXPECTED_ACK_TABLE_SIZE; ++i) {
+      the_mesh.expected_ack_table[i] = {routes[i % 3], true};
+    }
+
+    // The transport marks itself disconnected during the synchronous callback.
+    wifi_interface.connected = false;
+    cancelCompanionWiFiSession(nullptr);
+    wifi_interface.connected = true;
+    const bool cancelled = owner == &wifi_interface;
+    assert(the_mesh.stream_cancels == int(cancelled));
+    assert(the_mesh.streaming == !cancelled);
+    assert(the_mesh.pending_cancels == int(cancelled));
+    assert(the_mesh.radio_cancels == int(cancelled));
+    assert(the_mesh.trace_cancels == int(cancelled));
+    assert(the_mesh.signing_cancels == int(cancelled));
+    assert(the_mesh.pending_serial_reply_route == (cancelled ? nullptr : owner));
+    assert(the_mesh.command_radio_reply_route == (cancelled ? nullptr : owner));
+    assert(the_mesh.binary_trace_reply_route == (cancelled ? nullptr : owner));
+    assert(the_mesh.sign_data_reply_route == (cancelled ? nullptr : owner));
+    assert(!interface_manager.isReplyRouteFor(&wifi_interface));
+    if (!cancelled) assert(interface_manager.isReplyRouteFor(owner));
+    for (unsigned i = 0; i < EXPECTED_ACK_TABLE_SIZE; ++i) {
+      // Detach WiFi's notification, not the radio's accepted message/retry.
+      assert(the_mesh.expected_ack_table[i].reply_route ==
+             (i % 3 == 0 ? nullptr : routes[i % 3]));
+      assert(the_mesh.expected_ack_table[i].radio_retry);
+    }
+    assert(the_mesh.expirations == 1);
+  }
+}
+'''
+
+
+class CompanionWiFiSessionTest(unittest.TestCase):
+    def test_callback_is_registered_before_wifi_interface_can_start(self):
+        source = MAIN.read_text(encoding="utf-8")
+        registration = source.index(
+            "wifi_interface.setSessionChangedCallback(cancelCompanionWiFiSession, nullptr)"
+        )
+        setup = source.index("void setup()")
+        add = source.index("interface_manager.addInterface(InterfaceType::WiFi", setup)
+        start = source.index("startCompanionWiFi();", add)
+        self.assertLess(registration, add)
+        self.assertLess(add, start)
+
+    def test_session_reset_cancels_only_wifi_owned_work(self):
+        main = MAIN.read_text(encoding="utf-8")
+        mesh = MESH.read_text(encoding="utf-8")
+        callback = extract_braced(main, "static void cancelCompanionWiFiSession(")
+        operations = extract_braced(mesh, "void MyMesh::cancelSerialOperationsForRoute(")
+        source = HARNESS.replace("@CANCEL_OPERATIONS@", operations).replace(
+            "@CANCEL_SESSION@", callback
+        )
+        with tempfile.TemporaryDirectory() as directory:
+            executable = Path(directory) / ("wifi_session.exe" if os.name == "nt" else "wifi_session")
+            result = subprocess.run(
+                [os.environ.get("CXX", "g++"), "-std=c++17", "-Werror",
+                 f"-I{ROOT / 'test/mocks'}", f"-I{ROOT / 'src'}",
+                 "-x", "c++", "-", "-o", str(executable)],
+                input=source, text=True, capture_output=True,
+            )
+            self.assertEqual(result.returncode, 0, result.stderr)
+            result = subprocess.run([str(executable)], text=True, capture_output=True)
+            self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+
+
+if __name__ == "__main__":
+    unittest.main()

test/test_serial_mode_switch/test_serial_mode_switch.cpp

@@ -448,6 +448,54 @@ TEST(MultiSerialInterface, ForgetsOnlyTheDisconnectedHostReplyRoute) {
   EXPECT_TRUE(manager.isReplyRouteFor(&bluetooth));
 }
 
+TEST(MultiSerialInterface, SessionChangeDuringLockedReadKeepsNewFrameOnItsTransport) {
+  MultiSerialInterface manager;
+  class ReconnectingInterface : public FakeSerialInterface {
+    MultiSerialInterface& manager;
+  public:
+    bool change_session = false;
+    explicit ReconnectingInterface(MultiSerialInterface& owner) : manager(owner) {}
+    size_t checkRecvFrame(uint8_t dest[]) override {
+      if (change_session) {
+        change_session = false;
+        // WiFi's new-IP callback cancels the old contact iterator and then
+        // forgets its route while the manager is still inside this read.
+        manager.unlockReplyRoute();
+        manager.forgetReplyRouteForDisconnected(this);
+      }
+      return FakeSerialInterface::checkRecvFrame(dest);
+    }
+  } wifi(manager);
+  FakeSerialInterface usb, bluetooth;
+  wifi.connected = usb.connected = bluetooth.connected = true;
+  ASSERT_TRUE(manager.addInterface(InterfaceType::WiFi, &wifi));
+  ASSERT_TRUE(manager.addInterface(InterfaceType::USB, &usb));
+  ASSERT_TRUE(manager.addInterface(InterfaceType::Bluetooth, &bluetooth));
+  manager.enable();
+
+  wifi.received_frames.push_back({0x04});
+  uint8_t command[MAX_FRAME_SIZE] = {};
+  ASSERT_EQ(manager.checkRecvFrame(command), 1u);
+  manager.lockReplyRoute();
+
+  wifi.change_session = true;
+  wifi.received_frames.push_back({0x16});
+  ASSERT_EQ(manager.checkRecvFrame(command), 1u);
+  ASSERT_EQ(command[0], 0x16);
+  ASSERT_TRUE(manager.isReplyRouteFor(&wifi));
+  const uint8_t device_info[] = {0x0D, 0x43};
+  ASSERT_EQ(manager.writeFrame(device_info, sizeof(device_info)), sizeof(device_info));
+  ASSERT_EQ(wifi.sent_frames.size(), 1u);
+  EXPECT_TRUE(usb.sent_frames.empty());
+  EXPECT_TRUE(bluetooth.sent_frames.empty());
+
+  // Cancelling the prior WiFi iterator also releases its lock, so another
+  // transport may dispatch its own command on the next pass.
+  usb.received_frames.push_back({0x16});
+  ASSERT_EQ(manager.checkRecvFrame(command), 1u);
+  EXPECT_TRUE(manager.isReplyRouteFor(&usb));
+}
+
 TEST(MultiSerialInterface, CapturedAsyncReplySurvivesLaterRouteChanges) {
   MultiSerialInterface manager;
   FakeSerialInterface usb;

test/test_serial_wifi_sessions.py

@@ -0,0 +1,79 @@
+#!/usr/bin/env python3
+"""Exercise production Wi-Fi transport reconnect ownership and partial writes."""
+
+from pathlib import Path
+import os
+import shutil
+import subprocess
+import sys
+import tempfile
+import unittest
+
+
+ROOT = Path(__file__).resolve().parents[1]
+FIXTURE = ROOT / "test/fixtures/serial_wifi_sessions"
+# MinGW does not ship these runtimes. Non-PIE avoids ASan address-space
+# collisions in Linux hosts, matching the other native regression harnesses.
+SANITIZER_FLAGS = (
+    ["-fsanitize=address,undefined", "-fno-sanitize-recover=all", "-fno-pie", "-no-pie"]
+    if sys.platform.startswith("linux") else []
+)
+
+SCENARIOS = (
+    "same IP live replacement keeps replies",
+    "same IP reconnect after disconnect keeps replies",
+    "different IP receives only its own replies",
+    "old IP returns before grace deadline and swaps queues",
+    "old IP expires at exactly 30 seconds",
+    "same IP reconnect does not extend another IP grace deadline",
+    "old IP returns before deadline across millis rollover",
+    "old IP expires across millis rollover",
+    "third IP evicts older parked backlog",
+    "partial writes retain every byte",
+    "higher priority reply cannot interrupt partial frame",
+    "same IP reconnect replays whole partial frame",
+    "different IP return replays whole partial frame",
+    "disable clears both queues and active connection",
+    "end clears both queues and active connection",
+    "different IP callback precedes new input and resets partial RX",
+    "same IP preserves pending operations but resets partial RX",
+    "zero-byte write retains pending frame",
+    "both bounded queues retain four complete responses",
+    "empty different IP takeover preserves existing held backlog",
+    "empty takeover does not extend expiry and parked data does not busy-loop",
+    "full queue admission cannot evict a partly sent frame",
+    "disable cancels owner once and resets partial RX state",
+    "maximum-sized partly sent frame stays intact before higher priority response",
+    "maximum-sized partial frame and parked queue survive owner swap",
+)
+
+
+class SerialWifiSessionTests(unittest.TestCase):
+    def test_production_transport_sessions(self):
+        compiler = os.environ.get("CXX") or shutil.which("g++") or shutil.which("clang++")
+        if compiler is None:
+            self.skipTest("a host C++17 compiler is required")
+        with tempfile.TemporaryDirectory(prefix="meshcore-wifi-sessions-") as directory:
+            binary = Path(directory) / "wifi-sessions.exe"
+            compiled = subprocess.run(
+                [compiler, "-std=c++17", "-Wall", "-Wextra", "-Werror",
+                 *SANITIZER_FLAGS,
+                 f"-I{FIXTURE / 'mocks'}", f"-I{ROOT / 'src'}",
+                 str(FIXTURE / "test.cpp"),
+                 str(ROOT / "src/helpers/esp32/SerialWifiInterface.cpp"),
+                 "-o", str(binary)],
+                capture_output=True, text=True, timeout=60,
+            )
+            self.assertEqual(compiled.returncode, 0, compiled.stdout + compiled.stderr)
+            for scenario, description in enumerate(SCENARIOS):
+                with self.subTest(scenario=description):
+                    checked = subprocess.run(
+                        [str(binary), str(scenario)], capture_output=True,
+                        text=True, timeout=10,
+                    )
+                    self.assertEqual(checked.returncode, 0, checked.stdout + checked.stderr)
+                    self.assertIn(f"PASS: Wi-Fi session scenario {scenario}", checked.stdout)
+
+
+if __name__ == "__main__":
+    unittest.main()