Merge rp2-lwip-seg-pool-and-heap review fixes into the combined test branch

Segment pool entries are 20 bytes on this build, not the nominal 16 that
struct tcp_seg suggests; the comment now quotes the measured figure.

The MEM_SIZE justification did not close as written: two connections at
~6KB each is ~12KB, which fits inside the old 16KB heap. The missing
term is that mem.c is first-fit, so what has to be free is a contiguous
1.5KB block, and at 75% occupancy interleaved with ARP, DHCP/DNS and
mDNS allocations the largest run collapses long before the total does.
That is also why the failure looked intermittent. Spelled out in both
the Python comment and the header template.

Hoist the lwIP sizing values to module constants so the two numeric
invariants can be tested: the segment pool must stay above the per-PCB
send queue, and MEM_SIZE must stay under 64000 or lwIP widens
mem_size_t to u32_t. Both were previously enforced by prose alone, and
the first would silently rebuild the starvation this change removes.
The generated header is byte identical.

esphome/components/api/api_frame_helper.h

@@ -149,7 +149,7 @@ class APIFrameHelper {
   // holding data too long waiting for Nagle's timer causes buffer exhaustion
   // and dropped messages.
   //
-  // ESP32 (TCP_SND_BUF=4×MSS+) / RP2040 (8×MSS) / LibreTiny (4×MSS): 4 logs per cycle
+  // ESP32 (TCP_SND_BUF=4×MSS+) / RP2040 (4×MSS) / LibreTiny (4×MSS): 4 logs per cycle
   // ESP8266 (2×MSS): 3 logs per cycle (tightest buffers)
   //
   // Flow (ESP32/RP2040/LT): Log 1 (Nagle on) -> Log 2 -> Log 3 -> Log 4 (NODELAY, flush)
@@ -312,7 +312,7 @@ class APIFrameHelper {
   // Values 1..LOG_NAGLE_COUNT count log messages in the current Nagle batch.
   // After LOG_NAGLE_COUNT logs, we flush by re-enabling NODELAY and resetting to 0.
   // ESP8266 has the tightest TCP send buffer (2×MSS) and needs conservative batching.
-  // ESP32 (4×MSS+), RP2040 (8×MSS), and LibreTiny (4×MSS) can coalesce more.
+  // ESP32 (4×MSS+), RP2040 (4×MSS), and LibreTiny (4×MSS) can coalesce more.
 #ifdef USE_ESP8266
   static constexpr uint8_t LOG_NAGLE_COUNT = 2;
 #else

esphome/components/bluetooth_connection/__init__.py

@@ -9,6 +9,7 @@ from collections.abc import Awaitable, Callable
 from dataclasses import dataclass
 
 import esphome.codegen as cg
+from esphome.components import rp2040_ble
 from esphome.config_helpers import (
     filter_source_files_from_platform,
     frameworks_for_platforms,
@@ -36,9 +37,12 @@ CODEOWNERS = ["@bdraco", "@jesserockz"]
 
 bluetooth_connection_ns = cg.esphome_ns.namespace("bluetooth_connection")
 
-# arduino-pico's prebuilt BTstack is compiled with MAX_NR_GATT_CLIENTS 1;
-# raising this needs an upstream change (the layer itself supports N).
-RP2_MAX_CONNECTIONS = 1
+# arduino-pico's prebuilt BTstack is compiled with MAX_NR_GATT_CLIENTS 1 and
+# MAX_NR_HCI_CONNECTIONS 2; for more than one backend, rp2040_ble's
+# btstack_memory.cpp replaces those pools via linker --wrap (requested by
+# _rp2_register), sized from ESPHOME_BLE_GATT_CLIENT_COUNT. The cap itself
+# belongs to the platform stack that owns the pools.
+RP2_MAX_CONNECTIONS = rp2040_ble.MAX_CONNECTIONS
 
 # Slot limits for the hub platforms running the connection-capable proxy;
 # the backend registry itself is _PLATFORM_BACKENDS below.
@@ -53,6 +57,19 @@ BluedroidGattClient = bluetooth_connection_ns.class_(
 
 CONF_BACKEND_ID = "backend_id"
 
+DOMAIN = "bluetooth_connection"
+
+
+@dataclass
+class _ConnectionData:
+    rp2_backend_count: int = 0
+
+
+def _get_data() -> _ConnectionData:
+    if DOMAIN not in CORE.data:
+        CORE.data[DOMAIN] = _ConnectionData()
+    return CORE.data[DOMAIN]
+
 
 def _esp32_schema_fragment() -> cv.Schema:
     from esphome.components import esp32_ble_tracker
@@ -61,8 +78,6 @@ def _esp32_schema_fragment() -> cv.Schema:
 
 
 def _rp2_schema_fragment() -> cv.Schema:
-    from esphome.components import rp2040_ble
-
     return cv.Schema(
         {cv.GenerateID(rp2040_ble.CONF_RP2040_BLE_ID): cv.use_id(rp2040_ble.RP2040BLE)}
     )
@@ -77,15 +92,29 @@ async def _esp32_register(backend: cg.MockObj, config: ConfigType) -> None:
 
 
 async def _rp2_register(backend: cg.MockObj, config: ConfigType) -> None:
-    from esphome.components import rp2040_ble
+    from esphome.components import ota
 
+    # The backend drops its link when an OTA starts (esp32 tracker parity).
+    ota.request_ota_state_listeners()
+    # More than one backend outgrows the prebuilt BTstack pools: swap them for
+    # the ESPHOME_BLE_GATT_CLIENT_COUNT-sized ones in rp2040_ble's
+    # btstack_memory.cpp. Keyed to backend registrations (the same event that
+    # grows the count that sizes the pools), so single-backend builds emit no
+    # flags and stay byte-identical to previous releases.
+    data = _get_data()
+    data.rp2_backend_count += 1
+    if data.rp2_backend_count == 2:
+        rp2040_ble.add_btstack_pool_overrides()
     await cg.register_parented(backend, config[rp2040_ble.CONF_RP2040_BLE_ID])
 
 
 @dataclass(frozen=True)
 class _PlatformBackend:
-    """One platform's backend: codegen class, extra schema keys (lazy so the
-    platform stack is only imported when targeted), and stack registration."""
+    """One platform's backend: codegen class, extra schema keys, and stack
+    registration. The esp32 fragments import their stack lazily because those
+    imports register esp32-only automations as a side effect; rp2040_ble is
+    side-effect-free, so it is imported at module scope (the cap constant
+    needs it there anyway)."""
 
     backend_class: cg.MockObjClass
     schema_fragment: Callable[[], cv.Schema]

esphome/components/bluetooth_connection/bluetooth_connection_hub.cpp

@@ -103,10 +103,10 @@ void BluetoothConnection::on_connection_state(bool connected, uint16_t mtu, int
     if (this->connection_type_ == ConnectionType::V3_WITH_CACHE) {
       // The API client has the services cached; never discover them. No
       // discovery phase needs the fast interval, so settle straight into the
-      // shared steady-state parameters. On esp32 the backend already set the
-      // same values as prefer-params before opening, so this request is
-      // usually redundant there - kept because rp2 has no prefer-params and
-      // the explicit update is its only path to the steady-state interval.
+      // shared steady-state parameters. Both backends already open cached
+      // connections with these values (esp32 prefer-params, rp2 initiating
+      // params), so this request is normally redundant - kept as a backstop
+      // in case the initial parameters were negotiated away.
       this->state_ = ClientState::ESTABLISHED;
       int param_err = this->backend_->update_connection_params(ble_device_base::MEDIUM_MIN_CONN_INTERVAL,
                                                                ble_device_base::MEDIUM_MAX_CONN_INTERVAL, 0,

esphome/components/bluetooth_connection/bluetooth_connection_hub.h

@@ -77,8 +77,9 @@ class BluetoothConnection final : public ble_device_base::GattClientListener {
   bool connected() const { return this->state_ == ClientState::ESTABLISHED; }
   void set_connection_type(ConnectionType ct) {
     this->connection_type_ = ct;
-    // The bluedroid backend branches on the type itself (prefer-params and
-    // the with-cache report at OPEN_EVT); the others ignore it.
+    // Both backends branch on the type before connecting (bluedroid picks
+    // prefer-params and the with-cache report at OPEN_EVT; rp2 picks the
+    // initiating parameters), so this must be set before the connect starts.
     this->backend_->set_connection_type(ct);
   }
   // Latched at discovery completion rather than read from the backend table:

esphome/components/bluetooth_connection/bluetooth_connection_rp2.cpp

@@ -26,6 +26,13 @@ using ble_device_base::GATT_ERR_NO_MEMORY;
 // and keeps the scan inhibited, so the engine cancels after 20 s. The
 // disconnect timeout mirrors the esp32 CLOSE_EVT safety net.
 static constexpr uint32_t CONNECT_TIMEOUT_MS = 20000;
+// Budget after a cancel is in flight: its completion normally lands within
+// tens of ms, and while the engine waits it pins the stack-wide connect slot,
+// so a lost completion must cost seconds, not another full connect budget.
+static constexpr uint32_t CONNECT_CANCEL_TIMEOUT_MS = 2000;
+// Pending engines re-attempt gap_connect on this cadence instead of every
+// loop pass: the DISALLOWED path (teardown overlap) takes BluetoothLock.
+static constexpr uint32_t CONNECT_RETRY_INTERVAL_MS = 50;
 // Can-send windows normally open within a connection interval (tens of ms).
 static constexpr uint32_t WRITE_NO_RSP_TIMEOUT_MS = 500;
 
@@ -54,6 +61,7 @@ RP2GattClient *RP2GattClient::instances[ESPHOME_BLE_GATT_CLIENT_COUNT] = {};
 uint8_t RP2GattClient::instance_count = 0;
 btstack_packet_callback_registration_t RP2GattClient::hci_event_registration = {};
 btstack_packet_callback_registration_t RP2GattClient::sm_event_registration = {};
+RP2GattClient *RP2GattClient::connect_owner = nullptr;
 // NOLINTEND(cppcoreguidelines-avoid-non-const-global-variables)
 
 static ESPBTUUID uuid_from_btstack(uint16_t uuid16, const uint8_t uuid128[16]) {
@@ -84,6 +92,7 @@ void RP2GattClient::setup() {
     // One locked section: the slot store lands before the count bump, and a
     // live HCI handler (N > 1 builds) cannot read a half-written registry.
     BluetoothLock lock;
+    this->engine_index_ = instance_count;
     instances[instance_count] = this;
     instance_count++;
     // One HCI event handler for all engine instances (BTstack supports
@@ -96,9 +105,24 @@ void RP2GattClient::setup() {
     }
   }
 
+#ifdef USE_OTA_STATE_LISTENER
+  ota::get_global_ota_callback()->add_global_state_listener(this);
+#endif
+
   this->disable_loop();
 }
 
+#ifdef USE_OTA_STATE_LISTENER
+void RP2GattClient::on_ota_global_state(ota::OTAState state, float progress, uint8_t error, ota::OTAComponent *comp) {
+  // esp32 parity (its tracker disconnects every client at OTA start): free
+  // the shared radio for the transfer. No restore needed; the client
+  // reconnects, and on success the device reboots anyway.
+  if (state == ota::OTA_STARTED && this->state_ != EngineState::IDLE) {
+    this->gatt_disconnect();
+  }
+}
+#endif
+
 float RP2GattClient::get_setup_priority() const { return setup_priority::AFTER_BLUETOOTH; }
 
 void RP2GattClient::dump_config() { ESP_LOGCONFIG(TAG, "RP2 GATT client (BTstack)"); }
@@ -124,34 +148,56 @@ void RP2GattClient::hci_packet_handler(uint8_t type, uint16_t channel, uint8_t *
       if (hci_event_gap_meta_get_subevent_code(packet) != GAP_SUBEVENT_LE_CONNECTION_COMPLETE) {
         break;
       }
-      bd_addr_t peer;
-      gap_subevent_le_connection_complete_get_peer_address(packet, peer);
       uint8_t status = gap_subevent_le_connection_complete_get_status(packet);
       hci_con_handle_t con_handle = gap_subevent_le_connection_complete_get_connection_handle(packet);
-      // Route to the engine that is waiting for this peer.
-      for (uint8_t i = 0; i < instance_count; i++) {
-        RP2GattClient *inst = instances[i];
-        if (inst->state_ == EngineState::CONNECTING && memcmp(inst->peer_addr_, peer, sizeof(bd_addr_t)) == 0) {
-          inst->enqueue_event_irq_(RP2GattEvent::CONNECTED, status, con_handle);
-          break;
+      bd_addr_t peer;
+      gap_subevent_le_connection_complete_get_peer_address(packet, peer);
+      // Route by ownership, not address: gap_connect refuses a new
+      // create-connection until the previous completion is processed, so the
+      // event belongs to the owner by construction. Cancel completions carry
+      // a zeroed peer address on this controller, so an address match would
+      // drop them and pin the owner until its backstop.
+      RP2GattClient *inst = connect_owner;
+      static constexpr bd_addr_t ZERO_ADDR = {};
+      if (inst != nullptr && memcmp(peer, ZERO_ADDR, sizeof(bd_addr_t)) != 0 &&
+          memcmp(inst->peer_addr_, peer, sizeof(bd_addr_t)) != 0) {
+        // Addressed completion for a peer the owner is not connecting to: a
+        // success delayed past a cancel and an ownership handoff (the cancel
+        // idles the stack's request immediately) must not stamp the old
+        // procedure's link onto the new owner. Zero-address (cancel)
+        // completions need no such guard: BTstack only emits them while its
+        // request state is idle, and a new owner re-arms that state when it
+        // claims the token, so a stale cancel completion is swallowed by the
+        // stack, never re-attributed. A successful stale link still needs
+        // disposal (same hazard as the unowned branch below).
+        if (status == 0) {
+          gap_disconnect(con_handle);
         }
+        break;
       }
+      connect_owner = nullptr;
+      if (inst == nullptr) {
+        if (status == 0) {
+          // Nobody owns this late link (the owner escalated first): tear it
+          // down here or the hci_connection_t leaks and the peer answers
+          // DISALLOWED until reboot.
+          gap_disconnect(con_handle);
+        }
+        break;
+      }
+      if (status == 0) {
+        // Stamp the handle here in the BTstack context: a disconnection
+        // racing the queued CONNECTED event arrives in this same context
+        // and must route by handle (it carries no address).
+        inst->con_handle_ = con_handle;
+      }
+      inst->enqueue_event_irq_(RP2GattEvent::CONNECTED, status, con_handle);
       break;
     }
     case HCI_EVENT_DISCONNECTION_COMPLETE: {
-      hci_con_handle_t con_handle = hci_event_disconnection_complete_get_connection_handle(packet);
-      RP2GattClient *inst = instance_for_con_handle(con_handle);
-      if (inst == nullptr && instance_count == 1) {
-        // The main loop may not have recorded the handle yet (the CONNECTED
-        // event is still queued); with a single engine the connecting
-        // instance is unambiguous, so route there to close the
-        // accept-then-drop window. With multiple engines the event has no
-        // address to match on, so it must be dropped instead of guessed.
-        RP2GattClient *candidate = instances[0];
-        if (candidate->con_handle_ == HCI_CON_HANDLE_INVALID && candidate->state_ != EngineState::IDLE) {
-          inst = candidate;
-        }
-      }
+      // Routable even against a still-queued CONNECTED event: the handle is
+      // stamped in this context at connection-complete time.
+      RP2GattClient *inst = instance_for_con_handle(hci_event_disconnection_complete_get_connection_handle(packet));
       if (inst != nullptr) {
         inst->enqueue_event_irq_(RP2GattEvent::DISCONNECTED, hci_event_disconnection_complete_get_reason(packet), 0);
       }
@@ -393,44 +439,73 @@ void RP2GattClient::loop() {
   if (dropped > 0) {
     // Control events must not be lost; the connection state is no longer
     // trustworthy — recover with a forced teardown.
-    ESP_LOGE(TAG, "Dropped %u GATT control events, disconnecting", dropped);
+    ESP_LOGE(TAG, "[%u] Dropped %u GATT control events, disconnecting", this->engine_index_, dropped);
     this->gatt_disconnect();
   }
   uint16_t notify_dropped = this->notify_queue_.get_and_reset_dropped_count();
   if (notify_dropped > 0) {
-    ESP_LOGW(TAG, "Dropped %u GATT notifications (queue full)", notify_dropped);
+    ESP_LOGW(TAG, "[%u] Dropped %u GATT notifications (queue full)", this->engine_index_, notify_dropped);
   }
 
-  if (this->state_ == EngineState::CONNECTING || this->state_ == EngineState::MTU_EXCHANGE) {
+  if (this->state_ == EngineState::CONNECT_PENDING) {
     uint32_t now = millis();
     if (now - this->connect_started_ > CONNECT_TIMEOUT_MS) {
-      ESP_LOGW(TAG, "Connect timeout");
-      if (this->state_ == EngineState::CONNECTING && this->con_handle_ == HCI_CON_HANDLE_INVALID) {
-        if (!this->connect_cancel_attempted_) {
-          this->connect_cancel_attempted_ = true;
-          BluetoothLock lock;
+      // Never reached the radio; nothing stack-side to cancel.
+      ESP_LOGW(TAG, "[%u] Connect timeout (queued)", this->engine_index_);
+      this->fail_connection_(HCI_REASON_CONNECTION_TIMEOUT);
+    } else if (now - this->connect_retry_ms_ >= CONNECT_RETRY_INTERVAL_MS) {
+      this->connect_retry_ms_ = now;
+      if (int err = this->try_gap_connect_(); err != 0) {
+        this->fail_connection_(static_cast<uint8_t>(err));
+      }
+    }
+  } else if (this->state_ == EngineState::CONNECTING || this->state_ == EngineState::MTU_EXCHANGE) {
+    uint32_t now = millis();
+    bool cancel_in_flight = this->state_ == EngineState::CONNECTING && this->con_handle_ == HCI_CON_HANDLE_INVALID &&
+                            this->connect_cancel_attempted_;
+    uint32_t budget = cancel_in_flight ? CONNECT_CANCEL_TIMEOUT_MS : CONNECT_TIMEOUT_MS;
+    if (now - this->connect_started_ > budget) {
+      ESP_LOGW(TAG, "[%u] Connect timeout", this->engine_index_);
+      bool link_up = this->state_ != EngineState::CONNECTING;
+      bool cancel_sent = false;
+      if (!link_up) {
+        BluetoothLock lock;
+        // Handle check under the lock: a success completion can stamp it in
+        // the BTstack context right up to this point, and escalating past a
+        // live link would orphan it (the queued CONNECTED event is dropped
+        // by the state guard once fail_connection_ runs).
+        link_up = this->con_handle_ != HCI_CON_HANDLE_INVALID;
+        if (!link_up && connect_owner == this) {
+          // gap_connect_cancel is stack-global; only the engine whose
+          // create-connection is in flight may issue it. First timeout:
+          // cancel and give the completion a grace period. Second: the
+          // completion was lost, re-issue the cancel in case the procedure
+          // still runs (a no-op on an idle stack), then escalate.
           gap_connect_cancel();
-          // The cancel produces a connection-complete event with a failure
-          // status, which drives the normal failure path; restart the timer
-          // so a lost event escalates below instead of wedging here.
-          this->connect_started_ = now;
-        } else {
-          // The cancel's completion never arrived: reclaim the slot and the
-          // scan rather than cancelling forever.
-          this->fail_connection_(HCI_REASON_CONNECTION_TIMEOUT);
+          cancel_sent = !this->connect_cancel_attempted_;
         }
-      } else {
-        // The link is up (MTU exchange stalled): tear it down properly so the
-        // controller frees its side; the DISCONNECTING safety net below
-        // reclaims state if the disconnection event is lost. Dropping engine
-        // state without gap_disconnect would leak the live link and the
-        // single GATT slot for the rest of the boot.
+        this->connect_cancel_attempted_ = true;
+      }
+      if (link_up) {
+        // The link is up (stamped mid-timeout or MTU exchange stalled): tear
+        // it down properly so the controller frees its side; the
+        // DISCONNECTING safety net below reclaims state if the disconnection
+        // event is lost. Dropping engine state without gap_disconnect would
+        // leak the live link and this engine's GATT slot for the rest of the
+        // boot.
         this->gatt_disconnect();
+      } else if (cancel_sent) {
+        // The cancel produces a connection-complete event with a failure
+        // status, which drives the normal failure path; restart the timer so
+        // a lost event escalates on the short cancel budget.
+        this->connect_started_ = now;
+      } else {
+        this->fail_connection_(HCI_REASON_CONNECTION_TIMEOUT);
       }
     }
   } else if (this->state_ == EngineState::DISCONNECTING) {
     if (millis() - this->disconnecting_started_ > ble_device_base::GATT_DISCONNECT_TIMEOUT_MS) {
-      ESP_LOGW(TAG, "Disconnect timeout, forcing idle");
+      ESP_LOGW(TAG, "[%u] Disconnect timeout, forcing idle", this->engine_index_);
       this->handle_disconnected_(HCI_REASON_CONNECTION_TIMEOUT);
     }
   } else if (this->state_ == EngineState::READY && this->op_type_ == OpType::WRITE_CHAR_NO_RSP &&
@@ -446,7 +521,7 @@ void RP2GattClient::loop() {
       }
     }
     if (timed_out) {
-      ESP_LOGW(TAG, "Deferred write timeout, handle=0x%04x", this->op_handle_);
+      ESP_LOGW(TAG, "[%u] Deferred write timeout, handle=0x%04x", this->engine_index_, this->op_handle_);
       this->listener_->on_write_result(this->op_handle_, GATT_CLIENT_BUSY);
     }
   } else if (this->state_ == EngineState::IDLE || (this->state_ == EngineState::READY && !this->op_in_flight_() &&
@@ -467,7 +542,7 @@ void RP2GattClient::handle_event_(const RP2GattEvent &event) {
     case RP2GattEvent::MTU_EXCHANGED:
       if (this->state_ == EngineState::MTU_EXCHANGE) {
         this->mtu_ = event.value;
-        ESP_LOGD(TAG, "MTU %u", this->mtu_);
+        ESP_LOGD(TAG, "[%u] MTU %u", this->engine_index_, this->mtu_);
         this->state_ = EngineState::READY;
         // Scanning resumes and runs alongside the established connection.
         this->release_scan_inhibit_();
@@ -515,7 +590,7 @@ void RP2GattClient::handle_connected_(uint8_t status, uint16_t con_handle) {
     return;
   }
   if (status != 0) {
-    ESP_LOGW(TAG, "Connect failed, status=0x%02x", status);
+    ESP_LOGW(TAG, "[%u] Connect failed, status=0x%02x", this->engine_index_, status);
     this->fail_connection_(status);
     return;
   }
@@ -539,7 +614,7 @@ void RP2GattClient::handle_connected_(uint8_t status, uint16_t con_handle) {
   }
   this->con_handle_ = con_handle;
   this->state_ = EngineState::MTU_EXCHANGE;
-  ESP_LOGD(TAG, "Link up, handle=0x%04x, negotiating MTU", con_handle);
+  ESP_LOGD(TAG, "[%u] Link up, handle=0x%04x, negotiating MTU", this->engine_index_, con_handle);
   BluetoothLock lock;
   // One wildcard listener covers notifications/indications for every
   // characteristic on this connection; the CCCD writes come from the API
@@ -564,6 +639,24 @@ void RP2GattClient::release_scan_inhibit_() {
 }
 
 void RP2GattClient::fail_connection_(uint8_t reason) {
+  {
+    // Timeout escalation can fire with the completion event lost; release the
+    // stack-wide connect slot so pending engines can proceed. Until the old
+    // completion is processed, gap_connect answers any peer with DISALLOWED
+    // (the request-level guard in hci.c); a cancel idles that request
+    // immediately, and a late addressed completion from the old procedure is
+    // then dropped by the owner-peer cross-check in the handler.
+    BluetoothLock lock;
+    if (connect_owner == this) {
+      connect_owner = nullptr;
+    }
+    if (this->state_ == EngineState::CONNECTING && this->con_handle_ != HCI_CON_HANDLE_INVALID) {
+      // A success completion stamped the handle between the escalation
+      // decision and this lock: tear the link down before cleanup wipes the
+      // handle, or it leaks its pool block for the rest of the boot.
+      gap_disconnect(this->con_handle_);
+    }
+  }
   this->cleanup_link_state_();
   this->release_scan_inhibit_();
   this->state_ = EngineState::IDLE;
@@ -577,14 +670,19 @@ void RP2GattClient::cleanup_link_state_() {
   while ((stale = this->notify_queue_.pop()) != nullptr) {
     this->notify_pool_.release(stale);
   }
-  // The wildcard listener is registered on the normal connect path right
-  // after con_handle_ is recorded; the cancel branch tears down before
-  // registering, where stop_listening on an unregistered entry is a no-op.
-  if (this->con_handle_ != HCI_CON_HANDLE_INVALID) {
+  // con_handle_ may be stamped in the BTstack context before the main loop
+  // registers the listener, so a valid handle does not imply a registration;
+  // stop_listening on an unregistered entry is a benign no-op. One lock
+  // scope around check and reset so an IRQ stamp cannot land in between
+  // (unreachable today — ownership is released before cleanup — but the
+  // invariant lives three functions away).
+  {
     BluetoothLock lock;
-    gatt_client_stop_listening_for_characteristic_value_updates(&this->notification_registration_);
+    if (this->con_handle_ != HCI_CON_HANDLE_INVALID) {
+      gatt_client_stop_listening_for_characteristic_value_updates(&this->notification_registration_);
+    }
+    this->con_handle_ = HCI_CON_HANDLE_INVALID;
   }
-  this->con_handle_ = HCI_CON_HANDLE_INVALID;
   this->notify_subscription_count_ = 0;
   this->cancel_requested_ = false;
   this->op_type_ = OpType::NONE;
@@ -596,7 +694,7 @@ void RP2GattClient::handle_disconnected_(uint8_t reason) {
   if (this->state_ == EngineState::IDLE) {
     return;
   }
-  ESP_LOGD(TAG, "Disconnected, reason=0x%02x", reason);
+  ESP_LOGD(TAG, "[%u] Disconnected, reason=0x%02x", this->engine_index_, reason);
   this->fail_connection_(reason);
 }
 
@@ -654,7 +752,7 @@ int RP2GattClient::discover_services() {
     RAMAllocator<ServiceArena> allocator(RAMAllocator<ServiceArena>::ALLOC_INTERNAL);
     this->arena_ = allocator.allocate(1);
     if (this->arena_ == nullptr) {
-      ESP_LOGE(TAG, "Service table allocation failed");
+      ESP_LOGE(TAG, "[%u] Service table allocation failed", this->engine_index_);
       return ble_device_base::GATT_ERR_NO_MEMORY;
     }
     new (this->arena_) ServiceArena();
@@ -760,8 +858,8 @@ void RP2GattClient::advance_discovery_(uint8_t att_status) {
 
 void RP2GattClient::finish_discovery_(int error) {
   this->discovery_phase_ = DiscoveryPhase::NONE;
-  ESP_LOGD(TAG, "Discovery done (err=%d): %u services, %u characteristics, %u descriptors", error, this->service_count_,
-           this->char_count_, this->desc_count_);
+  ESP_LOGD(TAG, "[%u] Discovery done (err=%d): %u services, %u characteristics, %u descriptors", this->engine_index_,
+           error, this->service_count_, this->char_count_, this->desc_count_);
   if (error == 0 && this->truncated_) {
     // A partial table must not stream: V3 clients cache the database
     // permanently, so an incomplete one would be wrong forever.
@@ -839,22 +937,68 @@ int RP2GattClient::connect(uint64_t address, uint8_t addr_type) {
   this->parent_->inhibit_scan();
   this->connect_cancel_attempted_ = false;
   this->cancel_requested_ = false;
+  // Bounds the queued wait; restarted when gap_connect is accepted so the
+  // radio attempt gets its full budget (HA's own ~20 s timeout arbitrates the
+  // sum via a disconnect request).
+  this->connect_started_ = millis();
+  if (int err = this->try_gap_connect_(); err != 0) {
+    this->release_scan_inhibit_();
+    return err;
+  }
+  this->enable_loop();
+  return 0;
+}
+
+// One outgoing LE create-connection exists stack-wide: issue it if no other
+// engine owns it, otherwise park in CONNECT_PENDING for loop() to retry.
+// Returns nonzero only for hard failures (state untouched; caller cleans up).
+int RP2GattClient::try_gap_connect_() {
+  // Unlocked peek: single core, aligned pointer; a stale value costs one loop
+  // pass and the locked re-check below is authoritative. Keeps the per-loop
+  // pending retry from taking BluetoothLock just to find the radio busy.
+  if (connect_owner != nullptr) {
+    this->state_ = EngineState::CONNECT_PENDING;
+    return 0;
+  }
   uint8_t status;
   {
     BluetoothLock lock;
-    gap_set_connection_parameters(CONN_SCAN_INTERVAL, CONN_SCAN_WINDOW, FAST_MIN_CONN_INTERVAL, FAST_MAX_CONN_INTERVAL,
-                                  0, FAST_CONN_TIMEOUT, CONN_CE_MIN, CONN_CE_MAX);
-    status = gap_connect(this->peer_addr_, this->peer_addr_type_);
+    if (connect_owner != nullptr) {
+      status = ERROR_CODE_COMMAND_DISALLOWED;
+    } else {
+      // esp32 parity: cached connections come up at MEDIUM already (nothing
+      // consumes the fast interval without a discovery phase), so there is no
+      // post-connect update procedure to race or silently lose; sustained
+      // FAST intervals also starve WiFi on the shared CYW43 radio.
+      // Without-cache runs FAST for discovery and steps down in
+      // finish_discovery_.
+      bool cached = this->connection_type_ == ble_device_base::ConnectionType::V3_WITH_CACHE;
+      gap_set_connection_parameters(CONN_SCAN_INTERVAL, CONN_SCAN_WINDOW,
+                                    cached ? MEDIUM_MIN_CONN_INTERVAL : FAST_MIN_CONN_INTERVAL,
+                                    cached ? MEDIUM_MAX_CONN_INTERVAL : FAST_MAX_CONN_INTERVAL, 0,
+                                    cached ? MEDIUM_CONN_TIMEOUT : FAST_CONN_TIMEOUT, CONN_CE_MIN, CONN_CE_MAX);
+      status = gap_connect(this->peer_addr_, this->peer_addr_type_);
+      if (status == 0) {
+        connect_owner = this;
+        // Still under the lock: a synthesized failure completion can fire in
+        // the BTstack context the instant it releases, and completion routing
+        // requires CONNECTING — set after the fact, the event is discarded
+        // and the engine burns its whole budget waiting for it.
+        this->state_ = EngineState::CONNECTING;
+        this->connect_started_ = millis();
+      }
+    }
   }
-  if (status != 0) {
-    ESP_LOGW(TAG, "gap_connect failed, status=0x%02x", status);
-    this->release_scan_inhibit_();
-    return status;
+  if (status == 0) {
+    return 0;
   }
-  this->state_ = EngineState::CONNECTING;
-  this->connect_started_ = millis();
-  this->enable_loop();
-  return 0;
+  if (status == ERROR_CODE_COMMAND_DISALLOWED) {
+    // Radio busy with another engine's connect; resolved from loop().
+    this->state_ = EngineState::CONNECT_PENDING;
+    return 0;
+  }
+  ESP_LOGW(TAG, "[%u] gap_connect failed, status=0x%02x", this->engine_index_, status);
+  return status;
 }
 
 int RP2GattClient::gatt_disconnect() {
@@ -863,6 +1007,10 @@ int RP2GattClient::gatt_disconnect() {
       return GATT_ERR_NOT_CONNECTED;
     case EngineState::DISCONNECTING:
       return 0;  // already on its way down
+    case EngineState::CONNECT_PENDING:
+      // Nothing issued stack-side; the invalid handle takes the refused
+      // path below without touching the stack.
+      break;
     case EngineState::CONNECTING: {
       if (this->con_handle_ == HCI_CON_HANDLE_INVALID) {
         // The cancel can lose the race against a successful connection
@@ -871,9 +1019,18 @@ int RP2GattClient::gatt_disconnect() {
         // attempt, so a lost completion escalates on the next timeout tick.
         this->cancel_requested_ = true;
         this->connect_cancel_attempted_ = true;
+        // Grace period for the cancel completion: the client's disconnect
+        // often lands right at the engine's own deadline, and without the
+        // restart the loop timeout fires first and reports before the
+        // completion can finish the teardown cleanly.
+        this->connect_started_ = millis();
         BluetoothLock lock;
-        gap_connect_cancel();
-        // Completion arrives as a failed connection-complete event.
+        // Owner: the cancel completes as a failed connection-complete. Not
+        // the owner (completion already resolved in the BTstack context): the
+        // queued event drives the same teardown, nothing to cancel.
+        if (connect_owner == this) {
+          gap_connect_cancel();
+        }
         return 0;
       }
       break;
@@ -881,20 +1038,23 @@ int RP2GattClient::gatt_disconnect() {
     default:
       break;
   }
-  uint8_t status;
-  {
-    BluetoothLock lock;
-    status = gap_disconnect(this->con_handle_);
-  }
-  if (status != 0) {
-    // Refused (handle already gone): complete via the event queue so the
-    // listener cannot re-enter disconnect() mid-call. BluetoothLock stops
-    // the IRQ producer, so this main-loop push is SPSC-safe.
-    ESP_LOGW(TAG, "gap_disconnect failed, status=0x%02x", status);
+  uint8_t status = ERROR_CODE_UNKNOWN_CONNECTION_IDENTIFIER;
+  if (this->con_handle_ != HCI_CON_HANDLE_INVALID) {
     {
       BluetoothLock lock;
-      this->enqueue_event_irq_(RP2GattEvent::DISCONNECTED, HCI_REASON_CONNECTION_TIMEOUT, 0);
+      status = gap_disconnect(this->con_handle_);
     }
+    if (status != 0) {
+      ESP_LOGW(TAG, "[%u] gap_disconnect failed, status=0x%02x", this->engine_index_, status);
+    }
+  }
+  if (status != 0) {
+    // Refused (handle already gone) or never issued (CONNECT_PENDING):
+    // complete via the event queue so the listener cannot re-enter
+    // disconnect mid-call. BluetoothLock stops the IRQ producer, so this
+    // main-loop push is SPSC-safe.
+    BluetoothLock lock;
+    this->enqueue_event_irq_(RP2GattEvent::DISCONNECTED, HCI_REASON_CONNECTION_TIMEOUT, 0);
   }
   this->state_ = EngineState::DISCONNECTING;
   this->disconnecting_started_ = millis();

esphome/components/bluetooth_connection/bluetooth_connection_rp2.h

@@ -19,6 +19,10 @@
 #include "esphome/core/helpers.h"
 #include "esphome/core/lock_free_queue.h"
 
+#ifdef USE_OTA_STATE_LISTENER
+#include "esphome/components/ota/ota_backend.h"
+#endif
+
 #include <btstack.h>
 
 #include <array>
@@ -71,7 +75,13 @@ static constexpr uint8_t RP2_GATT_EVENT_QUEUE_SIZE = 8;
 // full 512 B ATT payload, so depth buys burst tolerance at ~516 B per slot.
 static constexpr uint8_t RP2_GATT_NOTIFY_QUEUE_SIZE = 4;
 
-class RP2GattClient final : public Component, public Parented<rp2040_ble::RP2040BLE> {
+class RP2GattClient final : public Component,
+                            public Parented<rp2040_ble::RP2040BLE>
+#ifdef USE_OTA_STATE_LISTENER
+    ,
+                            public ota::OTAGlobalStateListener
+#endif
+{
  public:
   void setup() override;
   void loop() override;
@@ -95,18 +105,26 @@ class RP2GattClient final : public Component, public Parented<rp2040_ble::RP2040
   int pair();
   int update_connection_params(uint16_t min_interval, uint16_t max_interval, uint16_t latency, uint16_t timeout);
   ble_device_base::GattServiceTable get_service_table();
-  // No connection-type branching on this backend.
-  void set_connection_type(ble_device_base::ConnectionType ct) {}
+  // Cached connections initiate at MEDIUM parameters (esp32 parity); FAST is
+  // reserved for the discovery phase of uncached connects.
+  void set_connection_type(ble_device_base::ConnectionType ct) { this->connection_type_ = ct; }
   void release_services();
 
+#ifdef USE_OTA_STATE_LISTENER
+  // Drop the connection while an OTA runs (esp32 parity): an active link
+  // competes with the transfer for the shared radio.
+  void on_ota_global_state(ota::OTAState state, float progress, uint8_t error, ota::OTAComponent *comp) override;
+#endif
+
  protected:
   // Link/engine state. Discovery and GATT ops have their own cursors below —
   // the link stays READY while they run.
   enum class EngineState : uint8_t {
     IDLE,
-    CONNECTING,    // gap_connect issued, waiting for connection complete
-    MTU_EXCHANGE,  // link up, waiting for GATT_EVENT_MTU
-    READY,         // on_connection_state(true) delivered
+    CONNECT_PENDING,  // queued: another engine owns the stack-wide create-connection
+    CONNECTING,       // gap_connect issued, waiting for connection complete
+    MTU_EXCHANGE,     // link up, waiting for GATT_EVENT_MTU
+    READY,            // on_connection_state(true) delivered
     DISCONNECTING,
   };
 
@@ -143,6 +161,7 @@ class RP2GattClient final : public Component, public Parented<rp2040_ble::RP2040
   int issue_descriptor_query_(uint16_t char_index);
   void finish_discovery_(int error);
   void fail_connection_(uint8_t reason);
+  int try_gap_connect_();
   void cleanup_link_state_();
   bool notify_subscribed_(uint16_t handle) const;
   static void can_write_no_rsp_trampoline(void *context);
@@ -171,8 +190,12 @@ class RP2GattClient final : public Component, public Parented<rp2040_ble::RP2040
 
   // Group 3: 4-byte types
   uint32_t connect_started_{0};
+  uint32_t connect_retry_ms_{0};  // last CONNECT_PENDING gap_connect attempt
   uint32_t disconnecting_started_{0};
   uint32_t write_no_rsp_started_{0};
+  // Unscoped C enum, so int-sized: lives with the 4-byte members to keep the
+  // padding at the tail.
+  bd_addr_type_t peer_addr_type_{BD_ADDR_TYPE_LE_PUBLIC};
 
   // Group 4: 2-byte types (table counters written from the handler during
   // discovery, read from the main loop after the phase's QUERY_COMPLETE)
@@ -191,8 +214,9 @@ class RP2GattClient final : public Component, public Parented<rp2040_ble::RP2040
   // listener's deliveries on this list (esp32 parity for enable=false).
   std::array<uint16_t, RP2_GATT_MAX_NOTIFY_SUBSCRIPTIONS> notify_subscriptions_{};
   uint8_t notify_subscription_count_{0};
-  bd_addr_t peer_addr_{};  // MSB-first, as gap_connect expects
-  bd_addr_type_t peer_addr_type_{BD_ADDR_TYPE_LE_PUBLIC};
+  uint8_t engine_index_{0};  // position in instances[]; tags log lines per slot
+  bd_addr_t peer_addr_{};    // MSB-first, as gap_connect expects
+  ble_device_base::ConnectionType connection_type_{ble_device_base::ConnectionType::V3_WITHOUT_CACHE};
   EngineState state_{EngineState::IDLE};
   DiscoveryPhase discovery_phase_{DiscoveryPhase::NONE};
   OpType op_type_{OpType::NONE};
@@ -214,6 +238,12 @@ class RP2GattClient final : public Component, public Parented<rp2040_ble::RP2040
   static btstack_packet_callback_registration_t hci_event_registration;
   // NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
   static btstack_packet_callback_registration_t sm_event_registration;
+  // The engine whose gap_connect is in flight: BTstack allows one outgoing LE
+  // create-connection stack-wide, and gap_connect_cancel is global, so only
+  // the owner may cancel. Written under BluetoothLock from the main loop,
+  // cleared in the BTstack context when the procedure resolves.
+  // NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
+  static RP2GattClient *connect_owner;
 };
 
 }  // namespace esphome::bluetooth_connection

esphome/components/bluetooth_proxy/__init__.py

@@ -151,20 +151,23 @@ def _validate_no_active(config: ConfigType) -> ConfigType:
 @functools.cache
 def _rp2_config_schema() -> cv.All:
     """Full proxy on the rp2 BLE hub: active connections through the BTstack
-    GATT client backend in bluetooth_connection. The slot limit comes from the
-    prebuilt BTstack library (one connection today); the code is built for N."""
+    GATT client backend in bluetooth_connection. Multi-slot builds replace the
+    prebuilt library's one-client BTstack pools via linker --wrap, owned by
+    rp2040_ble and requested when a second backend registers."""
     connection_schema = bluetooth_connection.hub_connection_schema(PLATFORM_RP2)
 
     def populate_connections(config: ConfigType) -> ConfigType:
+        from esphome.components import rp2040_ble
+
         # One wrapper + backend pair per slot, declared during validation so
         # their ids exist for codegen (the esp32 arm's `connections` pattern).
         if not config[CONF_ACTIVE]:
             return config
+        connection_slots: int = config[CONF_CONNECTION_SLOTS]
+        rp2040_ble.consume_connection_slots(connection_slots, "bluetooth_proxy")(config)
         return {
             **config,
-            CONF_CONNECTIONS: [
-                connection_schema({}) for _ in range(config[CONF_CONNECTION_SLOTS])
-            ],
+            CONF_CONNECTIONS: [connection_schema({}) for _ in range(connection_slots)],
         }
 
     max_conn = bluetooth_connection.HUB_MAX_CONNECTIONS[PLATFORM_RP2]
@@ -182,8 +185,8 @@ def _rp2_config_schema() -> cv.All:
                         min=1,
                         max=max_conn,
                         msg=f"rp2 supports at most {max_conn} connection slot(s); "
-                        "the framework's BTstack library is built with "
-                        f"MAX_NR_GATT_CLIENTS {max_conn}",
+                        "the BTstack pool overrides in rp2040_ble are sized "
+                        f"for {max_conn}",
                     ),
                 ),
             }

esphome/components/rp2/__init__.py

@@ -388,6 +388,53 @@ async def to_code(config):
     _configure_lwip()
 
 
+# --- lwIP sizing. See _configure_lwip() for the platform comparison table. ---
+
+# TCP_SND_BUF: 4×MSS=5,840 matches ESP32. Down from arduino-pico's 8×MSS.
+# ESPAsyncWebServer allocates malloc(tcp_sndbuf()) per response chunk.
+LWIP_TCP_SND_BUF = "(4*TCP_MSS)"
+
+# TCP_WND: receive window. 4×MSS matches ESP32. Down from arduino-pico's 8×MSS.
+LWIP_TCP_WND = "(4*TCP_MSS)"
+
+# TCP_SND_QUEUELEN: max pbufs queued per PCB for the send buffer
+# ESP-IDF formula: (4 * TCP_SND_BUF + (TCP_MSS - 1)) / TCP_MSS
+# With 4×MSS: (4*5840 + 1459) / 1460 = 17 — match ESP32
+LWIP_TCP_SND_QUEUELEN = 17
+
+# MEMP_NUM_TCP_SEG: segment pool shared by every PCB, so it cannot be the
+# per-PCB queue length — lwIP's sanity check only demands >=, which is the
+# floor for a single connection. 2× lets two PCBs queue fully before the rest
+# start seeing ERR_MEM. Measured at 20 bytes per entry on this build, so the
+# whole pool is under 700 bytes.
+LWIP_MEMP_NUM_TCP_SEG = 2 * LWIP_TCP_SND_QUEUELEN
+
+# PBUF_POOL_SIZE: RP2040 has 264KB RAM, more generous than LibreTiny.
+# 16 matches ESP32 (vs arduino-pico's 24). Receive side only; the send path
+# copies into PBUF_RAM out of MEM_SIZE.
+LWIP_PBUF_POOL_SIZE = 16
+
+# MEM_SIZE: the lwIP heap that backs PBUF_RAM, which is where tcp_write()
+# copies outgoing data.
+#
+# TCP_OVERSIZE defaults to TCP_MSS, so every queued segment takes a full
+# MSS-sized block no matter how little was written: struct pbuf (16) +
+# PBUF_TRANSPORT offset (54) + TCP_MSS (1460) + the heap block header, about
+# 1.5KB each. A PCB at a full TCP_SND_BUF holds four of them, ~6KB.
+#
+# Two of those is ~12KB of arduino-pico's 16KB heap, and that is already
+# failing: mem.c is first-fit, so what has to be free is a *contiguous* 1.5KB
+# block, and at 75% occupancy interleaved with the small allocations (ARP
+# output queue, DHCP/DNS, mDNS TX) the largest run collapses long before the
+# total does. That is why the failure looks intermittent. With api's
+# max_connections on rp2 at 4, a third sender has nothing left to take. It
+# surfaces as ERR_MEM from tcp_write(), then a refused send_message().
+#
+# 32KB is arduino-pico's own next tier (its __LWIP_MEMMULT=2 boards).
+# Must stay under 64000 or lwIP widens mem_size_t to u32_t.
+LWIP_MEM_SIZE = 32768
+
+
 def _configure_lwip() -> None:
     """Configure lwIP options for RP2040 by generating a custom lwipopts.h.
 
@@ -407,25 +454,39 @@ def _configure_lwip() -> None:
     ────────────────────────────────────────────────────────────────
     TCP_SND_BUF               2×MSS   4×MSS  8×MSS         4×MSS
     TCP_WND                   4×MSS   4×MSS  8×MSS         4×MSS
+    TCP_SND_QUEUELEN          ~8      17     32            17
     MEM_LIBC_MALLOC           1       1      0             0*
     MEMP_MEM_MALLOC           1       1      0             0**
-    MEM_SIZE                  N/A***  N/A*** 16KB          16KB
+    MEM_SIZE                  N/A***  N/A*** 16KB          32KB
     PBUF_POOL_SIZE            10      16     24            16
-    MEMP_NUM_TCP_SEG          10      16     32            17
+    MEMP_NUM_TCP_SEG          10      16     32            34****
     MEMP_NUM_TCP_PCB          5       16     5             dynamic
-    MEMP_NUM_TCP_PCB_LISTEN   4       16     8****         dynamic
+    MEMP_NUM_TCP_PCB_LISTEN   4       16     8*****        dynamic
     MEMP_NUM_UDP_PCB          4       16     7             dynamic
-    TCP_SND_QUEUELEN          ~8      17     32            17
 
     * MEM_LIBC_MALLOC must stay 0: arduino-pico uses
       PICO_CYW43_ARCH_THREADSAFE_BACKGROUND which runs lwIP callbacks from
       a low-priority pendsv IRQ. The pico-sdk explicitly blocks
       MEM_LIBC_MALLOC=1 because libc malloc uses mutexes (unsafe in IRQ).
-    ** MEMP_MEM_MALLOC must stay 0: the dedicated lwIP heap (MEM_SIZE=16KB)
-      is too small to hold all pools dynamically. The PBUF_POOL alone needs
-      ~24KB (16 × 1524 bytes). Increasing MEM_SIZE would negate BSS savings.
-    *** ESP8266/ESP32 use MEM_LIBC_MALLOC=1 (system heap, no dedicated pool).
-    **** opt.h default; arduino-pico doesn't override MEMP_NUM_TCP_PCB_LISTEN.
+    ** MEMP_MEM_MALLOC must stay 0, and the reason is IRQ safety, not size.
+      memp_malloc() pops the pool free list inside SYS_ARCH_PROTECT, but
+      lwIP's heap picks its protection from LWIP_ALLOW_MEM_FREE_FROM_OTHER_
+      CONTEXT (default 0), so under NO_SYS=1 mem_malloc()/mem_free() run
+      with no protection at all. Setting MEMP_MEM_MALLOC=1 also moves the
+      allocation outside the guard entirely (memp.c calls mem_malloc before
+      SYS_ARCH_PROTECT). RX pbufs would then be allocated from the pendsv
+      IRQ on the same unguarded free list the main loop uses for
+      tcp_write(), which corrupts the heap. Tried on hardware; it faults
+      within seconds on CYW43. Ethernet survives only because that path
+      polls from the main loop and never runs a second context.
+    *** ESP8266/ESP32 use MEM_LIBC_MALLOC=1 (system heap, no dedicated
+      pool), so their MEMP_NUM_* and PBUF_POOL_SIZE values are not caps.
+      Ours are hard limits; do not copy their numbers without that in mind.
+    **** MEMP_NUM_TCP_SEG is a *global* pool while TCP_SND_QUEUELEN is
+      *per-PCB*, so it must not be sized to the per-PCB value: one busy
+      connection would drain it for every other PCB. 2× covers two PCBs at
+      full depth, and MEM_SIZE is the real backing limit past that.
+    ***** opt.h default; arduino-pico doesn't override MEMP_NUM_TCP_PCB_LISTEN.
     "dynamic" = auto-calculated from component socket registrations via
     socket.get_socket_counts() with minimums of 8 TCP / 6 UDP / 2 TCP_LISTEN.
     """
@@ -444,44 +505,20 @@ def _configure_lwip() -> None:
     # UDP PCBs (2) are absorbed by the generous minimum of 6.
     listening_tcp = max(MIN_TCP_LISTEN_SOCKETS, sc.tcp_listen)
 
-    # TCP_SND_BUF: 4×MSS=5,840 matches ESP32. Down from arduino-pico's 8×MSS.
-    # ESPAsyncWebServer allocates malloc(tcp_sndbuf()) per response chunk.
-    tcp_snd_buf = "(4*TCP_MSS)"
-
-    # TCP_WND: receive window. 4×MSS matches ESP32. Down from arduino-pico's 8×MSS.
-    tcp_wnd = "(4*TCP_MSS)"
-
-    # TCP_SND_QUEUELEN: max pbufs queued for send buffer
-    # ESP-IDF formula: (4 * TCP_SND_BUF + (TCP_MSS - 1)) / TCP_MSS
-    # With 4×MSS: (4*5840 + 1459) / 1460 = 17 — match ESP32
-    tcp_snd_queuelen = 17
-    # MEMP_NUM_TCP_SEG: segment pool, must be >= TCP_SND_QUEUELEN (lwIP sanity check)
-    memp_num_tcp_seg = tcp_snd_queuelen
-
-    # PBUF_POOL_SIZE: RP2040 has 264KB RAM, more generous than LibreTiny.
-    # 16 matches ESP32 (vs arduino-pico's 24). With MEMP_MEM_MALLOC=1,
-    # this is a max count (allocated on demand from heap).
-    pbuf_pool_size = 16
-
     # Build the lwIP override defines for the Jinja2 template.
     # The template uses #include_next to chain to the framework's original
     # lwipopts.h, then #undef/#define only the values we need to change.
     #
-    # Note: MEMP_MEM_MALLOC stays 0 (framework default). While the memp
-    # allocations use the dedicated lwIP heap (IRQ-safe), the 16KB MEM_SIZE
-    # is too small to hold all pools dynamically under stress. The PBUF_POOL
-    # alone needs ~24KB (16 × 1524 bytes). Increasing MEM_SIZE would negate
-    # the BSS savings.
-    #
-    # MEM_LIBC_MALLOC stays 0 (framework default): arduino-pico uses
-    # PICO_CYW43_ARCH_THREADSAFE_BACKGROUND which runs lwIP callbacks from
-    # a low-priority pendsv IRQ where libc malloc (mutex-based) is unsafe.
+    # Both malloc flags stay 0 (framework defaults); see the docstring. The
+    # static pools are the only IRQ-safe allocator on this platform, so the
+    # fix is to size them correctly rather than to make them dynamic.
     lwip_defines: dict[str, str] = {
-        "TCP_SND_BUF": tcp_snd_buf,
-        "TCP_WND": tcp_wnd,
-        "TCP_SND_QUEUELEN": str(tcp_snd_queuelen),
-        "MEMP_NUM_TCP_SEG": str(memp_num_tcp_seg),
-        "PBUF_POOL_SIZE": str(pbuf_pool_size),
+        "TCP_SND_BUF": LWIP_TCP_SND_BUF,
+        "TCP_WND": LWIP_TCP_WND,
+        "TCP_SND_QUEUELEN": str(LWIP_TCP_SND_QUEUELEN),
+        "MEM_SIZE": str(LWIP_MEM_SIZE),
+        "MEMP_NUM_TCP_SEG": str(LWIP_MEMP_NUM_TCP_SEG),
+        "PBUF_POOL_SIZE": str(LWIP_PBUF_POOL_SIZE),
         "MEMP_NUM_TCP_PCB": str(tcp_sockets),
         "MEMP_NUM_TCP_PCB_LISTEN": str(listening_tcp),
         "MEMP_NUM_UDP_PCB": str(udp_sockets),
@@ -500,7 +537,8 @@ def _configure_lwip() -> None:
     udp_min = " (min)" if udp_sockets > sc.udp else ""
     listen_min = " (min)" if listening_tcp > sc.tcp_listen else ""
     _LOGGER.info(
-        "Configuring lwIP: TCP=%d%s [%s], UDP=%d%s [%s], TCP_LISTEN=%d%s [%s]",
+        "Configuring lwIP: %d byte heap; TCP=%d%s [%s], UDP=%d%s [%s], TCP_LISTEN=%d%s [%s]",
+        LWIP_MEM_SIZE,
         tcp_sockets,
         tcp_min,
         sc.tcp_details,

esphome/components/rp2/lwipopts.h.jinja

@@ -20,13 +20,24 @@
 #undef TCP_WND
 #define TCP_WND                       {{ TCP_WND }}
 
-// Queued segment limits: derived from 4xMSS buffer size, matching ESP32
+// Per-PCB send queue: derived from 4xMSS buffer size, matching ESP32
 #undef TCP_SND_QUEUELEN
 #define TCP_SND_QUEUELEN              {{ TCP_SND_QUEUELEN }}
 
+// Segment pool: global across every PCB, so it is sized above the per-PCB
+// queue length rather than equal to it. lwIP's sanity check only requires
+// >= TCP_SND_QUEUELEN, which is the floor for a single connection.
 #undef MEMP_NUM_TCP_SEG
 #define MEMP_NUM_TCP_SEG              {{ MEMP_NUM_TCP_SEG }}
 
+// lwIP heap backing PBUF_RAM, which is what tcp_write() copies into.
+// Raised from arduino-pico's 16KB: TCP_OVERSIZE is TCP_MSS, so a single PCB
+// at a full TCP_SND_BUF pins about 6KB. Two of those left the 16KB heap at
+// 75%, and mem.c is first-fit, so the largest contiguous run ran out well
+// before the total did.
+#undef MEM_SIZE
+#define MEM_SIZE                      {{ MEM_SIZE }}
+
 // Packet buffer pool: 16 matches ESP32 (down from 24)
 #undef PBUF_POOL_SIZE
 #define PBUF_POOL_SIZE                {{ PBUF_POOL_SIZE }}

esphome/components/rp2040_ble/__init__.py

@@ -1,6 +1,9 @@
+from collections.abc import Callable, MutableMapping
+
 import esphome.codegen as cg
 import esphome.config_validation as cv
 from esphome.const import CONF_ENABLE_ON_BOOT, CONF_ID
+from esphome.core import CORE
 from esphome.types import ConfigType
 
 DEPENDENCIES = ["rp2"]
@@ -8,6 +11,15 @@ CODEOWNERS = ["@bdraco"]
 
 CONF_RP2040_BLE_ID = "rp2040_ble_id"
 
+KEY_RP2040_BLE = "rp2040_ble"
+KEY_USED_CONNECTION_SLOTS = "used_connection_slots"
+
+# Hard platform cap on concurrent GATT connections: the BTstack pool overrides
+# in btstack_memory.cpp are sized from ESPHOME_BLE_GATT_CLIENT_COUNT with this
+# as the ceiling. 3 matches the esp32 default and stays within the
+# controller's resources (MAX_NR_CONTROLLER_ACL_BUFFERS 3).
+MAX_CONNECTIONS = 3
+
 rp2040_ble_ns = cg.esphome_ns.namespace("rp2040_ble")
 RP2040BLE = rp2040_ble_ns.class_("RP2040BLE", cg.Component)
 
@@ -30,13 +42,67 @@ def _validate_board(config: ConfigType) -> ConfigType:
     return config
 
 
-FINAL_VALIDATE_SCHEMA = _validate_board
+def consume_connection_slots(
+    value: int, consumer: str
+) -> Callable[[MutableMapping], MutableMapping]:
+    """Reserve BLE connection slots for a component (the esp32_ble pattern);
+    the total is checked against MAX_CONNECTIONS in final validation."""
+
+    def _consume_connection_slots(config: MutableMapping) -> MutableMapping:
+        data: dict = CORE.data.setdefault(KEY_RP2040_BLE, {})
+        slots: list[str] = data.setdefault(KEY_USED_CONNECTION_SLOTS, [])
+        slots.extend([consumer] * value)
+        return config
+
+    return _consume_connection_slots
+
+
+def validate_connection_slots() -> None:
+    """Fail when consumers claimed more slots than the platform cap."""
+    # Skip in testing mode to allow component grouping (esp32_ble parity).
+    if CORE.testing_mode:
+        return
+    used = CORE.data.get(KEY_RP2040_BLE, {}).get(KEY_USED_CONNECTION_SLOTS, [])
+    if len(used) > MAX_CONNECTIONS:
+        raise cv.Invalid(
+            f"BLE components require {len(used)} connection slots but the "
+            f"rp2 maximum is {MAX_CONNECTIONS}. "
+            f"Components: {', '.join(used)}"
+        )
+
+
+def _final_validate(config: ConfigType) -> ConfigType:
+    _validate_board(config)
+    validate_connection_slots()
+    return config
+
+
+FINAL_VALIDATE_SCHEMA = _final_validate
 
 
 # Once per registered scan listener; sizes the controller's StaticVector
 # listener storage.
 request_scan_listener_slot = cg.slot_counter("RP2040_BLE_SCAN_LISTENER_COUNT")
 
+# The four btstack_memory accessors whose static pools are baked into the
+# prebuilt liblwip-bt.a; every internal use crosses an object boundary in the
+# archive, so --wrap intercepts them all (see btstack_memory.cpp).
+_BTSTACK_POOL_SYMBOLS = (
+    "btstack_memory_gatt_client_get",
+    "btstack_memory_gatt_client_free",
+    "btstack_memory_hci_connection_get",
+    "btstack_memory_hci_connection_free",
+)
+
+
+def add_btstack_pool_overrides() -> None:
+    """Emit the --wrap flags that swap the prebuilt BTstack pools for the
+    ESPHOME_BLE_GATT_CLIENT_COUNT-sized ones in btstack_memory.cpp. Called by
+    bluetooth_connection when a second GATT backend registers; idempotent
+    (build flags are a set)."""
+    for symbol in _BTSTACK_POOL_SYMBOLS:
+        cg.add_build_flag(f"-Wl,--wrap={symbol}")
+
 
 async def to_code(config: ConfigType) -> None:
     var = cg.new_Pvariable(config[CONF_ID])

esphome/components/rp2040_ble/btstack_memory.cpp

@@ -0,0 +1,118 @@
+// Replaces the gatt_client / hci_connection static pools baked into
+// arduino-pico's prebuilt liblwip-bt.a (built with MAX_NR_GATT_CLIENTS 1,
+// MAX_NR_HCI_CONNECTIONS 2) with pools sized from ESPHOME_BLE_GATT_CLIENT_COUNT.
+// add_btstack_pool_overrides() in this component's codegen emits the matching
+// -Wl,--wrap flags, requested by bluetooth_connection when more than one GATT
+// backend registers; single-backend builds emit no flags and this file
+// compiles to nothing, leaving the prebuilt pools in charge. Layout safety:
+// the framework defines ENABLE_CLASSIC / ENABLE_BLE for every user TU
+// whenever PIO_FRAMEWORK_ARDUINO_ENABLE_BLUETOOTH is set (this component
+// always sets it), so sizeof() here matches the archive.
+
+#include "esphome/core/defines.h"
+
+#if defined(USE_RP2040_BLE) && defined(USE_BLE_GATT_CLIENT) && (ESPHOME_BLE_GATT_CLIENT_COUNT > 1)
+
+#include <btstack.h>
+
+#include <cstring>
+
+namespace esphome::rp2040_ble {
+namespace {
+
+// Pinned against arduino-pico 6.0.0's prebuilt archives: a framework bump (or
+// a changed ENABLE_* macro) shifting the struct layout must fail the build
+// here, not overrun the pool blocks at runtime. Sizes differ per core
+// architecture (measured from each archive's own storage symbols). GCC only:
+// the clang-tidy frontend lays these structs out differently, and the guard
+// targets the real link.
+#ifndef __clang__
+#ifdef __riscv
+static_assert(sizeof(gatt_client_t) == 140 && sizeof(hci_connection_t) == 3740, "BTstack layout changed");
+#else
+static_assert(sizeof(gatt_client_t) == 128 && sizeof(hci_connection_t) == 3688, "BTstack layout changed");
+#endif
+#endif  // __clang__
+
+// One gatt_client_t per configured connection slot. An hci_connection_t is
+// held from gap_connect() to DISCONNECTION_COMPLETE (scanning holds none);
+// +1 mirrors the prebuilt library's own headroom (2 connections for 1 GATT
+// client) so a teardown/re-connect overlap can never starve a slot.
+constexpr int HCI_CONNECTION_POOL_SIZE = ESPHOME_BLE_GATT_CLIENT_COUNT + 1;
+
+// NOLINTBEGIN(cppcoreguidelines-avoid-non-const-global-variables,cert-err58-cpp)
+gatt_client_t gatt_client_storage[ESPHOME_BLE_GATT_CLIENT_COUNT];
+btstack_memory_pool_t gatt_client_pool;
+hci_connection_t hci_connection_storage[HCI_CONNECTION_POOL_SIZE];
+btstack_memory_pool_t hci_connection_pool;
+
+// Static init: pool_create only links a free list through its own storage,
+// and BTstack first allocates long after static construction.
+struct PoolInit {
+  PoolInit() {
+    btstack_memory_pool_create(&gatt_client_pool, gatt_client_storage, ESPHOME_BLE_GATT_CLIENT_COUNT,
+                               sizeof(gatt_client_t));
+    btstack_memory_pool_create(&hci_connection_pool, hci_connection_storage, HCI_CONNECTION_POOL_SIZE,
+                               sizeof(hci_connection_t));
+  }
+} pool_init;
+// NOLINTEND(cppcoreguidelines-avoid-non-const-global-variables,cert-err58-cpp)
+
+}  // namespace
+
+// Exact semantics of btstack_memory.c's static-pool arm: zeroed block on
+// success, NULL when exhausted; free returns the block to the pool. The
+// prebuilt pools stay resident in .bss (~7.4 KB, kept live by
+// btstack_memory_init in the archive) — dead weight here, not a leak.
+// NOLINTBEGIN(bugprone-reserved-identifier,cert-dcl37-c,cert-dcl51-cpp,readability-identifier-naming)
+extern "C" gatt_client_t *__real_btstack_memory_gatt_client_get(void);
+extern "C" void __real_btstack_memory_gatt_client_free(gatt_client_t *gatt_client);
+extern "C" hci_connection_t *__real_btstack_memory_hci_connection_get(void);
+extern "C" void __real_btstack_memory_hci_connection_free(hci_connection_t *hci_connection);
+
+namespace {
+// Fails the link if the corresponding --wrap flag is missing: __real_* only
+// exists while --wrap is in effect, and each wrap function anchors its own
+// symbol so dropping any single flag fails loudly. A code reference is used
+// because the framework links with --gc-sections, which discards an
+// unreferenced data anchor regardless of [[gnu::used]] (and this toolchain
+// does not emit SHF_GNU_RETAIN for [[gnu::retain]]).
+template<typename T> void anchor_wrap(T *symbol) { asm volatile("" ::"r"(symbol)); }
+}  // namespace
+
+extern "C" {
+
+gatt_client_t *__wrap_btstack_memory_gatt_client_get(void) {
+  anchor_wrap(&__real_btstack_memory_gatt_client_get);
+  void *buffer = btstack_memory_pool_get(&gatt_client_pool);
+  if (buffer != nullptr) {
+    memset(buffer, 0, sizeof(gatt_client_t));
+  }
+  return static_cast<gatt_client_t *>(buffer);
+}
+
+void __wrap_btstack_memory_gatt_client_free(gatt_client_t *gatt_client) {
+  anchor_wrap(&__real_btstack_memory_gatt_client_free);
+  btstack_memory_pool_free(&gatt_client_pool, gatt_client);
+}
+
+hci_connection_t *__wrap_btstack_memory_hci_connection_get(void) {
+  anchor_wrap(&__real_btstack_memory_hci_connection_get);
+  void *buffer = btstack_memory_pool_get(&hci_connection_pool);
+  if (buffer != nullptr) {
+    memset(buffer, 0, sizeof(hci_connection_t));
+  }
+  return static_cast<hci_connection_t *>(buffer);
+}
+
+void __wrap_btstack_memory_hci_connection_free(hci_connection_t *hci_connection) {
+  anchor_wrap(&__real_btstack_memory_hci_connection_free);
+  btstack_memory_pool_free(&hci_connection_pool, hci_connection);
+}
+
+}  // extern "C"
+// NOLINTEND(bugprone-reserved-identifier,cert-dcl37-c,cert-dcl51-cpp,readability-identifier-naming)
+
+}  // namespace esphome::rp2040_ble
+
+#endif  // USE_RP2040_BLE && USE_BLE_GATT_CLIENT && ESPHOME_BLE_GATT_CLIENT_COUNT > 1

esphome/core/defines.h

@@ -262,13 +262,13 @@
 #define USE_BLUETOOTH_PROXY
 // Mirror the codegen values per platform: _to_code_esp32() emits the connection
 // count (default 3) and the scanner-state push slot, _to_code_ble_hub() emits
-// the slot count (1 on rp2, 0 on advertisement-only hubs) — so static analysis
+// the slot count (3 on rp2, 0 on advertisement-only hubs) — so static analysis
 // checks the same instantiations a real build produces.
 #ifdef USE_ESP32
 #define USE_BLE_SCANNER_STATE_CALLBACK
 #define BLUETOOTH_PROXY_MAX_CONNECTIONS 3
 #elif defined(USE_RP2)
-#define BLUETOOTH_PROXY_MAX_CONNECTIONS 1
+#define BLUETOOTH_PROXY_MAX_CONNECTIONS 3
 #else
 #define BLUETOOTH_PROXY_MAX_CONNECTIONS 0
 #endif
@@ -482,7 +482,7 @@
 #define ESPHOME_BLE_DEVICE_BASE_LISTENER_COUNT 1
 #define USE_BLE_SCAN_RESPONSE_MERGER
 #define USE_BLE_GATT_CLIENT
-#define ESPHOME_BLE_GATT_CLIENT_COUNT 1
+#define ESPHOME_BLE_GATT_CLIENT_COUNT 3
 #define USE_RP2040_VARIANT_RP2040
 #define USE_SPI
 #ifndef USE_ETHERNET

tests/component_tests/bluetooth_proxy/test_platform_gates.py

@@ -142,8 +142,8 @@ def test_rp2_defaults_to_the_full_proxy(
     _register_tracker(PLATFORM_RP2)
     validated = bluetooth_proxy.CONFIG_SCHEMA({})
     assert validated[CONF_ACTIVE] is True
-    assert validated[bluetooth_proxy.CONF_CONNECTION_SLOTS] == 1
-    assert len(validated[bluetooth_proxy.CONF_CONNECTIONS]) == 1
+    assert validated[bluetooth_proxy.CONF_CONNECTION_SLOTS] == 3
+    assert len(validated[bluetooth_proxy.CONF_CONNECTIONS]) == 3
 
 
 def test_rp2_accepts_explicit_passive(
@@ -159,11 +159,15 @@ def test_rp2_accepts_explicit_passive(
 def test_rp2_rejects_slots_beyond_the_btstack_limit(
     set_core_config: SetCoreConfigCallable,
 ) -> None:
-    # The prebuilt BTstack library allows exactly one GATT client connection.
+    # The BTstack pool overrides are sized for RP2_MAX_CONNECTIONS slots.
     set_core_config(PlatformFramework.RP2_ARDUINO)
     _register_tracker(PLATFORM_RP2)
-    with pytest.raises(cv.Invalid, match="at most 1 connection slot"):
-        bluetooth_proxy.CONFIG_SCHEMA({"connection_slots": 2})
+    with pytest.raises(cv.Invalid, match="at most 3 connection slot"):
+        bluetooth_proxy.CONFIG_SCHEMA({"connection_slots": 4})
+    # Fewer slots than the cap stay accepted (the prebuilt single-client pool
+    # path for 1, the wrap path for 2).
+    validated = bluetooth_proxy.CONFIG_SCHEMA({"connection_slots": 1})
+    assert len(validated[bluetooth_proxy.CONF_CONNECTIONS]) == 1
     # Values past even the loosest platform cap stop at the outer walkable
     # schema, which stays bounded for range walkers (device-builder sync);
     # in-range values get the platform message above.

tests/component_tests/rp2040_ble/config/rp2_proxy_default.yaml

@@ -0,0 +1,15 @@
+esphome:
+  name: poolwrap-rp2-default
+
+rp2:
+  board: rpipicow
+
+wifi:
+  ssid: MySSID
+  password: password1
+
+api:
+
+rp2_ble_tracker:
+
+bluetooth_proxy:

tests/component_tests/rp2040_ble/config/rp2_proxy_single_slot.yaml

@@ -0,0 +1,16 @@
+esphome:
+  name: poolwrap-rp2-single
+
+rp2:
+  board: rpipicow
+
+wifi:
+  ssid: MySSID
+  password: password1
+
+api:
+
+rp2_ble_tracker:
+
+bluetooth_proxy:
+  connection_slots: 1

tests/component_tests/rp2040_ble/config/rp2_proxy_two_slots.yaml

@@ -0,0 +1,16 @@
+esphome:
+  name: poolwrap-rp2-two
+
+rp2:
+  board: rpipicow
+
+wifi:
+  ssid: MySSID
+  password: password1
+
+api:
+
+rp2_ble_tracker:
+
+bluetooth_proxy:
+  connection_slots: 2

tests/component_tests/rp2040_ble/test_connection_slots.py

@@ -0,0 +1,41 @@
+"""Connection-slot accounting: consumers claim against MAX_CONNECTIONS and
+final validation rejects over-subscription with the consumer list."""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from pathlib import Path
+
+import pytest
+
+from esphome import config_validation as cv
+from esphome.components import rp2040_ble
+from esphome.core import CORE
+
+
+def test_proxy_claims_its_slots_through_the_shared_accounting(
+    generate_main: Callable[[str | Path], str],
+    component_config_path: Callable[[str], Path],
+) -> None:
+    # A default (3-slot) proxy build records one claim per slot, attributed
+    # to the consumer, and passes final validation.
+    generate_main(component_config_path("rp2_proxy_default.yaml"))
+    used = CORE.data[rp2040_ble.KEY_RP2040_BLE][rp2040_ble.KEY_USED_CONNECTION_SLOTS]
+    assert used == ["bluetooth_proxy"] * 3
+
+
+def test_oversubscription_is_rejected_with_the_consumer_list() -> None:
+    # No YAML shape reaches this today (the proxy schema caps at the same
+    # limit); the guard exists for a second consumer such as ble_client.
+    rp2040_ble.consume_connection_slots(3, "bluetooth_proxy")({})
+    rp2040_ble.consume_connection_slots(1, "ble_client")({})
+    with pytest.raises(
+        cv.Invalid,
+        match=r"4 connection slots.*maximum is 3.*bluetooth_proxy.*ble_client",
+    ):
+        rp2040_ble.validate_connection_slots()
+
+
+def test_at_cap_passes() -> None:
+    rp2040_ble.consume_connection_slots(3, "bluetooth_proxy")({})
+    rp2040_ble.validate_connection_slots()

tests/component_tests/rp2040_ble/test_pool_wrap.py

@@ -0,0 +1,52 @@
+"""The rp2 BTstack pool overrides: multi-slot builds emit the --wrap flags
+that swap the prebuilt single-client pools for the codegen-sized ones;
+single-slot builds emit none and stay byte-identical to previous releases."""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from pathlib import Path
+
+from esphome.core import CORE
+
+from ..helpers import get_define_value
+
+# Spelled out rather than derived from rp2040_ble's symbol tuple, so a typo
+# in the component's list fails here instead of mirroring into the test.
+WRAP_FLAGS = (
+    "-Wl,--wrap=btstack_memory_gatt_client_get",
+    "-Wl,--wrap=btstack_memory_gatt_client_free",
+    "-Wl,--wrap=btstack_memory_hci_connection_get",
+    "-Wl,--wrap=btstack_memory_hci_connection_free",
+)
+
+
+def test_default_slots_emit_the_pool_wrap(
+    generate_main: Callable[[str | Path], str],
+    component_config_path: Callable[[str], Path],
+) -> None:
+    generate_main(component_config_path("rp2_proxy_default.yaml"))
+    assert all(flag in CORE.build_flags for flag in WRAP_FLAGS)
+    assert get_define_value("ESPHOME_BLE_GATT_CLIENT_COUNT") == "3"
+    assert get_define_value("BLUETOOTH_PROXY_MAX_CONNECTIONS") == "3"
+
+
+def test_two_slots_emit_the_pool_wrap(
+    generate_main: Callable[[str | Path], str],
+    component_config_path: Callable[[str], Path],
+) -> None:
+    # Two slots: the wrap pools are smaller than the cap, sized from the count.
+    generate_main(component_config_path("rp2_proxy_two_slots.yaml"))
+    assert all(flag in CORE.build_flags for flag in WRAP_FLAGS)
+    assert get_define_value("ESPHOME_BLE_GATT_CLIENT_COUNT") == "2"
+    assert get_define_value("BLUETOOTH_PROXY_MAX_CONNECTIONS") == "2"
+
+
+def test_single_slot_keeps_the_prebuilt_pools(
+    generate_main: Callable[[str | Path], str],
+    component_config_path: Callable[[str], Path],
+) -> None:
+    generate_main(component_config_path("rp2_proxy_single_slot.yaml"))
+    assert not any(flag in CORE.build_flags for flag in WRAP_FLAGS)
+    assert get_define_value("ESPHOME_BLE_GATT_CLIENT_COUNT") == "1"
+    assert get_define_value("BLUETOOTH_PROXY_MAX_CONNECTIONS") == "1"

tests/components/bluetooth_connection/validate.rp2040-ard.yaml

@@ -6,6 +6,7 @@ packages:
 
 rp2_ble_tracker:
 
+# Two slots: the one shape where the wrap pools are smaller than the cap.
 bluetooth_proxy:
   active: true
-  connection_slots: 1
+  connection_slots: 2

tests/components/bluetooth_proxy/test.rp2040-ard.yaml

@@ -1,5 +1,7 @@
 # Full proxy on the rp2 BLE hub: active defaults to true here (esp32 parity),
-# so this compiles the BTstack GATT client backend and one connection slot.
+# so this compiles the BTstack GATT client backend with the default three
+# connection slots, exercising the rp2040_ble/btstack_memory.cpp pool --wrap
+# link.
 # No explicit ble_hub_id: the generated binding resolves the single declared
 # hub, and an inline id here would collide with rp2_ble_tracker's own fixture
 # once CI merges both components into one grouped rp2040-ard build (grouped

tests/components/bluetooth_proxy/test.rp2350-ard.yaml

@@ -0,0 +1,9 @@
+# Pico 2 W build of the full proxy: links the rp2350 framework archive, so
+# the pool --wrap overrides and their per-architecture layout asserts are
+# exercised for this chip too (see test.rp2040-ard.yaml for the slot shape).
+packages:
+  common: !include common.yaml
+
+rp2_ble_tracker:
+
+bluetooth_proxy:

tests/test_build_components/build_components_base.rp2350-ard.yaml

@@ -2,8 +2,10 @@ esphome:
   name: componenttestrp2040pico2ard
   friendly_name: $component_name
 
+# rpipico2w: superset of rpipico2 with the CYW43 radio, so wireless
+# components (wifi, BLE) can share this target too.
 rp2:
-  board: rpipico2
+  board: rpipico2w
 
 logger:
   level: VERY_VERBOSE

tests/unit_tests/components/test_rp2.py

@@ -93,3 +93,29 @@ def test_rp2040_submodule_imports_resolve_to_rp2_submodules() -> None:
 
     assert rp2040_boards is rp2_boards
     assert rp2040_generate is rp2_generate
+
+
+def test_lwip_segment_pool_exceeds_per_pcb_queue() -> None:
+    """The segment pool is global while the send queue is per-PCB.
+
+    lwIP's sanity check only requires ``MEMP_NUM_TCP_SEG >= TCP_SND_QUEUELEN``,
+    which is the floor for a *single* connection: at equality one busy PCB can
+    drain the pool for every other PCB. Dropping back to equality would rebuild
+    the starvation this sizing exists to prevent, and nothing in the build would
+    complain.
+    """
+    from esphome.components import rp2
+
+    assert rp2.LWIP_MEMP_NUM_TCP_SEG >= rp2.LWIP_TCP_SND_QUEUELEN
+    assert rp2.LWIP_MEMP_NUM_TCP_SEG >= 2 * rp2.LWIP_TCP_SND_QUEUELEN
+
+
+def test_lwip_mem_size_keeps_mem_size_t_narrow() -> None:
+    """``MEM_SIZE`` above 64000 silently widens lwIP's ``mem_size_t`` to
+    ``u32_t`` (``lwip/mem.h``), growing the header on every heap block. Raising
+    the heap past that bound is a real option, but it should be a deliberate
+    one rather than a side effect of tuning.
+    """
+    from esphome.components import rp2
+
+    assert rp2.LWIP_MEM_SIZE < 64000