Merge remote-tracking branch 'upstream/dev' into neutral-ble-client

The registry entry now carries its materializer gate, so rp2 builds stop
receiving a Bluedroid define. Also: parent before the capacity check in
register_gatt_node (neutral parity) and BluedroidServiceTable is
non-copyable (owning raw pointer).

esphome/components/ble_client/__init__.py

@@ -1,8 +1,23 @@
+from collections.abc import Callable
+import functools
+from typing import Any
+
 from esphome import automation
 from esphome.automation import maybe_simple_id
 import esphome.codegen as cg
-from esphome.components import esp32_ble, esp32_ble_client, esp32_ble_tracker
-from esphome.components.esp32_ble import BTLoggers
+from esphome.components import ble_device_base, bluetooth_connection
+from esphome.components.ble_device_base import (
+    BT_UUID16_FORMAT as bt_uuid16_format,
+    BT_UUID32_FORMAT as bt_uuid32_format,
+    BT_UUID128_FORMAT as bt_uuid128_format,
+    as_hex,
+    as_reversed_hex_array,
+    bt_uuid,
+)
+from esphome.config_helpers import (
+    filter_source_files_from_platform,
+    frameworks_for_platforms,
+)
 import esphome.config_validation as cv
 from esphome.const import (
     CONF_CHARACTERISTIC_UUID,
@@ -15,13 +30,53 @@ from esphome.const import (
     CONF_SERVICE_UUID,
     CONF_TRIGGER_ID,
     CONF_VALUE,
+    PLATFORM_ESP32,
+    PlatformFramework,
 )
-from esphome.core import ID
+from esphome.core import CORE, ID
+from esphome.enum import StrEnum
+from esphome.schema_extractors import SCHEMA_EXTRACT, schema_extractor
 from esphome.types import ConfigType
 
-AUTO_LOAD = ["esp32_ble_client"]
+# The esp32 BLE stack (esp32_ble, esp32_ble_tracker) is imported lazily inside
+# the esp32 schema/codegen arms: importing those modules registers esp32-only
+# automations as a side effect, which must not leak into the neutral
+# platforms' registries (the bluetooth_proxy pattern).
+
+
+def _legacy_engine() -> bool:
+    """True when the build uses the legacy raw-gattc engine - one line to
+    flip when esp32 moves to the neutral engine (with
+    USE_BLE_CLIENT_LEGACY_ENGINE in _to_code_esp32)."""
+    return CORE.is_esp32
+
+
+def AUTO_LOAD() -> list[str]:
+    """The engine's closure per platform: the legacy esp32 engine builds on
+    esp32_ble_client plus bluetooth_connection (the shared service-table
+    materializer; its sources compile empty in builds without a neutral
+    node), the neutral engine on the bluetooth_connection backend. The
+    platform-less arm is the union for manifest-resolving tooling."""
+    if _legacy_engine() or CORE.target_platform is None:
+        return ["bluetooth_connection", "esp32_ble_client"]
+    return ["bluetooth_connection"]
+
+
 CODEOWNERS = ["@buxtronix", "@clydebarrow"]
-DEPENDENCIES = ["esp32_ble_tracker"]
+
+FILTER_SOURCE_FILES = filter_source_files_from_platform(
+    {
+        "ble_client.cpp": {
+            PlatformFramework.ESP32_ARDUINO,
+            PlatformFramework.ESP32_IDF,
+        },
+        # Every framework of every non-esp32 registry platform: a platform
+        # that validates the neutral arm must also compile the neutral engine.
+        "ble_client_gatt.cpp": frameworks_for_platforms(
+            set(bluetooth_connection.GATT_CLIENT_PLATFORMS) - {PLATFORM_ESP32}
+        ),
+    }
+)
 
 CONF_DESCRIPTOR_UUID = "descriptor_uuid"
 CONF_ON_NOTIFY = "on_notify"
@@ -58,7 +113,9 @@ def notify_from_on_notify(config: ConfigType) -> ConfigType:
 
 
 ble_client_ns = cg.esphome_ns.namespace("ble_client")
-BLEClient = ble_client_ns.class_("BLEClient", esp32_ble_client.BLEClientBase)
+# One codegen class for both engines: the exclusively-gated headers resolve
+# the name to exactly one C++ definition per build.
+BLEClient = ble_client_ns.class_("BLEClient", cg.Component)
 BLEClientNode = ble_client_ns.class_("BLEClientNode")
 BLEClientNodeConstRef = BLEClientNode.operator("ref").operator("const")
 # Triggers
@@ -105,62 +162,179 @@ CONF_AUTO_CONNECT = "auto_connect"
 
 MULTI_CONF = True
 
-CONFIG_SCHEMA = cv.All(
-    cv.Schema(
-        {
-            cv.GenerateID(): cv.declare_id(BLEClient),
-            cv.Required(CONF_MAC_ADDRESS): cv.mac_address,
-            cv.Optional(CONF_NAME): cv.string,
-            cv.Optional(CONF_AUTO_CONNECT, default=True): cv.boolean,
-            cv.Optional(CONF_ON_CONNECT): automation.validate_automation(
-                {
-                    cv.GenerateID(CONF_TRIGGER_ID): cv.declare_id(
-                        BLEClientConnectTrigger
-                    ),
-                }
-            ),
-            cv.Optional(CONF_ON_DISCONNECT): automation.validate_automation(
-                {
-                    cv.GenerateID(CONF_TRIGGER_ID): cv.declare_id(
-                        BLEClientDisconnectTrigger
-                    ),
-                }
-            ),
-            cv.Optional(CONF_ON_PASSKEY_REQUEST): automation.validate_automation(
-                {
-                    cv.GenerateID(CONF_TRIGGER_ID): cv.declare_id(
-                        BLEClientPasskeyRequestTrigger
-                    ),
-                }
-            ),
-            cv.Optional(CONF_ON_PASSKEY_NOTIFICATION): automation.validate_automation(
-                {
-                    cv.GenerateID(CONF_TRIGGER_ID): cv.declare_id(
-                        BLEClientPasskeyNotificationTrigger
-                    ),
-                }
-            ),
-            cv.Optional(
-                CONF_ON_NUMERIC_COMPARISON_REQUEST
-            ): automation.validate_automation(
-                {
-                    cv.GenerateID(CONF_TRIGGER_ID): cv.declare_id(
-                        BLEClientNumericComparisonRequestTrigger
-                    ),
-                }
-            ),
-        }
+# Keys shared by both engines' schemas.
+_COMMON_SCHEMA = cv.Schema(
+    {
+        cv.GenerateID(): cv.declare_id(BLEClient),
+        cv.Required(CONF_MAC_ADDRESS): cv.mac_address,
+        cv.Optional(CONF_AUTO_CONNECT, default=True): cv.boolean,
+        cv.Optional(CONF_ON_CONNECT): automation.validate_automation(
+            {
+                cv.GenerateID(CONF_TRIGGER_ID): cv.declare_id(BLEClientConnectTrigger),
+            }
+        ),
+        cv.Optional(CONF_ON_DISCONNECT): automation.validate_automation(
+            {
+                cv.GenerateID(CONF_TRIGGER_ID): cv.declare_id(
+                    BLEClientDisconnectTrigger
+                ),
+            }
+        ),
+    }
+).extend(cv.COMPONENT_SCHEMA)
+
+
+@functools.cache
+def _esp32_config_schema() -> cv.All:
+    """The legacy engine's schema, byte-compatible with what esp32 always had
+    (including the Bluedroid security triggers)."""
+    from esphome.components import esp32_ble_tracker
+
+    return cv.All(
+        _COMMON_SCHEMA.extend(
+            {
+                # Accepted-but-unused legacy key; not propagated to the
+                # neutral schema.
+                cv.Optional(CONF_NAME): cv.string,
+                cv.Optional(CONF_ON_PASSKEY_REQUEST): automation.validate_automation(
+                    {
+                        cv.GenerateID(CONF_TRIGGER_ID): cv.declare_id(
+                            BLEClientPasskeyRequestTrigger
+                        ),
+                    }
+                ),
+                cv.Optional(
+                    CONF_ON_PASSKEY_NOTIFICATION
+                ): automation.validate_automation(
+                    {
+                        cv.GenerateID(CONF_TRIGGER_ID): cv.declare_id(
+                            BLEClientPasskeyNotificationTrigger
+                        ),
+                    }
+                ),
+                cv.Optional(
+                    CONF_ON_NUMERIC_COMPARISON_REQUEST
+                ): automation.validate_automation(
+                    {
+                        cv.GenerateID(CONF_TRIGGER_ID): cv.declare_id(
+                            BLEClientNumericComparisonRequestTrigger
+                        ),
+                    }
+                ),
+            }
+        ).extend(esp32_ble_tracker.ESP_BLE_DEVICE_SCHEMA),
+        bluetooth_connection.consume_gatt_slot("ble_client"),
     )
-    .extend(cv.COMPONENT_SCHEMA)
-    .extend(esp32_ble_tracker.ESP_BLE_DEVICE_SCHEMA),
-    esp32_ble.consume_connection_slots(1, "ble_client"),
-)
+
+
+@functools.cache
+def _gatt_config_schema(platform: str) -> cv.All:
+    """The neutral engine's schema: the shared keys plus the hub reference
+    (parsed-advertisement sightings) and the GATT backend declaration.
+    Keyed by platform - the backend fragment differs per platform."""
+    return cv.All(
+        _COMMON_SCHEMA.extend(ble_device_base.BLE_DEVICE_SCHEMA).extend(
+            bluetooth_connection.gatt_client_schema(platform)
+        ),
+        bluetooth_connection.consume_gatt_slot("ble_client"),
+    )
+
+
+@schema_extractor("schema")
+def _validate_platform(config: ConfigType) -> ConfigType:
+    if config is SCHEMA_EXTRACT:
+        # Deliberate gap (the bluetooth_proxy pattern): the dumper gets only
+        # this shape, so the neutral arm's ble_hub_id is absent from editor
+        # schemas and the esp32-only keys are advertised on every platform.
+        # The language-schema dumper runs without a platform; expose the
+        # esp32 (legacy-engine) shape.
+        return _esp32_config_schema()
+    if _legacy_engine():
+        return _esp32_config_schema()(config)
+    if CORE.target_platform in bluetooth_connection.GATT_CLIENT_PLATFORMS:
+        return _gatt_config_schema(CORE.target_platform)(config)
+    raise cv.Invalid(f"ble_client is not supported on {CORE.target_platform}")
+
+
+CONFIG_SCHEMA = _validate_platform
 
 CONF_BLE_CLIENT_ID = "ble_client_id"
 
+
+class BLEClientFeatures(StrEnum):
+    """Per-platform engine capabilities consumers declare against."""
+
+    # The platform-neutral node interface (on_connected/table + completion
+    # callbacks) - every platform with a ble_client engine.
+    GATT_NODE = "gatt_node"
+    # The raw esp32 GATT client event stream (gattc/gap handlers,
+    # node_state) - the legacy engine only.
+    RAW_GATTC = "raw_gattc"
+    # Pairing dialog replies and bond management (Bluedroid GAP/SMP).
+    SECURITY = "security"
+
+
+def _engine_features() -> set[BLEClientFeatures]:
+    """Features the validated platform's engine provides."""
+    if _legacy_engine():
+        return {
+            BLEClientFeatures.GATT_NODE,
+            BLEClientFeatures.RAW_GATTC,
+            BLEClientFeatures.SECURITY,
+        }
+    if CORE.target_platform in bluetooth_connection.GATT_CLIENT_PLATFORMS:
+        return {BLEClientFeatures.GATT_NODE}
+    return set()
+
+
+def requires_feature(
+    feature: BLEClientFeatures, description: str
+) -> Callable[[Any], Any]:
+    """Validator gating a consumer to platforms whose engine provides
+    `feature`, naming the missing capability in the error."""
+
+    def validator(value: Any) -> Any:
+        features = _engine_features()
+        if feature not in features:
+            available = (
+                f"; this platform's engine provides: {', '.join(sorted(features))}"
+                if features
+                else ""
+            )
+            raise cv.Invalid(
+                f"{description} requires the ble_client '{feature}' feature, "
+                f"which {CORE.target_platform} does not provide{available}"
+            )
+        return value
+
+    return validator
+
+
+# The one choke point for every node component still on the raw esp32 event
+# stream; migrating to the neutral interface (NODE_BLE_CLIENT_SCHEMA +
+# register_gatt_node) lifts it.
+_legacy_engine_only = requires_feature(
+    BLEClientFeatures.RAW_GATTC,
+    "This component drives the raw ESP32 GATT client events and has not "
+    "been migrated to the platform-neutral node interface yet; it",
+)
+
 BLE_CLIENT_SCHEMA = cv.Schema(
     {
-        cv.GenerateID(CONF_BLE_CLIENT_ID): cv.use_id(BLEClient),
+        cv.GenerateID(CONF_BLE_CLIENT_ID): cv.All(
+            cv.use_id(BLEClient), _legacy_engine_only
+        ),
+    }
+)
+
+# For node components on the neutral interface: valid wherever ble_client
+# itself is.
+NODE_BLE_CLIENT_SCHEMA = cv.Schema(
+    {
+        cv.GenerateID(CONF_BLE_CLIENT_ID): cv.All(
+            cv.use_id(BLEClient),
+            requires_feature(BLEClientFeatures.GATT_NODE, "This component"),
+        ),
     }
 )
 
@@ -170,11 +344,31 @@ async def register_ble_node(var, config):
     cg.add(parent.register_ble_node(var))
 
 
+def _request_gatt_node_build() -> None:
+    """Node storage and the one define meaning "the neutral node surface is
+    compiled in", plus the esp32 bridge/materializer defines."""
+    _request_node_slot()
+    cg.add_define("USE_BLE_CLIENT_GATT_NODES")
+    if _legacy_engine():
+        # Deliberately not ble_device_base.request_gatt_client(): that would
+        # claim a phantom backend slot on combined proxy builds.
+        cg.add_define("USE_BLE_GATT_CLIENT")
+        cg.add_define("USE_BLE_GATT_BACKEND_BLUEDROID")
+        cg.add_define("USE_BLUEDROID_GATT_SERVICE_TABLE")
+
+
+async def register_gatt_node(var, config):
+    """Register a node on the platform-neutral interface (both engines)."""
+    parent = await cg.get_variable(config[CONF_BLE_CLIENT_ID])
+    _request_gatt_node_build()
+    cg.add(parent.register_gatt_node(var))
+
+
 BLE_WRITE_ACTION_SCHEMA = cv.Schema(
     {
         cv.GenerateID(CONF_ID): cv.use_id(BLEClient),
-        cv.Required(CONF_SERVICE_UUID): esp32_ble_tracker.bt_uuid,
-        cv.Required(CONF_CHARACTERISTIC_UUID): esp32_ble_tracker.bt_uuid,
+        cv.Required(CONF_SERVICE_UUID): bt_uuid,
+        cv.Required(CONF_CHARACTERISTIC_UUID): bt_uuid,
         cv.Required(CONF_VALUE): cv.templatable(cv.ensure_list(cv.hex_uint8_t)),
     }
 )
@@ -185,25 +379,34 @@ BLE_CONNECT_ACTION_SCHEMA = maybe_simple_id(
     }
 )
 
-BLE_NUMERIC_COMPARISON_REPLY_ACTION_SCHEMA = cv.Schema(
-    {
-        cv.GenerateID(CONF_ID): cv.use_id(BLEClient),
-        cv.Required(CONF_ACCEPT): cv.templatable(cv.boolean),
-    }
+BLE_NUMERIC_COMPARISON_REPLY_ACTION_SCHEMA = cv.All(
+    requires_feature(BLEClientFeatures.SECURITY, "This action"),
+    cv.Schema(
+        {
+            cv.GenerateID(CONF_ID): cv.use_id(BLEClient),
+            cv.Required(CONF_ACCEPT): cv.templatable(cv.boolean),
+        }
+    ),
 )
 
-BLE_PASSKEY_REPLY_ACTION_SCHEMA = cv.Schema(
-    {
-        cv.GenerateID(CONF_ID): cv.use_id(BLEClient),
-        cv.Required(CONF_PASSKEY): cv.templatable(cv.int_range(min=0, max=999999)),
-    }
+BLE_PASSKEY_REPLY_ACTION_SCHEMA = cv.All(
+    requires_feature(BLEClientFeatures.SECURITY, "This action"),
+    cv.Schema(
+        {
+            cv.GenerateID(CONF_ID): cv.use_id(BLEClient),
+            cv.Required(CONF_PASSKEY): cv.templatable(cv.int_range(min=0, max=999999)),
+        }
+    ),
 )
 
 
-BLE_REMOVE_BOND_ACTION_SCHEMA = cv.Schema(
-    {
-        cv.GenerateID(CONF_ID): cv.use_id(BLEClient),
-    }
+BLE_REMOVE_BOND_ACTION_SCHEMA = cv.All(
+    requires_feature(BLEClientFeatures.SECURITY, "This action"),
+    cv.Schema(
+        {
+            cv.GenerateID(CONF_ID): cv.use_id(BLEClient),
+        }
+    ),
 )
 
 
@@ -237,6 +440,8 @@ async def ble_connect_to_code(config, action_id, template_arg, args):
 )
 async def ble_write_to_code(config, action_id, template_arg, args):
     parent = await cg.get_variable(config[CONF_ID])
+    # The action registers itself as a neutral node in its constructor.
+    _request_gatt_node_build()
     var = cg.new_Pvariable(action_id, template_arg, parent)
 
     value = config[CONF_VALUE]
@@ -251,38 +456,20 @@ async def ble_write_to_code(config, action_id, template_arg, args):
         arr = cg.static_const_array(arr_id, cg.ArrayInitializer(*value))
         cg.add(var.set_value_simple(arr, len(value)))
 
-    if len(config[CONF_SERVICE_UUID]) == len(esp32_ble_tracker.bt_uuid16_format):
-        cg.add(
-            var.set_service_uuid16(esp32_ble_tracker.as_hex(config[CONF_SERVICE_UUID]))
-        )
-    elif len(config[CONF_SERVICE_UUID]) == len(esp32_ble_tracker.bt_uuid32_format):
-        cg.add(
-            var.set_service_uuid32(esp32_ble_tracker.as_hex(config[CONF_SERVICE_UUID]))
-        )
-    elif len(config[CONF_SERVICE_UUID]) == len(esp32_ble_tracker.bt_uuid128_format):
-        uuid128 = esp32_ble_tracker.as_reversed_hex_array(config[CONF_SERVICE_UUID])
+    if len(config[CONF_SERVICE_UUID]) == len(bt_uuid16_format):
+        cg.add(var.set_service_uuid16(as_hex(config[CONF_SERVICE_UUID])))
+    elif len(config[CONF_SERVICE_UUID]) == len(bt_uuid32_format):
+        cg.add(var.set_service_uuid32(as_hex(config[CONF_SERVICE_UUID])))
+    elif len(config[CONF_SERVICE_UUID]) == len(bt_uuid128_format):
+        uuid128 = as_reversed_hex_array(config[CONF_SERVICE_UUID])
         cg.add(var.set_service_uuid128(uuid128))
 
-    if len(config[CONF_CHARACTERISTIC_UUID]) == len(esp32_ble_tracker.bt_uuid16_format):
-        cg.add(
-            var.set_char_uuid16(
-                esp32_ble_tracker.as_hex(config[CONF_CHARACTERISTIC_UUID])
-            )
-        )
-    elif len(config[CONF_CHARACTERISTIC_UUID]) == len(
-        esp32_ble_tracker.bt_uuid32_format
-    ):
-        cg.add(
-            var.set_char_uuid32(
-                esp32_ble_tracker.as_hex(config[CONF_CHARACTERISTIC_UUID])
-            )
-        )
-    elif len(config[CONF_CHARACTERISTIC_UUID]) == len(
-        esp32_ble_tracker.bt_uuid128_format
-    ):
-        uuid128 = esp32_ble_tracker.as_reversed_hex_array(
-            config[CONF_CHARACTERISTIC_UUID]
-        )
+    if len(config[CONF_CHARACTERISTIC_UUID]) == len(bt_uuid16_format):
+        cg.add(var.set_char_uuid16(as_hex(config[CONF_CHARACTERISTIC_UUID])))
+    elif len(config[CONF_CHARACTERISTIC_UUID]) == len(bt_uuid32_format):
+        cg.add(var.set_char_uuid32(as_hex(config[CONF_CHARACTERISTIC_UUID])))
+    elif len(config[CONF_CHARACTERISTIC_UUID]) == len(bt_uuid128_format):
+        uuid128 = as_reversed_hex_array(config[CONF_CHARACTERISTIC_UUID])
         cg.add(var.set_char_uuid128(uuid128))
 
     return var
@@ -339,14 +526,45 @@ async def remove_bond_to_code(config, action_id, template_arg, args):
     return cg.new_Pvariable(action_id, template_arg, parent)
 
 
-async def to_code(config):
+async def _to_code_esp32(config: ConfigType) -> cg.MockObj:
+    from esphome.components import esp32_ble, esp32_ble_tracker
+    from esphome.components.esp32_ble import BTLoggers
+
     # Register the loggers this component needs
     esp32_ble.register_bt_logger(BTLoggers.GATT, BTLoggers.SMP)
     cg.add_define("USE_ESP32_BLE_UUID")
+    cg.add_define("USE_BLE_CLIENT_LEGACY_ENGINE")
 
     var = cg.new_Pvariable(config[CONF_ID])
     await cg.register_component(var, config)
     await esp32_ble_tracker.register_client(var, config)
+    return var
+
+
+# Sizes the neutral client's node storage; the client itself requests a
+# baseline slot so the define exists on every build that compiles the engine.
+_request_node_slot = cg.slot_counter("ESPHOME_BLE_CLIENT_MAX_NODES")
+
+
+async def _to_code_gatt(config: ConfigType) -> cg.MockObj:
+    # The engine always carries the node surface (the client itself owns the
+    # baseline slot).
+    _request_gatt_node_build()
+    backend = await bluetooth_connection.new_gatt_backend(config)
+    var = cg.new_Pvariable(config[CONF_ID])
+    await cg.register_component(var, config)
+    cg.add(var.set_backend(backend))
+    # Sighting-gated connects: the client listens for the peer's parsed
+    # advertisements through the hub.
+    await ble_device_base.register_ble_device(var, config)
+    return var
+
+
+async def to_code(config: ConfigType) -> None:
+    if _legacy_engine():
+        var = await _to_code_esp32(config)
+    else:
+        var = await _to_code_gatt(config)
     cg.add(var.set_address(config[CONF_MAC_ADDRESS].as_hex))
     cg.add(var.set_auto_connect(config[CONF_AUTO_CONNECT]))
     for conf in config.get(CONF_ON_CONNECT, []):

esphome/components/ble_client/automation.cpp

@@ -1,11 +0,0 @@
-#ifdef USE_ESP32
-
-#include "automation.h"
-
-namespace esphome::ble_client {
-
-const char *const Automation::TAG = "ble_client.automation";
-
-}  // namespace esphome::ble_client
-
-#endif

esphome/components/ble_client/automation.h

@@ -1,27 +1,14 @@
 #pragma once
 
-#ifdef USE_ESP32
+#include "esphome/core/defines.h"
 
-#include <utility>
-#include <vector>
+#ifdef USE_BLE_CLIENT_LEGACY_ENGINE
 
 #include "esphome/core/automation.h"
 #include "esphome/components/ble_client/ble_client.h"
-#include "esphome/core/helpers.h"
-#include "esphome/core/log.h"
-
-// Maximum bytes to log in hex format for BLE writes (many logging buffers are 256 chars)
-static constexpr size_t BLE_WRITE_MAX_LOG_BYTES = 64;
 
 namespace esphome::ble_client {
 
-// placeholder class for static TAG .
-class Automation {
- public:
-  // could be made inline with C++17
-  static const char *const TAG;
-};
-
 // implement on_connect automation.
 class BLEClientConnectTrigger final : public Trigger<>, public BLEClientNode {
  public:
@@ -93,144 +80,6 @@ class BLEClientNumericComparisonRequestTrigger final : public Trigger<uint32_t>,
   }
 };
 
-// implement the ble_client.ble_write action.
-template<typename... Ts> class BLEClientWriteAction final : public Action<Ts...>, public BLEClientNode {
- public:
-  BLEClientWriteAction(BLEClient *ble_client) {
-    ble_client->register_ble_node(this);
-    ble_client_ = ble_client;
-  }
-
-  void set_service_uuid16(uint16_t uuid) { this->service_uuid_ = espbt::ESPBTUUID::from_uint16(uuid); }
-  void set_service_uuid32(uint32_t uuid) { this->service_uuid_ = espbt::ESPBTUUID::from_uint32(uuid); }
-  void set_service_uuid128(uint8_t *uuid) { this->service_uuid_ = espbt::ESPBTUUID::from_raw(uuid); }
-
-  void set_char_uuid16(uint16_t uuid) { this->char_uuid_ = espbt::ESPBTUUID::from_uint16(uuid); }
-  void set_char_uuid32(uint32_t uuid) { this->char_uuid_ = espbt::ESPBTUUID::from_uint32(uuid); }
-  void set_char_uuid128(uint8_t *uuid) { this->char_uuid_ = espbt::ESPBTUUID::from_raw(uuid); }
-
-  void set_value_template(std::vector<uint8_t> (*func)(Ts...)) {
-    this->value_.func = func;
-    this->len_ = -1;  // Sentinel value indicates template mode
-  }
-
-  // Store pointer to static data in flash (no RAM copy)
-  void set_value_simple(const uint8_t *data, size_t len) {
-    this->value_.data = data;
-    this->len_ = len;  // Length >= 0 indicates static mode
-  }
-
-  void play(const Ts &...x) override {}
-
-  void play_complex(const Ts &...x) override {
-    this->num_running_++;
-    this->var_ = std::make_tuple(x...);
-
-    bool result;
-    if (this->len_ >= 0) {
-      // Static mode: write directly from flash pointer
-      result = this->write(this->value_.data, this->len_);
-    } else {
-      // Template mode: call function and write the vector
-      std::vector<uint8_t> value = this->value_.func(x...);
-      result = this->write(value);
-    }
-
-    // on write failure, continue the automation chain rather than stopping so that e.g. disconnect can work.
-    if (!result)
-      this->play_next_(x...);
-  }
-
-  /**
-   * Note about logging: the esph_log_X macros are used here because the CI checks complain about use of the ESP LOG
-   * macros in header files (Can't even write it in a comment!)
-   * Not sure why, because they seem to work just fine.
-   * The problem is that the implementation of a templated class can't be placed in a .cpp file when using C++ less than
-   * 17, so the methods have to be here.  The esph_log_X macros are equivalent in function, but don't trigger the CI
-   * errors.
-   */
-  // initiate the write. Return true if all went well, will be followed by a WRITE_CHAR event.
-  bool write(const uint8_t *data, size_t len) {
-    if (this->node_state != espbt::ClientState::ESTABLISHED) {
-      esph_log_w(Automation::TAG, "Cannot write to BLE characteristic - not connected");
-      return false;
-    }
-#if ESPHOME_LOG_LEVEL >= ESPHOME_LOG_LEVEL_VERY_VERBOSE
-    char hex_buf[format_hex_pretty_size(BLE_WRITE_MAX_LOG_BYTES)];
-    esph_log_vv(Automation::TAG, "Will write %d bytes: %s", len, format_hex_pretty_to(hex_buf, data, len));
-#endif
-    esp_err_t err =
-        esp_ble_gattc_write_char(this->parent()->get_gattc_if(), this->parent()->get_conn_id(), this->char_handle_, len,
-                                 const_cast<uint8_t *>(data), this->write_type_, ESP_GATT_AUTH_REQ_NONE);
-    if (err != ESP_OK) {
-      esph_log_e(Automation::TAG, "Error writing to characteristic: %s!", esp_err_to_name(err));
-      return false;
-    }
-    return true;
-  }
-
-  bool write(const std::vector<uint8_t> &value) { return this->write(value.data(), value.size()); }
-
-  void gattc_event_handler(esp_gattc_cb_event_t event, esp_gatt_if_t gattc_if,
-                           esp_ble_gattc_cb_param_t *param) override {
-    switch (event) {
-      case ESP_GATTC_WRITE_CHAR_EVT:
-        // upstream code checked the MAC address, verify the characteristic.
-        if (param->write.handle == this->char_handle_)
-          this->parent()->run_later([this]() { this->play_next_tuple_(this->var_); });
-        break;
-      case ESP_GATTC_DISCONNECT_EVT:
-        if (this->num_running_ != 0)
-          this->stop_complex();
-        break;
-      case ESP_GATTC_SEARCH_CMPL_EVT: {
-        auto *chr = this->parent()->get_characteristic(this->service_uuid_, this->char_uuid_);
-        if (chr == nullptr) {
-          char char_buf[esp32_ble::UUID_STR_LEN];
-          char service_buf[esp32_ble::UUID_STR_LEN];
-          esph_log_w("ble_write_action", "Characteristic %s was not found in service %s",
-                     this->char_uuid_.to_str(char_buf), this->service_uuid_.to_str(service_buf));
-          break;
-        }
-        this->char_handle_ = chr->handle;
-        this->char_props_ = chr->properties;
-        if (this->char_props_ & ESP_GATT_CHAR_PROP_BIT_WRITE) {
-          this->write_type_ = ESP_GATT_WRITE_TYPE_RSP;
-          esph_log_d(Automation::TAG, "Write type: ESP_GATT_WRITE_TYPE_RSP");
-        } else if (this->char_props_ & ESP_GATT_CHAR_PROP_BIT_WRITE_NR) {
-          this->write_type_ = ESP_GATT_WRITE_TYPE_NO_RSP;
-          esph_log_d(Automation::TAG, "Write type: ESP_GATT_WRITE_TYPE_NO_RSP");
-        } else {
-          char char_buf[esp32_ble::UUID_STR_LEN];
-          esph_log_e(Automation::TAG, "Characteristic %s does not allow writing", this->char_uuid_.to_str(char_buf));
-          break;
-        }
-        this->node_state = espbt::ClientState::ESTABLISHED;
-        char char_buf[esp32_ble::UUID_STR_LEN];
-        esph_log_d(Automation::TAG, "Found characteristic %s on device %s", this->char_uuid_.to_str(char_buf),
-                   ble_client_->address_str());
-        break;
-      }
-      default:
-        break;
-    }
-  }
-
- private:
-  BLEClient *ble_client_;
-  ssize_t len_{-1};  // -1 = template mode, >=0 = static mode with length
-  union Value {
-    std::vector<uint8_t> (*func)(Ts...);  // Function pointer (stateless lambdas)
-    const uint8_t *data;                  // Pointer to static data in flash
-  } value_;
-  espbt::ESPBTUUID service_uuid_;
-  espbt::ESPBTUUID char_uuid_;
-  std::tuple<Ts...> var_{};
-  uint16_t char_handle_{};
-  esp_gatt_char_prop_t char_props_{};
-  esp_gatt_write_type_t write_type_{};
-};
-
 template<typename... Ts> class BLEClientPasskeyReplyAction final : public Action<Ts...> {
  public:
   BLEClientPasskeyReplyAction(BLEClient *ble_client) { parent_ = ble_client; }

esphome/components/ble_client/automation_gatt.h

@@ -0,0 +1,118 @@
+// Neutral twins of the shared ble_client automations. Class names, namespace,
+// and codegen-visible signatures are IDENTICAL to automation.h so generated
+// main.cpp compiles against whichever engine the build gates in; only the
+// internals differ (client callbacks and the neutral node interface instead
+// of raw gattc events). The Bluedroid-security automations (passkey, numeric
+// comparison, remove bond) have no neutral equivalent and stay esp32-only.
+
+#pragma once
+
+#include "esphome/core/defines.h"
+
+#if defined(USE_BLE_GATT_CLIENT) && !defined(USE_BLE_CLIENT_LEGACY_ENGINE)
+
+#include <tuple>
+
+#include "ble_client_gatt.h"
+#include "esphome/core/automation.h"
+
+namespace esphome::ble_client {
+
+class BLEClientConnectTrigger final : public Trigger<> {
+ public:
+  explicit BLEClientConnectTrigger(BLEClient *parent) {
+    parent->add_on_connect_callback([this]() { this->trigger(); });
+  }
+};
+
+class BLEClientDisconnectTrigger final : public Trigger<> {
+ public:
+  explicit BLEClientDisconnectTrigger(BLEClient *parent) {
+    // Fires only after a completed connection (never for failed attempts),
+    // matching the legacy CLOSE_EVT semantics.
+    parent->add_on_disconnect_callback([this]() { this->trigger(); });
+  }
+};
+
+template<typename... Ts> class BLEClientConnectAction final : public Action<Ts...> {
+ public:
+  BLEClientConnectAction(BLEClient *ble_client) {
+    ble_client_ = ble_client;
+    ble_client->add_on_connect_callback([this]() {
+      if (this->num_running_ != 0)
+        this->play_next_tuple_(this->var_);
+    });
+    // A connect attempt that dies (or a later disconnect) terminates the
+    // chain, mirroring the legacy DISCONNECT_EVT handling.
+    ble_client->add_on_connect_failed_callback([this]() {
+      if (this->num_running_ != 0)
+        this->stop_complex();
+    });
+    ble_client->add_on_disconnect_callback([this]() {
+      if (this->num_running_ != 0)
+        this->stop_complex();
+    });
+  }
+
+  // not used since we override play_complex_
+  void play(const Ts &...x) override {}
+
+  void play_complex(const Ts &...x) override {
+    // it makes no sense to have multiple instances of this running at the
+    // same time; cancel a re-trigger while still running.
+    if (this->num_running_ != 0) {
+      this->stop_complex();
+      return;
+    }
+    this->num_running_++;
+    if (this->ble_client_->connected()) {
+      this->play_next_(x...);
+    } else {
+      this->var_ = std::make_tuple(x...);
+      // No-op while already connecting; the callback resolves the wait.
+      this->ble_client_->connect();
+    }
+  }
+
+ private:
+  BLEClient *ble_client_;
+  std::tuple<Ts...> var_{};
+};
+
+template<typename... Ts> class BLEClientDisconnectAction final : public Action<Ts...> {
+ public:
+  BLEClientDisconnectAction(BLEClient *ble_client) {
+    ble_client_ = ble_client;
+    // Both terminal outcomes resolve the wait: a completed teardown and a
+    // connect attempt that died on the way down.
+    ble_client->add_on_disconnect_callback([this]() {
+      if (this->num_running_ != 0)
+        this->play_next_tuple_(this->var_);
+    });
+    ble_client->add_on_connect_failed_callback([this]() {
+      if (this->num_running_ != 0)
+        this->play_next_tuple_(this->var_);
+    });
+  }
+
+  // not used since we override play_complex_
+  void play(const Ts &...x) override {}
+
+  void play_complex(const Ts &...x) override {
+    this->num_running_++;
+    if (this->ble_client_->idle()) {
+      this->play_next_(x...);
+    } else {
+      this->var_ = std::make_tuple(x...);
+      this->ble_client_->disconnect();
+    }
+  }
+
+ private:
+  BLEClient *ble_client_;
+  std::tuple<Ts...> var_{};
+};
+
+}  // namespace esphome::ble_client
+
+#endif  // USE_BLE_GATT_CLIENT && !USE_BLE_CLIENT_LEGACY_ENGINE

esphome/components/ble_client/ble_client.cpp

@@ -2,10 +2,16 @@
 #include "esphome/components/esp32_ble_client/ble_client_base.h"
 #include "esphome/components/esp32_ble_tracker/esp32_ble_tracker.h"
 #include "esphome/core/application.h"
+#include "esphome/core/hal.h"
 #include "esphome/core/helpers.h"
 #include "esphome/core/log.h"
 
-#ifdef USE_ESP32
+#ifdef USE_BLE_CLIENT_LEGACY_ENGINE
+
+#ifdef USE_BLE_CLIENT_GATT_NODES
+#include "esphome/components/bluetooth_connection/bluetooth_connection.h"
+#include "esphome/components/bluetooth_connection/gatt_service_table_bluedroid.h"
+#endif
 
 namespace esphome::ble_client {
 
@@ -30,6 +36,10 @@ void BLEClient::dump_config() {
 bool BLEClient::parse_device(const espbt::ESPBTDevice &device) {
   if (!this->enabled)
     return false;
+#ifdef USE_BLE_CLIENT_GATT_NODES
+  if (device.address_uint64() == this->address_ && this->gatt_backoff_.holding_off())
+    return false;
+#endif
   return BLEClientBase::parse_device(device);
 }
 
@@ -40,24 +50,60 @@ void BLEClient::set_enabled(bool enabled) {
   if (!enabled) {
     ESP_LOGI(TAG, "[%s] Disabling BLE client.", this->address_str());
     this->disconnect();
+    return;
   }
+#ifdef USE_BLE_CLIENT_GATT_NODES
+  // A re-enable clears the backoff (neutral-engine parity).
+  this->gatt_backoff_.reset();
+#endif
 }
 
 bool BLEClient::gattc_event_handler(esp_gattc_cb_event_t event, esp_gatt_if_t esp_gattc_if,
                                     esp_ble_gattc_cb_param_t *param) {
+#ifdef USE_BLE_CLIENT_GATT_NODES
+  // Bridge-initiated registrations bypass the base's REG_FOR_NOTIFY handling:
+  // its automatic CCCD write would double the node's own.
+  // Handle-keyed: mixed legacy/neutral subscriptions to one characteristic
+  // are unsupported during the migration window.
+  if (event == ESP_GATTC_REG_FOR_NOTIFY_EVT && esp_gattc_if == this->gattc_if_ &&
+      this->take_pending_gatt_reg_(param->reg_for_notify.handle)) {
+    if (this->pending_notify_regs_ > 0)
+      this->pending_notify_regs_--;
+    int err = param->reg_for_notify.status == ESP_GATT_OK ? 0 : param->reg_for_notify.status;
+    this->notify_state_to_gatt_nodes_(param->reg_for_notify.handle, true, err);
+    // A retiring last registration must still release the cache.
+    this->maybe_release_services_();
+    return true;
+  }
+#endif
   if (!BLEClientBase::gattc_event_handler(event, esp_gattc_if, param))
     return false;
 
+#ifdef USE_BLE_CLIENT_GATT_NODES
+  // Before the legacy fan-out so gatt nodes resolve before any trigger fires.
+  if (!this->gatt_nodes_.empty()) {
+    if (event == ESP_GATTC_SEARCH_CMPL_EVT) {
+      // A failed discovery tears the link down; the on_connect trigger must
+      // not fire into the teardown.
+      if (!this->handle_gatt_search_cmpl_(param->search_cmpl.status))
+        return true;
+    } else {
+      this->dispatch_gatt_event_(event, param);
+    }
+  }
+#endif
   for (auto *node : this->nodes_)
     node->gattc_event_handler(event, esp_gattc_if, param);
+  this->maybe_release_services_();
+  return true;
+}
 
+void BLEClient::maybe_release_services_() {
   // The release frees the GATT cache that BLEClientBase's CCCD lookup still needs.
-  // The last REG_FOR_NOTIFY event clears the counter before node dispatch, so the release still runs here.
   if (!this->services_.empty() && !this->notify_registration_pending() && this->all_nodes_established_()) {
     this->release_services();
     ESP_LOGD(TAG, "All clients established, services released");
   }
-  return true;
 }
 
 void BLEClient::gap_event_handler(esp_gap_ble_cb_event_t event, esp_ble_gap_cb_param_t *param) {
@@ -65,10 +111,19 @@ void BLEClient::gap_event_handler(esp_gap_ble_cb_event_t event, esp_ble_gap_cb_p
 
   for (auto *node : this->nodes_)
     node->gap_event_handler(event, param);
+#ifdef USE_BLE_CLIENT_GATT_NODES
+  if (event == ESP_GAP_BLE_AUTH_CMPL_EVT && this->check_addr(param->ble_security.auth_cmpl.bd_addr)) {
+    int status = param->ble_security.auth_cmpl.success ? 0 : param->ble_security.auth_cmpl.fail_reason;
+    for (auto *node : this->gatt_nodes_)
+      node->on_pairing_result(status);
+  }
+#endif
 }
 
 void BLEClient::set_state(espbt::ClientState state) {
   BLEClientBase::set_state(state);
+  // ESTABLISHED never flows through here; gatt nodes are promoted after the
+  // on_connected fan-out.
   for (auto &node : nodes_)
     node->node_state = state;
 }
@@ -83,6 +138,218 @@ bool BLEClient::all_nodes_established_() {
   return true;
 }
 
+#ifdef USE_BLE_CLIENT_GATT_NODES
+
+void BLEClient::register_gatt_node(BLEClientNode *node) {
+  // Parent before the capacity check so a dropped node still has a usable
+  // parent() (neutral-engine parity).
+  node->set_ble_client_parent(this);
+  if (this->gatt_nodes_.size() == ESPHOME_BLE_CLIENT_MAX_NODES) {
+    // push_back past capacity is a silent no-op; an undersized slot count
+    // must be loud at boot, not an unresolvable node at runtime.
+    ESP_LOGE(TAG, "[%s] Node capacity exceeded; node dropped", this->address_str());
+    this->status_set_error(LOG_STR("node capacity exceeded"));
+    return;
+  }
+  this->gatt_nodes_.push_back(node);
+  // nodes_ covers the shared state bookkeeping; gatt_nodes_ is the neutral
+  // fan-out subset.
+  this->register_ble_node(node);
+}
+
+int BLEClient::find_pending_gatt_reg_(uint16_t handle) const {
+  for (uint8_t i = 0; i < this->pending_gatt_reg_count_; i++) {
+    if (this->pending_gatt_regs_[i] == handle)
+      return i;
+  }
+  return -1;
+}
+
+bool BLEClient::take_pending_gatt_reg_(uint16_t handle) {
+  int i = this->find_pending_gatt_reg_(handle);
+  if (i < 0)
+    return false;
+  // No duplicates (notify_characteristic refuses a re-push); swap-with-last.
+  this->pending_gatt_regs_[i] = this->pending_gatt_regs_[--this->pending_gatt_reg_count_];
+  return true;
+}
+
+void BLEClient::notify_state_to_gatt_nodes_(uint16_t handle, bool enabled, int error) {
+  if (error != 0) {
+    ESP_LOGW(TAG, "[%s] Notify %s on handle 0x%04x failed, status=%d", this->address_str(),
+             enabled ? "enable" : "disable", handle, error);
+  }
+  for (auto *node : this->gatt_nodes_)
+    node->on_notify_state(handle, enabled, error);
+}
+
+void BLEClient::dispatch_gatt_event_(esp_gattc_cb_event_t event, esp_ble_gattc_cb_param_t *param) {
+  switch (event) {
+    case ESP_GATTC_READ_CHAR_EVT:
+    case ESP_GATTC_READ_DESCR_EVT: {
+      bool ok = param->read.status == ESP_GATT_OK;
+      if (!ok) {
+        // Breadcrumb even when no node claims the handle.
+        ESP_LOGD(TAG, "[%s] Read on handle 0x%04x completed with status %d", this->address_str(), param->read.handle,
+                 param->read.status);
+      }
+      for (auto *node : this->gatt_nodes_) {
+        node->on_read_result(param->read.handle, ok ? param->read.value : nullptr, ok ? param->read.value_len : 0,
+                             ok ? 0 : param->read.status);
+      }
+      break;
+    }
+    case ESP_GATTC_WRITE_CHAR_EVT:
+    case ESP_GATTC_WRITE_DESCR_EVT:
+      if (param->write.status != ESP_GATT_OK) {
+        // Breadcrumb even when no node claims the handle.
+        ESP_LOGD(TAG, "[%s] Write on handle 0x%04x completed with status %d", this->address_str(), param->write.handle,
+                 param->write.status);
+      }
+      for (auto *node : this->gatt_nodes_) {
+        node->on_write_result(param->write.handle, param->write.status == ESP_GATT_OK ? 0 : param->write.status);
+      }
+      break;
+    case ESP_GATTC_NOTIFY_EVT:
+      for (auto *node : this->gatt_nodes_) {
+        node->on_notify(param->notify.handle, param->notify.value, param->notify.value_len);
+      }
+      break;
+    case ESP_GATTC_UNREG_FOR_NOTIFY_EVT:
+      // The base does no CCCD work for unregister; no interception needed.
+      this->notify_state_to_gatt_nodes_(
+          param->unreg_for_notify.handle, false,
+          param->unreg_for_notify.status == ESP_GATT_OK ? 0 : param->unreg_for_notify.status);
+      break;
+    default:
+      break;
+  }
+}
+
+bool BLEClient::handle_gatt_search_cmpl_(esp_gatt_status_t status) {
+  // The base ignores the search status; the neutral contract must not.
+  uint16_t service_total = 0;
+  bool counted = status == ESP_GATT_OK && bluetooth_connection::BluedroidServiceTable::count_services(
+                                              this->gattc_if_, this->conn_id_, &service_total);
+  if (!counted || service_total == 0) {
+    // A failed search poisons the whole discovery, legacy nodes included.
+    ESP_LOGW(TAG, "[%s] Discovery failed (status=%d, services=%u)", this->address_str(), status, service_total);
+    this->gatt_backoff_.register_failure(this->address_str());
+    this->disconnect();
+    return false;
+  }
+  // Stack-owned; nodes copy their handles during on_connected().
+  bluetooth_connection::BluedroidServiceTable table;
+  if (!table.build(this->gattc_if_, this->conn_id_, service_total, this->connection_index_)) {
+    if (!this->has_legacy_nodes_()) {
+      ESP_LOGW(TAG, "[%s] Service table build failed; treating as failed discovery", this->address_str());
+      this->gatt_backoff_.register_failure(this->address_str());
+      this->disconnect();
+      return false;
+    }
+    // Only the table build failed; legacy nodes read the base's services_
+    // and keep the link. Gatt nodes catch the next connection.
+    ESP_LOGW(TAG, "[%s] Service table build failed; gatt nodes skip this connection", this->address_str());
+    this->status_set_warning(LOG_STR("gatt nodes inactive: service table build failed"));
+  } else {
+    this->gatt_connected_ = true;
+    auto view = table.view();
+    for (auto *node : this->gatt_nodes_) {
+      node->on_connected(view);
+      if (this->state() != espbt::ClientState::ESTABLISHED) {
+        // The node tore the link down; remaining nodes get on_disconnected
+        // with no preceding on_connected, so leave a trace of why.
+        ESP_LOGW(TAG, "[%s] A node aborted the connection during setup", this->address_str());
+        return false;
+      }
+    }
+    this->gatt_backoff_.reset();
+    this->status_clear_warning();
+  }
+  // Promote so the legacy release condition can fire.
+  for (auto *node : this->gatt_nodes_)
+    node->node_state = espbt::ClientState::ESTABLISHED;
+  return true;
+}
+
+void BLEClient::on_disconnect_complete(esp_err_t reason) {
+  this->pending_gatt_reg_count_ = 0;
+  if (!this->gatt_connected_)
+    return;  // Never-established links report nothing (neutral parity).
+  this->gatt_connected_ = false;
+  for (auto *node : this->gatt_nodes_)
+    node->on_disconnected();
+}
+
+int BLEClient::check_and_log_error_(const char *operation, esp_err_t err) {
+  if (err != ESP_OK)
+    this->log_gattc_warning_(operation, err);
+  return err;
+}
+
+int BLEClient::write_characteristic(uint16_t handle, const uint8_t *data, uint16_t len, bool response) {
+  if (this->conn_id_ == UNSET_CONN_ID)
+    return ble_device_base::GATT_ERR_NOT_CONNECTED;
+  return this->check_and_log_error_(
+      "esp_ble_gattc_write_char",
+      esp_ble_gattc_write_char(this->gattc_if_, this->conn_id_, handle, len, const_cast<uint8_t *>(data),
+                               response ? ESP_GATT_WRITE_TYPE_RSP : ESP_GATT_WRITE_TYPE_NO_RSP,
+                               ESP_GATT_AUTH_REQ_NONE));
+}
+
+int BLEClient::read_characteristic(uint16_t handle) {
+  if (this->conn_id_ == UNSET_CONN_ID)
+    return ble_device_base::GATT_ERR_NOT_CONNECTED;
+  return this->check_and_log_error_("esp_ble_gattc_read_char", esp_ble_gattc_read_char(this->gattc_if_, this->conn_id_,
+                                                                                       handle, ESP_GATT_AUTH_REQ_NONE));
+}
+
+int BLEClient::read_descriptor(uint16_t handle) {
+  if (this->conn_id_ == UNSET_CONN_ID)
+    return ble_device_base::GATT_ERR_NOT_CONNECTED;
+  return this->check_and_log_error_(
+      "esp_ble_gattc_read_char_descr",
+      esp_ble_gattc_read_char_descr(this->gattc_if_, this->conn_id_, handle, ESP_GATT_AUTH_REQ_NONE));
+}
+
+int BLEClient::write_descriptor(uint16_t handle, const uint8_t *data, uint16_t len) {
+  if (this->conn_id_ == UNSET_CONN_ID)
+    return ble_device_base::GATT_ERR_NOT_CONNECTED;
+  return this->check_and_log_error_(
+      "esp_ble_gattc_write_char_descr",
+      esp_ble_gattc_write_char_descr(this->gattc_if_, this->conn_id_, handle, len, const_cast<uint8_t *>(data),
+                                     ESP_GATT_WRITE_TYPE_RSP, ESP_GATT_AUTH_REQ_NONE));
+}
+
+int BLEClient::notify_characteristic(uint16_t handle, bool enable) {
+  if (this->conn_id_ == UNSET_CONN_ID)
+    return ble_device_base::GATT_ERR_NOT_CONNECTED;
+  if (enable) {
+    if (this->find_pending_gatt_reg_(handle) >= 0) {
+      // ESP_OK: the in-flight registration's completion fans out to all nodes.
+      ESP_LOGW(TAG, "[%s] Notify registration already pending for handle 0x%04x", this->address_str(), handle);
+      return ESP_OK;
+    }
+    if (this->pending_gatt_reg_count_ == MAX_PENDING_NOTIFY_REGS) {
+      // An untracked registration would let the base's auto-CCCD through.
+      ESP_LOGE(TAG, "[%s] Too many pending notify registrations", this->address_str());
+      return ble_device_base::GATT_ERR_NO_MEMORY;
+    }
+    // The base helper's pending count holds the service-release until the
+    // (intercepted) completion.
+    esp_err_t err = this->register_for_notify(handle);
+    if (err == ESP_OK)
+      this->pending_gatt_regs_[this->pending_gatt_reg_count_++] = handle;
+    return this->check_and_log_error_("esp_ble_gattc_register_for_notify", err);
+  }
+  return this->check_and_log_error_("esp_ble_gattc_unregister_for_notify",
+                                    esp_ble_gattc_unregister_for_notify(this->gattc_if_, this->remote_bda_, handle));
+}
+
+int BLEClient::unpair() { return bluetooth_connection::unpair_device(this->get_address()); }
+
+#endif  // USE_BLE_CLIENT_GATT_NODES
+
 }  // namespace esphome::ble_client
 
 #endif

esphome/components/ble_client/ble_client.h

@@ -1,18 +1,20 @@
 #pragma once
 
+#include "esphome/core/defines.h"
+
+#ifdef USE_BLE_CLIENT_LEGACY_ENGINE
+
+#include "ble_client_node.h"
+#include "connect_backoff.h"
 #include "esphome/components/esp32_ble_client/ble_client_base.h"
 #include "esphome/components/esp32_ble_tracker/esp32_ble_tracker.h"
 #include "esphome/core/component.h"
 #include "esphome/core/helpers.h"
 
-#ifdef USE_ESP32
-
 #include <esp_bt_defs.h>
 #include <esp_gap_ble_api.h>
 #include <esp_gatt_common_api.h>
 #include <esp_gattc_api.h>
-#include <array>
-#include <string>
 #include <vector>
 
 namespace esphome::ble_client {
@@ -21,34 +23,6 @@ namespace espbt = esphome::esp32_ble_tracker;
 
 using namespace esp32_ble_client;
 
-class BLEClient;
-
-class BLEClientNode {
- public:
-  virtual void gattc_event_handler(esp_gattc_cb_event_t event, esp_gatt_if_t gattc_if,
-                                   esp_ble_gattc_cb_param_t *param){};
-  virtual void gap_event_handler(esp_gap_ble_cb_event_t event, esp_ble_gap_cb_param_t *param) {}
-  virtual void loop() {}
-  void set_address(uint64_t address) { address_ = address; }
-  espbt::ESPBTClient *client;
-  // This should be transitioned to Established once the node no longer needs
-  // the services/descriptors/characteristics of the parent client. This will
-  // allow some memory to be freed.
-  // The parent frees the peer's GATT cache once every node reports Established.
-  // Never report Established while an operation that reads that cache is outstanding.
-  // - esp_ble_gattc_register_for_notify() completes asynchronously.
-  // - Register from ESP_GATTC_SEARCH_CMPL_EVT, then set this from ESP_GATTC_REG_FOR_NOTIFY_EVT.
-  // - BLEClientBase::register_for_notify() holds the release until the registration completes.
-  espbt::ClientState node_state;
-
-  BLEClient *parent() { return this->parent_; }
-  void set_ble_client_parent(BLEClient *parent) { this->parent_ = parent; }
-
- protected:
-  BLEClient *parent_;
-  uint64_t address_;
-};
-
 class BLEClient final : public BLEClientBase {
  public:
   void setup() override;
@@ -64,7 +38,6 @@ class BLEClient final : public BLEClientBase {
   void set_enabled(bool enabled);
 
   void register_ble_node(BLEClientNode *node) {
-    node->client = this;
     node->set_ble_client_parent(this);
     this->nodes_.push_back(node);
   }
@@ -73,10 +46,57 @@ class BLEClient final : public BLEClientBase {
 
   void set_state(espbt::ClientState state) override;
 
+#ifdef USE_BLE_CLIENT_GATT_NODES
+  // ---- the neutral node surface (signatures shared with the non-esp32
+  // engine, so nodes on the neutral interface compile against either) ----
+  void register_gatt_node(BLEClientNode *node);
+
+  bool idle() const { return this->state() == espbt::ClientState::IDLE; }
+
+  int write_characteristic(uint16_t handle, const uint8_t *data, uint16_t len, bool response);
+  int read_characteristic(uint16_t handle);
+  int read_descriptor(uint16_t handle);
+  int write_descriptor(uint16_t handle, const uint8_t *data, uint16_t len);
+  /// Local registration only; per the neutral contract the CCCD write is the
+  /// node's job (the legacy auto-CCCD is suppressed for these handles).
+  int notify_characteristic(uint16_t handle, bool enable);
+  // pair() comes from BLEClientBase, matching the neutral engine's.
+  int unpair();
+#endif
+
  protected:
   bool all_nodes_established_();
+  void maybe_release_services_();
+#ifdef USE_BLE_CLIENT_GATT_NODES
+  int check_and_log_error_(const char *operation, esp_err_t err);
+  int find_pending_gatt_reg_(uint16_t handle) const;
+  void notify_state_to_gatt_nodes_(uint16_t handle, bool enabled, int error);
+  void dispatch_gatt_event_(esp_gattc_cb_event_t event, esp_ble_gattc_cb_param_t *param);
+  // False = failed discovery: the link comes down and the caller suppresses
+  // the legacy fan-out.
+  bool handle_gatt_search_cmpl_(esp_gatt_status_t status);
+  bool take_pending_gatt_reg_(uint16_t handle);
+  void on_disconnect_complete(esp_err_t reason) override;
+#endif
 
   std::vector<BLEClientNode *> nodes_;
+#ifdef USE_BLE_CLIENT_GATT_NODES
+  // Raise if a migrated node needs more concurrent registrations.
+  static constexpr uint8_t MAX_PENDING_NOTIFY_REGS = 4;
+
+  // Nodes on the neutral surface; fed the translated callbacks and
+  // auto-established after the on_connected fan-out. Every gatt node is
+  // also in nodes_ (registration pushes into both).
+  StaticVector<BLEClientNode *, ESPHOME_BLE_CLIENT_MAX_NODES> gatt_nodes_;
+  bool has_legacy_nodes_() const { return this->nodes_.size() > this->gatt_nodes_.size(); }
+  // Reconnect backoff after materializer failures.
+  ConnectBackoff gatt_backoff_;
+  // Bridge-initiated notify registrations awaiting REG_FOR_NOTIFY_EVT.
+  uint16_t pending_gatt_regs_[MAX_PENDING_NOTIFY_REGS];
+  uint8_t pending_gatt_reg_count_{0};
+  // on_connected fan-out started; on_disconnected is owed at teardown.
+  bool gatt_connected_{false};
+#endif
 };
 
 }  // namespace esphome::ble_client

esphome/components/ble_client/ble_client_gatt.cpp

@@ -0,0 +1,249 @@
+#include "ble_client_gatt.h"
+
+#if defined(USE_BLE_GATT_CLIENT) && !defined(USE_BLE_CLIENT_LEGACY_ENGINE)
+
+#include "esphome/core/hal.h"
+#include "esphome/core/log.h"
+
+namespace esphome::ble_client {
+
+static const char *const TAG = "ble_client";
+
+void BLEClient::register_ble_node(BLEClientNode *node) {
+  node->set_ble_client_parent(this);
+  if (this->nodes_.size() == ESPHOME_BLE_CLIENT_MAX_NODES) {
+    // push_back past capacity is a silent no-op; an undersized slot count
+    // must be loud at boot, not an unresolvable node at runtime.
+    ESP_LOGE(TAG, "[%s] Node capacity exceeded; node dropped", this->address_str_);
+    this->status_set_error(LOG_STR("node capacity exceeded"));
+    return;
+  }
+  this->nodes_.push_back(node);
+}
+
+void BLEClient::set_address(uint64_t address) {
+  this->address_ = address;
+  uint8_t mac[6];
+  ble_device_base::uint64_to_mac_msb_first(address, mac);
+  format_mac_addr_upper(mac, this->address_str_);
+}
+
+void BLEClient::set_enabled(bool enabled) {
+  if (enabled == this->enabled)
+    return;
+  ESP_LOGI(TAG, "[%s] %s", this->address_str_, enabled ? "Enabled" : "Disabled");
+  this->enabled = enabled;
+  if (!enabled) {
+    this->disconnect();
+    return;
+  }
+  // A re-enable clears the backoff; the next sighting connects (legacy
+  // parity: enabling does not itself connect).
+  this->backoff_.reset();
+}
+
+bool BLEClient::parse_device(const ble_device_base::ESPBTDevice &device) {
+  if (device.address_uint64() != this->address_)
+    return false;
+  // The sighting is the source of truth for the address type.
+  this->address_type_ = device.get_address_type();
+  this->address_type_known_ = true;
+  if (!this->enabled || !this->auto_connect_ || this->state_ != State::IDLE)
+    return true;
+  if (this->backoff_.holding_off())
+    return true;
+  this->attempt_connect_();
+  return true;
+}
+
+void BLEClient::connect() {
+  if (this->state_ != State::IDLE) {
+    ESP_LOGD(TAG, "[%s] Connect requested while busy, ignoring", this->address_str_);
+    return;
+  }
+  // An absent peer can inhibit scanning for the backend's full connect
+  // timeout, so this is worth a breadcrumb - but it is a supported action.
+  ESP_LOGI(TAG, "[%s] Connecting on request", this->address_str_);
+  if (!this->address_type_known_) {
+    // Legacy parity: without a sighting the address type defaults to
+    // public, which never matches a random-static peer.
+    ESP_LOGW(TAG, "[%s] No sighting yet; assuming a public address type", this->address_str_);
+  }
+  this->attempt_connect_();
+}
+
+void BLEClient::attempt_connect_() {
+  int err = this->backend_->connect(this->address_, this->address_type_);
+  if (err != 0) {
+    // A refused connect never produces a callback: stay idle, charge the
+    // backoff, and resolve any waiting connect action through the failure
+    // path so its chain terminates.
+    ESP_LOGW(TAG, "[%s] Connect refused, err=%d", this->address_str_, err);
+    this->backoff_.register_failure(this->address_str_);
+    this->defer([this]() { this->connect_failed_callbacks_.call(); });
+    return;
+  }
+  ESP_LOGD(TAG, "[%s] Connecting", this->address_str_);
+  this->state_ = State::CONNECTING;
+}
+
+void BLEClient::disconnect() {
+  if (this->state_ == State::IDLE) {
+    ESP_LOGD(TAG, "[%s] Disconnect requested while idle, ignoring", this->address_str_);
+    return;
+  }
+  // A deliberate teardown's failure report must not feed the backoff.
+  this->cancel_requested_ = true;
+  int err = this->backend_->gatt_disconnect();
+  if (err != 0) {
+    // Refused synchronously: backend and client disagree about the link
+    // state. Warn, then settle through the deliberate-cancel path.
+    ESP_LOGW(TAG, "[%s] Disconnect refused, err=%d; settling locally", this->address_str_, err);
+    this->on_connection_state(false, 0, err);
+  }
+}
+
+void BLEClient::on_connection_state(bool connected, uint16_t mtu, int error) {
+  if (connected) {
+    this->state_ = State::DISCOVERING;
+    int discover_err = this->backend_->discover_services();
+    if (discover_err != 0) {
+      // Synchronous refusal: no discovery completion will follow.
+      ESP_LOGW(TAG, "[%s] Service discovery refused, err=%d", this->address_str_, discover_err);
+      this->backoff_.register_failure(this->address_str_);
+      // Deliberate teardown: its report must not charge the backoff again.
+      this->disconnect();
+    }
+    return;
+  }
+  bool was_connected = this->state_ == State::CONNECTED;
+  bool cancelled = this->cancel_requested_;
+  this->cancel_requested_ = false;
+  this->state_ = State::IDLE;
+  if (was_connected) {
+    ESP_LOGI(TAG, "[%s] Disconnected, status=%d", this->address_str_, error);
+    for (auto *node : this->nodes_) {
+      node->on_disconnected();
+    }
+    // Continuations leave the backend's event-drain stack first.
+    this->defer([this]() { this->disconnect_callbacks_.call(); });
+  } else {
+    if (cancelled) {
+      // status carries the refusal code when the teardown settled
+      // synchronously; 0 on a backend-completed cancel.
+      ESP_LOGD(TAG, "[%s] Connect attempt cancelled, status=%d", this->address_str_, error);
+    } else {
+      ESP_LOGW(TAG, "[%s] Connect failed, status=%d", this->address_str_, error);
+      this->backoff_.register_failure(this->address_str_);
+    }
+    this->defer([this]() { this->connect_failed_callbacks_.call(); });
+  }
+}
+
+void BLEClient::on_service_discovery_done(int error) {
+  if (error != 0) {
+    ESP_LOGW(TAG, "[%s] Service discovery failed, status=%d", this->address_str_, error);
+    this->backoff_.register_failure(this->address_str_);
+    // The teardown is deliberate: do not charge the backoff again for its
+    // connection report.
+    this->disconnect();
+    return;
+  }
+  ble_device_base::GattServiceTable table{};
+  if (!this->nodes_.empty()) {
+    // Materialize only when a node will read it: a client with no nodes
+    // would pay the build/free cycle on every (re)connect for nothing.
+    table = this->backend_->get_service_table();
+    if (table.service_count == 0) {
+      // A failed materialization is indistinguishable from a service-less
+      // peer, and a real GATT peer always exposes at least GAP/GATT: fail
+      // the discovery before CONNECTED so the teardown resolves through
+      // connect_failed, never a spurious on_disconnect.
+      ESP_LOGW(TAG, "[%s] Service table is empty; treating as failed discovery", this->address_str_);
+      this->backend_->release_services();
+      this->backoff_.register_failure(this->address_str_);
+      this->disconnect();
+      return;
+    }
+  }
+  // CONNECTED before the fan-out so nodes may consult connected() from
+  // their own on_connected().
+  this->state_ = State::CONNECTED;
+  for (auto *node : this->nodes_) {
+    node->on_connected(table);
+    if (this->state_ != State::CONNECTED || this->cancel_requested_) {
+      // A node tore the link down mid-fan-out: on_disconnect fires with no
+      // preceding on_connect, so leave a trace of why.
+      ESP_LOGW(TAG, "[%s] A node aborted the connection during setup", this->address_str_);
+      this->backend_->release_services();
+      return;
+    }
+  }
+  this->backend_->release_services();
+  this->backoff_.reset();
+  ESP_LOGI(TAG, "[%s] Connected", this->address_str_);
+  this->defer([this]() { this->connect_callbacks_.call(); });
+}
+
+void BLEClient::on_write_result(uint16_t handle, int error) {
+  if (error != 0) {
+    // Breadcrumb even when no node claims the handle.
+    ESP_LOGD(TAG, "[%s] Write on handle 0x%04x completed with status %d", this->address_str_, handle, error);
+  }
+  for (auto *node : this->nodes_) {
+    node->on_write_result(handle, error);
+  }
+}
+
+void BLEClient::on_read_result(uint16_t handle, const uint8_t *data, uint16_t len, int error) {
+  if (error != 0) {
+    // Breadcrumb even when no node claims the handle.
+    ESP_LOGD(TAG, "[%s] Read on handle 0x%04x completed with status %d", this->address_str_, handle, error);
+  }
+  for (auto *node : this->nodes_) {
+    node->on_read_result(handle, data, len, error);
+  }
+}
+
+void BLEClient::on_notify_data(uint16_t handle, const uint8_t *data, uint16_t len) {
+  // Every node sees every notification and filters by handle (legacy parity).
+  for (auto *node : this->nodes_) {
+    node->on_notify(handle, data, len);
+  }
+}
+
+void BLEClient::on_notify_state(uint16_t handle, bool enabled, int error) {
+  if (error != 0) {
+    ESP_LOGW(TAG, "[%s] Notify %s on handle 0x%04x failed, status=%d", this->address_str_,
+             enabled ? "enable" : "disable", handle, error);
+  }
+  for (auto *node : this->nodes_) {
+    node->on_notify_state(handle, enabled, error);
+  }
+}
+
+void BLEClient::on_pairing_result(int status) {
+  if (status != 0) {
+    ESP_LOGW(TAG, "[%s] Pairing failed, status=%d", this->address_str_, status);
+  } else {
+    ESP_LOGI(TAG, "[%s] Paired", this->address_str_);
+  }
+  for (auto *node : this->nodes_) {
+    node->on_pairing_result(status);
+  }
+}
+
+void BLEClient::dump_config() {
+  ESP_LOGCONFIG(TAG,
+                "BLE Client:\n"
+                "  Address: %s\n"
+                "  Auto connect: %s",
+                this->address_str_, YESNO(this->auto_connect_));
+  if (this->enabled && this->state_ == State::IDLE) {
+    ESP_LOGCONFIG(TAG, "  Waiting for an advertisement from the device");
+  }
+}
+
+}  // namespace esphome::ble_client
+
+#endif  // USE_BLE_GATT_CLIENT && !USE_BLE_CLIENT_LEGACY_ENGINE

esphome/components/ble_client/ble_client_gatt.h

@@ -0,0 +1,149 @@
+// Platform-neutral ble_client engine on the ble_device_base GATT contract.
+//
+// Compiled on every platform with a GATT backend except esp32, which keeps
+// the legacy BLEClientBase engine (ble_client.h) until its raw-gattc node
+// family migrates - the exclusive gates make the same class names resolve to
+// exactly one definition per build, so codegen is shared.
+//
+// Connects are sighting-gated like the legacy engine: the client is a parsed
+// advertisement listener, captures the peer's address type from the sighting,
+// and asks the backend to connect only when enabled and idle.
+
+#pragma once
+
+#include "esphome/core/defines.h"
+
+#if defined(USE_BLE_GATT_CLIENT) && !defined(USE_BLE_CLIENT_LEGACY_ENGINE)
+
+#include "ble_client_node.h"
+#include "connect_backoff.h"
+#include "esphome/components/ble_device_base/ble_device.h"
+#include "esphome/components/ble_device_base/ble_gatt_client.h"
+#include "esphome/components/bluetooth_connection/bluetooth_connection.h"
+#include "esphome/components/bluetooth_connection/bluetooth_connection_gatt_backend.h"
+#include "esphome/core/component.h"
+#include "esphome/core/helpers.h"
+
+#include <cstdint>
+#include <functional>
+
+namespace esphome::ble_client {
+
+class BLEClient : public Component,
+                  public ble_device_base::ESPBTDeviceListener,
+                  public ble_device_base::GattClientListener {
+ public:
+  void dump_config() override;
+
+  // Public field for legacy parity (the switch platform republishes it).
+  bool enabled{true};
+
+  void set_backend(ble_device_base::BLEGattConnection *backend) {
+    this->backend_ = backend;
+    backend->set_listener(this);
+  }
+  void set_address(uint64_t address);
+  void set_auto_connect(bool auto_connect) { this->auto_connect_ = auto_connect; }
+  void set_enabled(bool enabled);
+  const char *address_str() const { return this->address_str_; }
+
+  void register_ble_node(BLEClientNode *node);
+  // One registration spelling shared with the esp32 engine's bridge.
+  void register_gatt_node(BLEClientNode *node) { this->register_ble_node(node); }
+
+  bool connected() const { return this->state_ == State::CONNECTED; }
+  bool idle() const { return this->state_ == State::IDLE; }
+
+  /// Action-initiated connect (no sighting needed; uses the last captured
+  /// address type, public until a sighting arrives). No-op unless idle.
+  void connect();
+  void disconnect();
+
+  /// Legacy-named deferral used by the automation twins: neutral listener
+  /// callbacks run inside the backend's event drain, so automation chain
+  /// continuations must leave that stack first.
+  void run_later(std::function<void()> &&f) { this->defer(std::move(f)); }  // NOLINT
+
+  // Backend ops for nodes and actions - the frozen node-facing surface.
+  // Only write_characteristic has an in-tree caller; subscribing means
+  // notify_characteristic plus a CCCD write_descriptor (the caller's job
+  // per the contract).
+  int write_characteristic(uint16_t handle, const uint8_t *data, uint16_t len, bool response) {
+    return this->backend_->write_characteristic(handle, data, len, response);
+  }
+  int read_characteristic(uint16_t handle) { return this->backend_->read_characteristic(handle); }
+  int read_descriptor(uint16_t handle) { return this->backend_->read_descriptor(handle); }
+  int write_descriptor(uint16_t handle, const uint8_t *data, uint16_t len) {
+    return this->backend_->write_descriptor(handle, data, len);
+  }
+  int notify_characteristic(uint16_t handle, bool enable) {
+    return this->backend_->notify_characteristic(handle, enable);
+  }
+  int pair() { return this->backend_->pair(); }
+  int unpair() { return bluetooth_connection::unpair_device(this->address_); }
+
+  // Automation callback registration.
+  template<typename F> void add_on_connect_callback(F &&callback) {
+    this->connect_callbacks_.add(std::forward<F>(callback));
+  }
+  template<typename F> void add_on_disconnect_callback(F &&callback) {
+    this->disconnect_callbacks_.add(std::forward<F>(callback));
+  }
+  // Fired when a connect attempt dies before being established; the user
+  // on_disconnect trigger deliberately does NOT fire here (legacy parity).
+  template<typename F> void add_on_connect_failed_callback(F &&callback) {
+    this->connect_failed_callbacks_.add(std::forward<F>(callback));
+  }
+
+  // ---- ble_device_base::ESPBTDeviceListener ----
+  bool parse_device(const ble_device_base::ESPBTDevice &device) override;
+
+  // ---- ble_device_base::GattClientListener ----
+  void on_connection_state(bool connected, uint16_t mtu, int error) override;
+  void on_service_discovery_done(int error) override;
+  void on_read_result(uint16_t handle, const uint8_t *data, uint16_t len, int error) override;
+  void on_write_result(uint16_t handle, int error) override;
+  void on_notify_data(uint16_t handle, const uint8_t *data, uint16_t len) override;
+  void on_notify_state(uint16_t handle, bool enabled, int error) override;
+  void on_pairing_result(int status) override;
+
+ protected:
+  enum class State : uint8_t { IDLE, CONNECTING, DISCOVERING, CONNECTED };
+
+  void attempt_connect_();
+
+  // Group 1: pointers / containers
+  ble_device_base::BLEGattConnection *backend_{nullptr};
+  // Codegen-sized (ESPHOME_BLE_CLIENT_MAX_NODES); filled during setup.
+  StaticVector<BLEClientNode *, ESPHOME_BLE_CLIENT_MAX_NODES> nodes_;
+
+  // Group 2: 8-byte types
+  uint64_t address_{0};
+
+  // Group 3: callback managers (pointer-sized when empty)
+  LazyCallbackManager<void()> connect_callbacks_;
+  LazyCallbackManager<void()> disconnect_callbacks_;
+  LazyCallbackManager<void()> connect_failed_callbacks_;
+
+  // Group 4: 4-byte types
+  // Backoff so an undiscoverable database or a dead peer cannot produce a
+  // battery-draining connect loop.
+  ConnectBackoff backoff_;
+
+  // Group 5: arrays
+  char address_str_[MAC_ADDRESS_PRETTY_BUFFER_SIZE]{};
+
+  // Group 6: 1-byte types
+  State state_{State::IDLE};
+  uint8_t address_type_{0};  // BLE_ADDR_TYPE_*, captured from the sighting
+  // Distinguishes a captured public type from the never-sighted default.
+  bool address_type_known_{false};
+  bool auto_connect_{true};
+  // A user-initiated teardown in flight; its failure report is not a
+  // connect failure and must not feed the backoff.
+  bool cancel_requested_{false};
+};
+
+}  // namespace esphome::ble_client
+
+#endif  // USE_BLE_GATT_CLIENT && !USE_BLE_CLIENT_LEGACY_ENGINE

esphome/components/ble_client/ble_client_node.h

@@ -0,0 +1,68 @@
+// The single BLEClientNode both ble_client engines share. The neutral
+// callback surface is the one interface node components build on; the raw
+// esp32 surface below it remains for components that have not migrated yet.
+
+#pragma once
+
+#include "esphome/core/defines.h"
+
+#ifdef USE_BLE_GATT_CLIENT
+#include "esphome/components/ble_device_base/ble_gatt_client.h"
+#endif
+#ifdef USE_BLE_CLIENT_LEGACY_ENGINE
+#include "esphome/components/esp32_ble_tracker/esp32_ble_tracker.h"
+
+#include <esp_gap_ble_api.h>
+#include <esp_gattc_api.h>
+#endif
+
+#include <cstdint>
+
+namespace esphome::ble_client {
+
+class BLEClient;
+
+class BLEClientNode {
+ public:
+#ifdef USE_BLE_CLIENT_GATT_NODES
+  // Neutral surface, delivered by both engines. The table is borrowed: copy
+  // handles during on_connected(). All nodes see all completions; filter by
+  // handle.
+  // A node that disconnects from inside on_connected() aborts the fan-out;
+  // the user's on_disconnect may then fire without a preceding on_connect.
+  virtual void on_connected(const ble_device_base::GattServiceTable &table) {}
+  virtual void on_disconnected() {}
+  virtual void on_notify(uint16_t handle, const uint8_t *data, uint16_t len) {}
+  // One in-flight registration per handle; its completion fans out to every
+  // node, so a refused duplicate request still sees on_notify_state.
+  virtual void on_notify_state(uint16_t handle, bool enabled, int error) {}
+  virtual void on_read_result(uint16_t handle, const uint8_t *data, uint16_t len, int error) {}
+  virtual void on_write_result(uint16_t handle, int error) {}
+  virtual void on_pairing_result(int status) {}
+#endif
+#ifdef USE_BLE_CLIENT_LEGACY_ENGINE
+  // Legacy raw surface; components overriding these need the legacy engine
+  // until migrated to the neutral surface above.
+  virtual void gattc_event_handler(esp_gattc_cb_event_t event, esp_gatt_if_t gattc_if,
+                                   esp_ble_gattc_cb_param_t *param) {}
+  virtual void gap_event_handler(esp_gap_ble_cb_event_t event, esp_ble_gap_cb_param_t *param) {}
+  virtual void loop() {}
+  // This should be transitioned to Established once the node no longer needs
+  // the services/descriptors/characteristics of the parent client. This will
+  // allow some memory to be freed.
+  // The parent frees the peer's GATT cache once every node reports Established.
+  // Never report Established while an operation that reads that cache is outstanding.
+  // - esp_ble_gattc_register_for_notify() completes asynchronously.
+  // - Register from ESP_GATTC_SEARCH_CMPL_EVT, then set this from ESP_GATTC_REG_FOR_NOTIFY_EVT.
+  // - BLEClientBase::register_for_notify() holds the release until the registration completes.
+  esp32_ble_tracker::ClientState node_state;
+#endif
+
+  BLEClient *parent() const { return this->parent_; }
+  void set_ble_client_parent(BLEClient *parent) { this->parent_ = parent; }
+
+ protected:
+  BLEClient *parent_{nullptr};
+};
+
+}  // namespace esphome::ble_client

esphome/components/ble_client/ble_write_action.h

@@ -0,0 +1,168 @@
+// The ble_client.ble_write action: a node on the platform-neutral interface,
+// so one implementation serves both engines (the esp32 bridge and the
+// neutral engine).
+
+#pragma once
+
+#include "esphome/core/defines.h"
+
+#ifdef USE_BLE_CLIENT_GATT_NODES
+
+#include <tuple>
+#include <vector>
+
+// One of the two engine headers resolves per build.
+#include "ble_client.h"
+#include "ble_client_gatt.h"
+#include "ble_client_node.h"
+#include "esphome/core/automation.h"
+#include "esphome/core/helpers.h"
+#include "esphome/core/log.h"
+
+namespace esphome::ble_client {
+
+static const char *const BLE_WRITE_TAG = "ble_client.automation";
+
+// Maximum bytes to log in hex format for BLE writes (many logging buffers are 256 chars)
+static constexpr size_t BLE_WRITE_MAX_LOG_BYTES = 64;
+
+template<typename... Ts> class BLEClientWriteAction final : public Action<Ts...>, public BLEClientNode {
+ public:
+  BLEClientWriteAction(BLEClient *ble_client) {
+    ble_client->register_gatt_node(this);
+    ble_client_ = ble_client;
+  }
+
+  void set_service_uuid16(uint16_t uuid) { this->service_uuid_ = ble_device_base::ESPBTUUID::from_uint16(uuid); }
+  void set_service_uuid32(uint32_t uuid) { this->service_uuid_ = ble_device_base::ESPBTUUID::from_uint32(uuid); }
+  void set_service_uuid128(uint8_t *uuid) { this->service_uuid_ = ble_device_base::ESPBTUUID::from_raw(uuid); }
+
+  void set_char_uuid16(uint16_t uuid) { this->char_uuid_ = ble_device_base::ESPBTUUID::from_uint16(uuid); }
+  void set_char_uuid32(uint32_t uuid) { this->char_uuid_ = ble_device_base::ESPBTUUID::from_uint32(uuid); }
+  void set_char_uuid128(uint8_t *uuid) { this->char_uuid_ = ble_device_base::ESPBTUUID::from_raw(uuid); }
+
+  void set_value_template(std::vector<uint8_t> (*func)(Ts...)) {
+    this->value_.func = func;
+    this->len_ = -1;  // Sentinel value indicates template mode
+  }
+
+  // Store pointer to static data in flash (no RAM copy)
+  void set_value_simple(const uint8_t *data, size_t len) {
+    this->value_.data = data;
+    this->len_ = len;  // Length >= 0 indicates static mode
+  }
+
+  void play(const Ts &...x) override {}
+
+  void play_complex(const Ts &...x) override {
+    this->num_running_++;
+    this->var_ = std::make_tuple(x...);
+
+    bool result;
+    if (this->len_ >= 0) {
+      result = this->write(this->value_.data, this->len_);
+    } else {
+      std::vector<uint8_t> value = this->value_.func(x...);
+      result = this->write(value.data(), value.size());
+    }
+
+    // on write failure, continue the automation chain rather than stopping so
+    // that e.g. disconnect can work.
+    if (!result)
+      this->play_next_(x...);
+  }
+
+  // Initiate the write; the completion arrives in on_write_result. The
+  // response-less path can complete synchronously inside the call, so the
+  // handle is armed before the backend is touched.
+  bool write(const uint8_t *data, size_t len) {
+    if (!this->ble_client_->connected()) {
+      esph_log_w(BLE_WRITE_TAG, "Cannot write to BLE characteristic - not connected");
+      return false;
+    }
+    if (!this->resolved_) {
+      esph_log_w(BLE_WRITE_TAG, "Cannot write to BLE characteristic - characteristic was not resolved");
+      return false;
+    }
+#if ESPHOME_LOG_LEVEL >= ESPHOME_LOG_LEVEL_VERY_VERBOSE
+    char hex_buf[format_hex_pretty_size(BLE_WRITE_MAX_LOG_BYTES)];
+    esph_log_vv(BLE_WRITE_TAG, "Will write %d bytes: %s", len, format_hex_pretty_to(hex_buf, data, len));
+#endif
+    int err = this->ble_client_->write_characteristic(this->char_handle_, data, len, this->write_response_);
+    if (err != 0) {
+      esph_log_e(BLE_WRITE_TAG, "Error writing to characteristic: %d!", err);
+      return false;
+    }
+    return true;
+  }
+
+  void on_connected(const ble_device_base::GattServiceTable &table) override {
+    const auto *service = ble_device_base::find_service(table, this->service_uuid_);
+    const auto *chr =
+        service == nullptr ? nullptr : ble_device_base::find_characteristic(table, *service, this->char_uuid_);
+    if (chr == nullptr) {
+      char char_buf[ble_device_base::UUID_STR_LEN];
+      char service_buf[ble_device_base::UUID_STR_LEN];
+      esph_log_w(BLE_WRITE_TAG, "Characteristic %s was not found in service %s", this->char_uuid_.to_str(char_buf),
+                 this->service_uuid_.to_str(service_buf));
+      return;
+    }
+    if (chr->properties & ble_device_base::GATT_CHAR_PROP_WRITE) {
+      this->write_response_ = true;
+    } else if (chr->properties & ble_device_base::GATT_CHAR_PROP_WRITE_NO_RSP) {
+      this->write_response_ = false;
+    } else {
+      char char_buf[ble_device_base::UUID_STR_LEN];
+      esph_log_e(BLE_WRITE_TAG, "Characteristic %s does not allow writing", this->char_uuid_.to_str(char_buf));
+      return;
+    }
+    this->char_handle_ = chr->value_handle;
+    this->resolved_ = true;
+    char char_buf[ble_device_base::UUID_STR_LEN];
+    esph_log_d(BLE_WRITE_TAG, "Found characteristic %s on device %s", this->char_uuid_.to_str(char_buf),
+               this->ble_client_->address_str());
+  }
+
+  void on_disconnected() override {
+    this->resolved_ = false;
+    this->char_handle_ = 0;
+    if (this->num_running_ != 0)
+      this->stop_complex();
+  }
+
+  void on_write_result(uint16_t handle, int error) override {
+    if (this->num_running_ == 0) {
+      return;
+    }
+    if (!this->resolved_ || handle != this->char_handle_) {
+      // A parked chain waiting on a completion that never matches would
+      // otherwise stall silently until disconnect.
+      esph_log_d(BLE_WRITE_TAG, "Write result for handle 0x%04x ignored, waiting on 0x%04x", handle,
+                 this->char_handle_);
+      return;
+    }
+    if (error != 0) {
+      // Continue the chain (legacy parity) but leave a breadcrumb.
+      esph_log_w(BLE_WRITE_TAG, "Write completed with status %d", error);
+    }
+    this->ble_client_->run_later([this]() { this->play_next_tuple_(this->var_); });
+  }
+
+ private:
+  BLEClient *ble_client_;
+  ssize_t len_{-1};  // -1 = template mode, >=0 = static mode with length
+  union Value {
+    std::vector<uint8_t> (*func)(Ts...);  // Function pointer (stateless lambdas)
+    const uint8_t *data;                  // Pointer to static data in flash
+  } value_;
+  ble_device_base::ESPBTUUID service_uuid_;
+  ble_device_base::ESPBTUUID char_uuid_;
+  std::tuple<Ts...> var_{};
+  uint16_t char_handle_{};
+  bool write_response_{false};
+  bool resolved_{false};
+};
+
+}  // namespace esphome::ble_client
+
+#endif  // USE_BLE_CLIENT_GATT_NODES

esphome/components/ble_client/connect_backoff.h

@@ -0,0 +1,43 @@
+#pragma once
+
+#include "esphome/core/defines.h"
+
+#ifdef USE_BLE_GATT_CLIENT
+
+#include "esphome/core/hal.h"
+#include "esphome/core/log.h"
+
+#include <cstdint>
+
+namespace esphome::ble_client {
+
+/// Reconnect backoff after repeated connect/discovery failures, shared by
+/// both engines. 256 ms ticks in a uint16_t keep it 4 bytes; the ~4.7 h tick
+/// wrap can at worst reinstate one stale hold-off of a minute.
+class ConnectBackoff {
+ public:
+  bool holding_off() const {
+    return this->failures_ != 0 && static_cast<uint16_t>(now() - this->start_) < this->failures_ * STEP_TICKS;
+  }
+  void register_failure(const char *address_str) {
+    if (this->failures_ < MAX_STEPS)
+      this->failures_++;
+    this->start_ = now();
+    esph_log_w("ble_client", "[%s] Holding off reconnect for %u s", address_str, this->failures_ * 10u);
+  }
+  void reset() { this->failures_ = 0; }
+
+ private:
+  // ~10 s per consecutive failure, capped so a flapping peer retries within
+  // a minute at worst.
+  static constexpr uint16_t STEP_TICKS = 40;  // x 256 ms
+  static constexpr uint8_t MAX_STEPS = 6;
+  static uint16_t now() { return static_cast<uint16_t>(millis() >> 8); }
+
+  uint16_t start_{0};
+  uint8_t failures_{0};
+};
+
+}  // namespace esphome::ble_client
+
+#endif  // USE_BLE_GATT_CLIENT

esphome/components/ble_device_base/ble_gatt_client.cpp

@@ -0,0 +1,48 @@
+#include "ble_gatt_client.h"
+
+#ifdef USE_BLE_GATT_CLIENT
+
+#include "esphome/core/log.h"
+
+namespace esphome::ble_device_base {
+
+static const char *const TAG = "ble_gatt_client";
+
+const GattCharacteristic *find_characteristic(const GattServiceTable &table, const GattService &service,
+                                              const ESPBTUUID &uuid) {
+  // 32-bit range math: a corrupt first/count pair cannot wrap past the check.
+  uint32_t end = uint32_t(service.first_characteristic) + service.characteristic_count;
+  if (end > table.characteristic_count) {
+    ESP_LOGW(TAG, "characteristic range out of bounds");
+    return nullptr;
+  }
+  for (uint32_t i = service.first_characteristic; i < end; i++) {
+    if (table.characteristics[i].uuid == uuid)
+      return &table.characteristics[i];
+  }
+  return nullptr;
+}
+
+const GattDescriptor *find_descriptor(const GattServiceTable &table, const GattCharacteristic &characteristic,
+                                      const ESPBTUUID &uuid) {
+  uint32_t end = uint32_t(characteristic.first_descriptor) + characteristic.descriptor_count;
+  if (end > table.descriptor_count) {
+    // Corrupt range, not a missing descriptor.
+    ESP_LOGW(TAG, "descriptor range out of bounds");
+    return nullptr;
+  }
+  for (uint32_t i = characteristic.first_descriptor; i < end; i++) {
+    if (table.descriptors[i].uuid == uuid)
+      return &table.descriptors[i];
+  }
+  return nullptr;
+}
+
+uint16_t find_cccd(const GattServiceTable &table, const GattCharacteristic &characteristic) {
+  const GattDescriptor *desc = find_descriptor(table, characteristic, ESPBTUUID::from_uint16(CCCD_UUID));
+  return desc != nullptr ? desc->handle : 0;
+}
+
+}  // namespace esphome::ble_device_base
+
+#endif  // USE_BLE_GATT_CLIENT

esphome/components/ble_device_base/ble_gatt_client.h

@@ -11,13 +11,16 @@
 // interface. All listener calls are delivered on the ESPHome main loop;
 // borrowed data pointers are valid only for the duration of the call.
 //
-// Error domain (plain int, forwarded to the API without translation):
+// Error domain (plain int, forwarded to the API without translation, so the
+// values are wire-frozen - API clients interpret them):
 //   0            success
-//   1..0x11      ATT error codes (Bluetooth spec; BTstack and Bluedroid agree)
+//   1..0x11      ATT error codes (Bluetooth spec) - reserved; a backend whose
+//                native error codes land in this window must remap them out
 //   GATT_ERR_NOT_CONNECTED (-1)  no connection to the peer (on esp32 a raw
 //                ESP_FAIL from the stack shares this value; both read as a
 //                failed, unusable connection on the client side)
 //   GATT_ERR_NO_MEMORY (-2)      backend storage exhausted
+//   -1..-15      reserved for future contract sentinels
 //   anything else: platform stack error/status code, surfaced opaquely.
 // Connection events carry HCI status/disconnect reason codes (same code
 // space on every controller).
@@ -99,9 +102,18 @@ class GattClientListener {
 // The BLEGattConnection op surface, asserted where the alias binds
 // (bluetooth_connection_gatt_backend.h). Operations return 0 when accepted (completion arrives
 // through the listener) or a synchronous error (busy, not connected, stack
-// rejection); one operation may be outstanding at a time. Semantics beyond
-// the signatures:
+// rejection); one operation may be outstanding at a time. An accepted
+// operation's completion is delivered from the event loop, NEVER
+// synchronously from inside the op call - a synchronous terminal
+// on_connection_state from within gatt_disconnect() would re-enter the
+// consumer mid-teardown. Semantics beyond the signatures:
 // - connect: addr_type is a BLE_ADDR_TYPE_* constant (ble_device.h).
+//   Returning 0 means the request is accepted, not that the radio acted: the
+//   backend owns integration with its platform's scan/connect arbitration
+//   (Bluedroid parks the request for the tracker's promote loop, which owns
+//   scan-stop/coex/one-connect-at-a-time; the rp2 backend opens immediately
+//   and relies on sighting-gated consumers). Consumers must not assume
+//   connect timing.
 // - gatt_disconnect: also cancels a connect in progress (named to coexist
 //   with a platform stack's own void disconnect() on one backend class).
 //   Nonzero means nothing to tear down and no completion will follow; an
@@ -143,6 +155,41 @@ concept BLEGattConnectionContract = requires(T conn, GattClientListener *listene
   { conn.set_connection_type(ConnectionType{}) } -> std::same_as<void>;
 };
 
+// ---- service table lookup helpers ----
+//
+// Neutral, bounds-checked walks over a materialized GattServiceTable for
+// direct consumers that resolve a known device's handles by UUID (streaming
+// consumers forward the raw database and never need these). Linear search:
+// the table exists only between discovery and release_services(), for one
+// small known device.
+
+/// Client Characteristic Configuration descriptor UUID (Bluetooth spec).
+static constexpr uint16_t CCCD_UUID = 0x2902;
+
+// Characteristic property bits (the Bluetooth-spec declaration byte carried
+// in GattCharacteristic::properties; the ESP-IDF macros for these do not
+// exist on the other platforms).
+static constexpr uint8_t GATT_CHAR_PROP_WRITE_NO_RSP = 0x04;
+static constexpr uint8_t GATT_CHAR_PROP_WRITE = 0x08;
+
+inline const GattService *find_service(const GattServiceTable &table, const ESPBTUUID &uuid) {
+  for (uint16_t i = 0; i < table.service_count; i++) {
+    if (table.services[i].uuid == uuid)
+      return &table.services[i];
+  }
+  return nullptr;
+}
+
+const GattCharacteristic *find_characteristic(const GattServiceTable &table, const GattService &service,
+                                              const ESPBTUUID &uuid);
+
+const GattDescriptor *find_descriptor(const GattServiceTable &table, const GattCharacteristic &characteristic,
+                                      const ESPBTUUID &uuid);
+
+/// Handle of the characteristic's Client Characteristic Configuration
+/// descriptor (0x2902), or 0 when it has none.
+uint16_t find_cccd(const GattServiceTable &table, const GattCharacteristic &characteristic);
+
 }  // namespace esphome::ble_device_base
 
 #endif  // USE_BLE_GATT_CLIENT

esphome/components/bluetooth_connection/__init__.py

@@ -1,12 +1,14 @@
 """Per-platform GATT connection backends and the helpers to embed one.
 
-Backends: esp32 Bluedroid, rp2 BTstack. No user-facing configuration; the
-Bluetooth proxy's codegen declares and registers the backend instances
-through gatt_client_schema()/hub_connection_schema() + new_gatt_backend().
+Backends: esp32 Bluedroid, rp2 BTstack. No user-facing configuration; a
+consumer's codegen declares and registers the backend instances — the
+Bluetooth proxy through its per-slot connection wrappers (a streaming
+consumer), and the neutral ble_client through gatt_client_schema() +
+new_gatt_backend().
 """
 
 from collections.abc import Awaitable, Callable
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 
 import esphome.codegen as cg
 from esphome.components import rp2040_ble
@@ -23,7 +25,8 @@ from esphome.types import ConfigType
 def AUTO_LOAD() -> list[str]:
     """ble_device_base plus the platform BLE stack the build's backend
     registers with (the Bluedroid header includes the tracker's), so
-    consumers need not know. The platform-less arm serves manifest tooling."""
+    consumers stay platform-blind. The platform-less arm serves tooling that
+    resolves the manifest without a target."""
     if CORE.is_esp32:
         return ["ble_device_base", "esp32_ble_tracker"]
     if CORE.is_rp2:
@@ -63,6 +66,8 @@ DOMAIN = "bluetooth_connection"
 @dataclass
 class _ConnectionData:
     rp2_backend_count: int = 0
+    # GATT connection slots claimed this run, for the platform cap check.
+    slot_consumers: list[str] = field(default_factory=list)
 
 
 def _get_data() -> _ConnectionData:
@@ -119,6 +124,10 @@ class _PlatformBackend:
     backend_class: cg.MockObjClass
     schema_fragment: Callable[[], cv.Schema]
     register: Callable[[cg.MockObj, ConfigType], Awaitable[None]]
+    # Selects the backend's alias-ladder arm (order-independent arms).
+    define: str
+    # The backend's on-demand materializer gate, when it has one.
+    materializer_define: str | None = None
 
 
 # The single registry of platforms with a GATT client backend; a platform
@@ -126,11 +135,20 @@ class _PlatformBackend:
 # platform's arm.
 _PLATFORM_BACKENDS: dict[str, _PlatformBackend] = {
     PLATFORM_ESP32: _PlatformBackend(
-        BluedroidGattClient, _esp32_schema_fragment, _esp32_register
+        BluedroidGattClient,
+        _esp32_schema_fragment,
+        _esp32_register,
+        "USE_BLE_GATT_BACKEND_BLUEDROID",
+        materializer_define="USE_BLUEDROID_GATT_SERVICE_TABLE",
+    ),
+    PLATFORM_RP2: _PlatformBackend(
+        RP2GattClient, _rp2_schema_fragment, _rp2_register, "USE_BLE_GATT_BACKEND_RP2"
     ),
-    PLATFORM_RP2: _PlatformBackend(RP2GattClient, _rp2_schema_fragment, _rp2_register),
 }
 
+# Gates dedicated-backend consumers (cv.only_on).
+GATT_CLIENT_PLATFORMS = list(_PLATFORM_BACKENDS)
+
 
 def _backend_entry(platform: str | None = None) -> _PlatformBackend:
     key = platform if platform is not None else CORE.target_platform
@@ -165,21 +183,89 @@ def hub_connection_schema(platform: str | None = None) -> cv.Schema:
     )
 
 
-async def new_gatt_backend(config: ConfigType) -> cg.MockObj:
+def consume_gatt_slot(
+    consumer: str, count: int = 1
+) -> Callable[[ConfigType], ConfigType]:
+    """Validator claiming GATT connection slots - the one spelling for every
+    claimant. Platforms whose BLE stack owns a connection budget (esp32, rp2)
+    are charged there and their stack's final validation reports an
+    overcommit; the neutral ledger covers any future backend platform without
+    one (the cap check in FINAL_VALIDATE_SCHEMA)."""
+
+    def validator(config: ConfigType) -> ConfigType:
+        _get_data().slot_consumers.extend([consumer] * count)
+        if CORE.is_esp32:
+            from esphome.components import esp32_ble
+
+            esp32_ble.consume_connection_slots(count, consumer)(config)
+        elif CORE.target_platform == PLATFORM_RP2:
+            rp2040_ble.consume_connection_slots(count, consumer)(config)
+        return config
+
+    return validator
+
+
+# Platforms whose BLE stack owns its own connection budget: consume_gatt_slot
+# charges it there, and the stack's final validation is the one place an
+# overcommit is reported (never two messages for one misconfiguration).
+_STACK_BUDGET_PLATFORMS = {PLATFORM_ESP32, PLATFORM_RP2}
+
+
+def _validate_slot_totals(config: ConfigType) -> ConfigType:
+    # Skipped in testing mode so grouped component builds can co-exist
+    # (mirrors esp32_ble.validate_connection_slots).
+    if CORE.testing_mode:
+        return config
+    if CORE.target_platform in _STACK_BUDGET_PLATFORMS:
+        return config
+    if (cap := HUB_MAX_CONNECTIONS.get(CORE.target_platform)) is None:
+        # Any backend platform without a stack budget must carry a cap here
+        # or fail loudly, never fail open.
+        if CORE.target_platform in _PLATFORM_BACKENDS:
+            raise cv.Invalid(
+                f"{CORE.target_platform} has a GATT backend but no slot cap "
+                "in HUB_MAX_CONNECTIONS"
+            )
+        return config
+    claimed = _get_data().slot_consumers
+    if len(claimed) > cap:
+        raise cv.Invalid(
+            f"{CORE.target_platform} supports at most {cap} GATT client "
+            f"connection(s); {len(claimed)} requested by: {', '.join(claimed)}"
+        )
+    return config
+
+
+FINAL_VALIDATE_SCHEMA = _validate_slot_totals
+
+
+async def new_gatt_backend(
+    config: ConfigType, *, service_table: bool = True
+) -> cg.MockObj:
     """Instantiate the backend declared by gatt_client_schema() and register
     it with its platform stack. The connection slot is claimed at validation
-    (the proxy's slot validators), not here.
+    (the consume_gatt_slot validators), not here.
+
+    service_table is honored by the Bluedroid backend only: forward
+    scaffolding for the first esp32 direct consumer, load-bearing on no
+    current build (rp2 ignores the define and always materializes - its
+    proxy hub streams through get_service_table(), so it must keep the
+    materializer regardless of the flag).
     """
     from esphome.components import ble_device_base
 
+    entry = _backend_entry()
     ble_device_base.request_gatt_client()
+    cg.add_define(entry.define)
+    if service_table and entry.materializer_define is not None:
+        cg.add_define(entry.materializer_define)
     backend = cg.new_Pvariable(config[CONF_BACKEND_ID])
     # The backend is the slot's real Component: component keys from the
     # connection entry (setup_priority, ...) apply to it. Consumers whose own
     # schema carries keys that register_component would misapply to the
     # backend (e.g. a polling interval) must not put them in this config.
     await cg.register_component(backend, config)
-    await _backend_entry().register(backend, config)
+    await entry.register(backend, config)
     return backend
 
 
@@ -187,6 +273,7 @@ async def new_gatt_backend(config: ConfigType) -> cg.MockObj:
 # list (this module cannot import bluetooth_proxy to derive it).
 SOURCE_FILE_FRAMEWORKS: dict[str, set[PlatformFramework]] = {
     "bluetooth_connection_bluedroid.cpp": frameworks_for_platforms([PLATFORM_ESP32]),
+    "gatt_service_table_bluedroid.cpp": frameworks_for_platforms([PLATFORM_ESP32]),
     # Every hub platform the proxy admits (the file compiles empty where
     # USE_BLE_GATT_CLIENT is not defined), so a platform gaining a backend
     # cannot hit a missing-symbol trap here.

esphome/components/bluetooth_connection/bluetooth_connection.cpp

@@ -46,7 +46,7 @@ BatchClose close_service_batch(api::BluetoothGATTGetServicesResponse &resp, size
 
 #endif  // USE_BLUETOOTH_PROXY_CONNECTIONS
 
-#if defined(USE_ESP32) && defined(USE_BLE_GATT_CLIENT)
+#if defined(USE_ESP32_BLE) && defined(USE_BLE_GATT_CLIENT)
 namespace esphome::bluetooth_connection {
 
 // Address-scoped Bluedroid maintenance. Gated with the connection surface:
@@ -65,4 +65,4 @@ conn_err_t clear_gatt_cache(uint64_t address) {
 }
 
 }  // namespace esphome::bluetooth_connection
-#endif  // USE_ESP32 && USE_BLE_GATT_CLIENT
+#endif  // USE_ESP32_BLE && USE_BLE_GATT_CLIENT

esphome/components/bluetooth_connection/bluetooth_connection.h

@@ -48,15 +48,16 @@ static constexpr conn_err_t CONN_OK = 0;
 // GATT contract so backend and wrapper cannot drift.
 static constexpr conn_err_t GATT_NOT_CONNECTED = ble_device_base::GATT_ERR_NOT_CONNECTED;
 
-// What the platform's connection backend supports beyond GATT operations;
-// the proxy derives its feature flags and legacy version from these.
-#if defined(USE_ESP32)
+// What the build's connection backend supports beyond GATT operations; the
+// proxy derives its feature flags and legacy version from these. Keyed on
+// the backend define, never the platform, so a second backend on one
+// platform carries its own facts.
+#if defined(USE_BLE_GATT_BACKEND_BLUEDROID)
 static constexpr bool SUPPORTS_PAIRING = true;
 static constexpr bool SUPPORTS_CACHE_CLEARING = true;
-#elif defined(USE_RP2040_BLE) && defined(USE_BLE_GATT_CLIENT)
+#elif defined(USE_BLE_GATT_BACKEND_RP2)
 // The rp2 BTstack backend pairs (just works + bonding); it has no service
-// cache to clear. Keyed on the backend, not the generic client define, so a
-// future backend without pairing keeps the stub arm below.
+// cache to clear.
 static constexpr bool SUPPORTS_PAIRING = true;
 static constexpr bool SUPPORTS_CACHE_CLEARING = false;
 #else
@@ -64,13 +65,14 @@ static constexpr bool SUPPORTS_PAIRING = false;
 static constexpr bool SUPPORTS_CACHE_CLEARING = false;
 #endif
 
-// Address-scoped (not connection-scoped) maintenance requests.
-#if (defined(USE_ESP32) || defined(USE_RP2040_BLE)) && defined(USE_BLE_GATT_CLIENT)
+// Address-scoped (not connection-scoped) maintenance requests; keyed on the
+// stack (the calls need no backend instance).
+#if (defined(USE_ESP32_BLE) || defined(USE_RP2040_BLE)) && defined(USE_BLE_GATT_CLIENT)
 conn_err_t unpair_device(uint64_t address);
 #else
 inline conn_err_t unpair_device(uint64_t) { return GATT_NOT_CONNECTED; }
 #endif
-#if defined(USE_ESP32) && defined(USE_BLE_GATT_CLIENT)
+#if defined(USE_ESP32_BLE) && defined(USE_BLE_GATT_CLIENT)
 conn_err_t clear_gatt_cache(uint64_t address);
 #else
 inline conn_err_t clear_gatt_cache(uint64_t) { return GATT_NOT_CONNECTED; }

esphome/components/bluetooth_connection/bluetooth_connection_bluedroid.cpp

@@ -2,10 +2,11 @@
 
 #if defined(USE_ESP32_BLE) && defined(USE_BLE_GATT_CLIENT)
 
+#include "bluetooth_connection.h"
+
 // The in-place streamer serves the proxy's service-discovery API; backend-only
 // builds compile without the proxy headers or the streamer.
 #ifdef USE_BLUETOOTH_PROXY_CONNECTIONS
-#include "bluetooth_connection.h"
 #include "bluetooth_connection_hub.h"
 
 #include "esphome/components/bluetooth_proxy/bluetooth_proxy.h"
@@ -300,6 +301,9 @@ int BluedroidGattClient::update_connection_params(uint16_t min_interval, uint16_
 
 void BluedroidGattClient::release_services() {
   this->service_total_ = 0;
+#ifdef USE_BLUEDROID_GATT_SERVICE_TABLE
+  this->table_.free();
+#endif
   // Always set: terminates any in-flight stream on every cache config.
   this->services_released_ = true;
 #ifndef CONFIG_BT_GATTC_CACHE_NVS_FLASH
@@ -312,6 +316,24 @@ void BluedroidGattClient::release_services() {
 #endif
 }
 
+#ifdef USE_BLUEDROID_GATT_SERVICE_TABLE
+ble_device_base::GattServiceTable BluedroidGattClient::get_service_table() {
+  // Lifetime: every teardown path (CLOSE_EVT, the safety timeout, stack-down,
+  // passive DISCONNECT) routes through release_services(), so a materialized
+  // table cannot outlive its link.
+  if (this->table_.empty() &&
+      (this->services_released_ || this->service_total_ == 0 ||
+       !this->table_.build(this->gattc_if_, this->conn_id_, this->service_total_, this->connection_index_))) {
+    // Released / no services / failed build all collapse to empty; the
+    // build failures warned above, log the quiet two.
+    ESP_LOGD(TAG, "[%d] No service table (released=%d, services=%u)", this->connection_index_, this->services_released_,
+             this->service_total_);
+    return {};
+  }
+  return this->table_.view();
+}
+#endif  // USE_BLUEDROID_GATT_SERVICE_TABLE
+
 // ---- internals ----
 
 bool BluedroidGattClient::check_addr_(const esp_bd_addr_t &addr) const {
@@ -358,6 +380,11 @@ void BluedroidGattClient::log_gattc_warning_(const char *operation, int code) {
 // ---- service streaming ----
 
 int BluedroidGattClient::handle_search_cmpl_(esp_gatt_status_t status) {
+#ifdef USE_BLUEDROID_GATT_SERVICE_TABLE
+  // Re-discovery moves the counts the table view derives offsets from; free
+  // the stale table.
+  this->table_.free();
+#endif
   // Step down from the fast discovery params.
   this->update_conn_params_(MEDIUM_MIN_CONN_INTERVAL, MEDIUM_MAX_CONN_INTERVAL, 0, MEDIUM_CONN_TIMEOUT, "medium");
   if (status != ESP_GATT_OK) {

esphome/components/bluetooth_connection/bluetooth_connection_bluedroid.h

@@ -11,6 +11,9 @@
 
 #if defined(USE_ESP32_BLE) && defined(USE_BLE_GATT_CLIENT)
 
+#include "bluetooth_connection.h"
+#include "gatt_service_table_bluedroid.h"
+
 #include "esphome/components/ble_device_base/ble_gatt_client.h"
 #include "esphome/components/esp32_ble_tracker/esp32_ble_tracker.h"
 #include "esphome/core/component.h"
@@ -72,11 +75,16 @@ class BluedroidGattClient final : public esp32_ble_tracker::ESPBTClient, public
   int notify_characteristic(uint16_t handle, bool enable);
   int pair();
   int update_connection_params(uint16_t min_interval, uint16_t max_interval, uint16_t latency, uint16_t timeout);
-  // Contract stub: the proxy streams in place; the on-demand materializer
-  // for direct consumers lands with #18205. NOTE: a direct consumer reaching
-  // this stub gets an empty table indistinguishable from a service-less
-  // peer - do not ship one against this backend before the materializer.
+  // On-demand table for direct consumers; the proxy streams instead, so the
+  // materializer compiles only under USE_BLUEDROID_GATT_SERVICE_TABLE (emitted by
+  // direct-consumer codegen, never by the proxy).
+#ifdef USE_BLUEDROID_GATT_SERVICE_TABLE
+  ble_device_base::GattServiceTable get_service_table();
+#else
+  // A direct consumer reaching this stub misconfigured its codegen
+  // (service_table=False): the empty table reads as a service-less peer.
   ble_device_base::GattServiceTable get_service_table() { return {}; }
+#endif
   void release_services();
 
 #ifdef USE_BLUETOOTH_PROXY_CONNECTIONS
@@ -105,6 +113,9 @@ class BluedroidGattClient final : public esp32_ble_tracker::ESPBTClient, public
 
   // Group 1: pointers / composed objects
   ble_device_base::GattClientListener *listener_{nullptr};
+#ifdef USE_BLUEDROID_GATT_SERVICE_TABLE
+  BluedroidServiceTable table_;
+#endif
   // Group 2: 4-byte types
   uint32_t disconnecting_started_{0};
 

esphome/components/bluetooth_connection/bluetooth_connection_gatt_backend.h

@@ -12,10 +12,12 @@
 
 #include "esphome/components/ble_device_base/ble_gatt_client.h"
 
-#if defined(USE_RP2040_BLE)
+// Arms are keyed on codegen-emitted per-backend defines (_PLATFORM_BACKENDS
+// in __init__.py), so they are order-independent.
+#if defined(USE_BLE_GATT_BACKEND_RP2)
 #include "bluetooth_connection_rp2.h"
 #define ESPHOME_BLE_GATT_CONNECTION_TYPE bluetooth_connection::RP2GattClient
-#elif defined(USE_ESP32_BLE)
+#elif defined(USE_BLE_GATT_BACKEND_BLUEDROID)
 #include "bluetooth_connection_bluedroid.h"
 #define ESPHOME_BLE_GATT_CONNECTION_TYPE bluetooth_connection::BluedroidGattClient
 #elif defined(USE_BLE_GATT_CLIENT_STUB_BACKEND)

esphome/components/bluetooth_connection/gatt_service_table_bluedroid.cpp

@@ -0,0 +1,198 @@
+#include "gatt_service_table_bluedroid.h"
+
+#if defined(USE_ESP32_BLE) && defined(USE_BLE_GATT_CLIENT) && defined(USE_BLUEDROID_GATT_SERVICE_TABLE)
+
+#include "esphome/core/log.h"
+
+namespace esphome::bluetooth_connection {
+
+static const char *const TAG = "gatt_service_table";
+
+// A stack that never reports end-of-range would otherwise walk forever.
+static constexpr uint16_t MAX_DESCRIPTORS_PER_CHARACTERISTIC = 64;
+
+// Shared enumeration for both build passes: an identical walk order is what
+// lets the counting pass size the block the filling pass fills.
+// INVALID_OFFSET/NOT_FOUND mean end-of-range; anything else is a failure.
+template<typename ServiceFn, typename CharFn, typename DescFn>
+bool BluedroidServiceTable::walk_(ServiceFn &&on_service, CharFn &&on_char, DescFn &&on_desc) {
+  for (uint16_t s = 0; s < this->service_total_; s++) {
+    esp_gattc_service_elem_t svc;
+    uint16_t svc_count = 1;
+    auto svc_status = esp_ble_gattc_get_service(this->gattc_if_, this->conn_id_, nullptr, &svc, &svc_count, s);
+    if (svc_status != ESP_GATT_OK || svc_count == 0) {
+      this->log_walk_warning_("esp_ble_gattc_get_service", svc_status);
+      return false;
+    }
+    if (!on_service(s, svc)) {
+      return false;
+    }
+    uint16_t svc_chars = 0;
+    auto count_status = esp_ble_gattc_get_attr_count(this->gattc_if_, this->conn_id_, ESP_GATT_DB_CHARACTERISTIC,
+                                                     svc.start_handle, svc.end_handle, 0, &svc_chars);
+    if (count_status != ESP_GATT_OK) {
+      this->log_walk_warning_("esp_ble_gattc_get_attr_count", count_status);
+      return false;
+    }
+    for (uint16_t c = 0; c < svc_chars; c++) {
+      esp_gattc_char_elem_t chr;
+      uint16_t char_count = 1;
+      auto status = esp_ble_gattc_get_all_char(this->gattc_if_, this->conn_id_, svc.start_handle, svc.end_handle, &chr,
+                                               &char_count, c);
+      if (status != ESP_GATT_OK || char_count == 0) {
+        // An early terminator contradicts svc_chars from the same cache;
+        // never build a silently truncated table.
+        this->log_walk_warning_("esp_ble_gattc_get_all_char", status);
+        return false;
+      }
+      if (!on_char(svc, chr)) {
+        return false;
+      }
+      for (uint16_t d = 0;; d++) {
+        if (d == MAX_DESCRIPTORS_PER_CHARACTERISTIC) {
+          // A stack that never reports end-of-range; fail like every other
+          // inconsistency instead of truncating the table silently.
+          ESP_LOGW(TAG, "[%d] Descriptor walk exceeded %u entries", this->log_index_,
+                   MAX_DESCRIPTORS_PER_CHARACTERISTIC);
+          return false;
+        }
+        esp_gattc_descr_elem_t desc;
+        uint16_t desc_count = 1;
+        auto desc_status =
+            esp_ble_gattc_get_all_descr(this->gattc_if_, this->conn_id_, chr.char_handle, &desc, &desc_count, d);
+        if (desc_status == ESP_GATT_INVALID_OFFSET || desc_status == ESP_GATT_NOT_FOUND) {
+          break;
+        }
+        if (desc_status != ESP_GATT_OK || desc_count == 0) {
+          this->log_walk_warning_("esp_ble_gattc_get_all_descr", desc_status);
+          return false;
+        }
+        if (!on_desc(chr, desc)) {
+          return false;
+        }
+      }
+    }
+  }
+  return true;
+}
+
+bool BluedroidServiceTable::count_services(esp_gatt_if_t gattc_if, uint16_t conn_id, uint16_t *total) {
+  uint16_t primary = 0;
+  uint16_t secondary = 0;
+  if (esp_ble_gattc_get_attr_count(gattc_if, conn_id, ESP_GATT_DB_PRIMARY_SERVICE, 0x0001, 0xFFFF, 0, &primary) !=
+          ESP_GATT_OK ||
+      esp_ble_gattc_get_attr_count(gattc_if, conn_id, ESP_GATT_DB_SECONDARY_SERVICE, 0x0001, 0xFFFF, 0, &secondary) !=
+          ESP_GATT_OK) {
+    // A failed count must not read as an authoritative empty database.
+    return false;
+  }
+  *total = primary + secondary;
+  return true;
+}
+
+bool BluedroidServiceTable::build(esp_gatt_if_t gattc_if, uint16_t conn_id, uint16_t service_total, uint8_t log_index) {
+  this->free();
+  this->gattc_if_ = gattc_if;
+  this->conn_id_ = conn_id;
+  this->service_total_ = service_total;
+  this->log_index_ = log_index;
+
+  // Pass 1: count, so one exact-size block holds the whole table.
+  uint16_t char_total = 0;
+  uint16_t desc_total = 0;
+  bool counted = this->walk_([](uint16_t, const esp_gattc_service_elem_t &) { return true; },
+                             [&](const esp_gattc_service_elem_t &, const esp_gattc_char_elem_t &) {
+                               char_total++;
+                               return true;
+                             },
+                             [&](const esp_gattc_char_elem_t &, const esp_gattc_descr_elem_t &) {
+                               desc_total++;
+                               return true;
+                             });
+  if (!counted) {
+    ESP_LOGW(TAG, "[%d] Service table walk failed during count", this->log_index_);
+    this->free();
+    return false;
+  }
+
+  // The arrays share one block; carving stays aligned because each struct's
+  // strictest member is the UUID and array sizes are multiples of it.
+  static_assert(alignof(ble_device_base::GattService) >= alignof(ble_device_base::GattCharacteristic) &&
+                alignof(ble_device_base::GattCharacteristic) >= alignof(ble_device_base::GattDescriptor));
+  size_t svc_bytes = this->service_total_ * sizeof(ble_device_base::GattService);
+  size_t char_bytes = char_total * sizeof(ble_device_base::GattCharacteristic);
+  size_t total_bytes = svc_bytes + char_bytes + desc_total * sizeof(ble_device_base::GattDescriptor);
+  RAMAllocator<uint8_t> allocator(RAMAllocator<uint8_t>::ALLOC_INTERNAL);
+  this->storage_ = allocator.allocate(total_bytes);
+  if (this->storage_ == nullptr) {
+    ESP_LOGW(TAG, "[%d] Service table allocation failed (%u bytes)", this->log_index_,
+             static_cast<unsigned>(total_bytes));
+    this->free();
+    return false;
+  }
+  auto *services = reinterpret_cast<ble_device_base::GattService *>(this->storage_);
+  auto *characteristics = reinterpret_cast<ble_device_base::GattCharacteristic *>(this->storage_ + svc_bytes);
+  auto *descriptors = reinterpret_cast<ble_device_base::GattDescriptor *>(this->storage_ + svc_bytes + char_bytes);
+
+  // Pass 2: fill, bounded by the pass-1 totals. A bound trip or a shortfall
+  // means the cached database changed between the passes; fail the build
+  // rather than serve an inconsistent table (the consumer retries).
+  uint16_t char_index = 0;
+  uint16_t desc_index = 0;
+  ble_device_base::GattService *cur_service = nullptr;
+  ble_device_base::GattCharacteristic *cur_char = nullptr;
+  bool filled = this->walk_(
+      [&](uint16_t s, const esp_gattc_service_elem_t &svc) {
+        cur_service = &services[s];
+        cur_service->uuid = ble_device_base::ESPBTUUID::from_uuid(svc.uuid);
+        cur_service->start_handle = svc.start_handle;
+        cur_service->end_handle = svc.end_handle;
+        cur_service->first_characteristic = char_index;
+        cur_service->characteristic_count = 0;
+        return true;
+      },
+      [&](const esp_gattc_service_elem_t &svc, const esp_gattc_char_elem_t &chr) {
+        if (char_index >= char_total) {
+          return false;
+        }
+        cur_char = &characteristics[char_index++];
+        cur_char->uuid = ble_device_base::ESPBTUUID::from_uuid(chr.uuid);
+        cur_char->value_handle = chr.char_handle;
+        // Bluedroid addresses descriptors by characteristic handle, so the
+        // table's end_handle only needs the service-bounded upper bound.
+        cur_char->end_handle = svc.end_handle;
+        cur_char->properties = chr.properties;
+        cur_char->first_descriptor = desc_index;
+        cur_char->descriptor_count = 0;
+        cur_service->characteristic_count++;
+        return true;
+      },
+      [&](const esp_gattc_char_elem_t &, const esp_gattc_descr_elem_t &desc) {
+        if (desc_index >= desc_total) {
+          return false;
+        }
+        descriptors[desc_index].uuid = ble_device_base::ESPBTUUID::from_uuid(desc.uuid);
+        descriptors[desc_index].handle = desc.handle;
+        desc_index++;
+        cur_char->descriptor_count++;
+        return true;
+      });
+  if (!filled || char_index != char_total || desc_index != desc_total) {
+    // Walk error or the database changed between passes; better an empty
+    // table than a corrupt one.
+    ESP_LOGW(TAG, "[%d] Service table walk mismatch, discarding", this->log_index_);
+    this->free();
+    return false;
+  }
+  this->char_total_ = char_total;
+  this->desc_total_ = desc_total;
+  return true;
+}
+
+void BluedroidServiceTable::log_walk_warning_(const char *operation, int code) {
+  ESP_LOGW(TAG, "[%d] %s failed, status=%d", this->log_index_, operation, code);
+}
+
+}  // namespace esphome::bluetooth_connection
+
+#endif  // USE_ESP32_BLE && USE_BLE_GATT_CLIENT && USE_BLUEDROID_GATT_SERVICE_TABLE

esphome/components/bluetooth_connection/gatt_service_table_bluedroid.h

@@ -0,0 +1,80 @@
+// Owning two-pass materializer of one Bluedroid GATT database snapshot into
+// the neutral GattServiceTable layout, shared by the BluedroidGattClient
+// backend and ble_client's esp32 engine.
+
+#pragma once
+
+#include "esphome/core/defines.h"
+
+#if defined(USE_ESP32_BLE) && defined(USE_BLE_GATT_CLIENT) && defined(USE_BLUEDROID_GATT_SERVICE_TABLE)
+
+#include "esphome/components/ble_device_base/ble_gatt_client.h"
+#include "esphome/core/helpers.h"
+
+#include <esp_gattc_api.h>
+
+namespace esphome::bluetooth_connection {
+
+class BluedroidServiceTable {
+ public:
+  ~BluedroidServiceTable() { this->free(); }
+  // Owns storage_; a copy would double-free.
+  BluedroidServiceTable() = default;
+  BluedroidServiceTable(const BluedroidServiceTable &) = delete;
+  BluedroidServiceTable &operator=(const BluedroidServiceTable &) = delete;
+
+  /// The service count build() requires: the stack's PRIMARY+SECONDARY
+  /// attribute totals, never the SEARCH_RES event count.
+  static bool count_services(esp_gatt_if_t gattc_if, uint16_t conn_id, uint16_t *total);
+
+  /// Two-pass build from the stack's cached database (service_total from
+  /// count_services()). log_index labels warnings. Frees any previous table
+  /// first; on failure the table is left empty.
+  bool build(esp_gatt_if_t gattc_if, uint16_t conn_id, uint16_t service_total, uint8_t log_index);
+
+  // The view is carved from the storage block and the counts on each call
+  // (a cold path) rather than cached, saving a per-instance table member.
+  ble_device_base::GattServiceTable view() const {
+    size_t svc_bytes = this->service_total_ * sizeof(ble_device_base::GattService);
+    size_t char_bytes = this->char_total_ * sizeof(ble_device_base::GattCharacteristic);
+    return {reinterpret_cast<const ble_device_base::GattService *>(this->storage_),
+            reinterpret_cast<const ble_device_base::GattCharacteristic *>(this->storage_ + svc_bytes),
+            reinterpret_cast<const ble_device_base::GattDescriptor *>(this->storage_ + svc_bytes + char_bytes),
+            this->service_total_,
+            this->char_total_,
+            this->desc_total_};
+  }
+
+  // Always resets the counts: a failed build must never leave a non-zero
+  // service_total_ behind a null table.
+  void free() {
+    if (this->storage_ != nullptr) {
+      RAMAllocator<uint8_t> allocator(RAMAllocator<uint8_t>::ALLOC_INTERNAL);
+      allocator.deallocate(this->storage_, 0);
+      this->storage_ = nullptr;
+    }
+    this->service_total_ = 0;
+    this->char_total_ = 0;
+    this->desc_total_ = 0;
+  }
+
+  bool empty() const { return this->storage_ == nullptr; }
+
+ private:
+  template<typename ServiceFn, typename CharFn, typename DescFn>
+  bool walk_(ServiceFn &&on_service, CharFn &&on_char, DescFn &&on_desc);
+  void log_walk_warning_(const char *operation, int code);
+
+  uint8_t *storage_{nullptr};
+  uint16_t service_total_{0};
+  uint16_t char_total_{0};
+  uint16_t desc_total_{0};
+  // Walk context, set by build().
+  uint16_t conn_id_{0};
+  esp_gatt_if_t gattc_if_{};  // uint8_t width
+  uint8_t log_index_{0};
+};
+
+}  // namespace esphome::bluetooth_connection
+
+#endif  // USE_ESP32_BLE && USE_BLE_GATT_CLIENT && USE_BLUEDROID_GATT_SERVICE_TABLE

esphome/components/bluetooth_proxy/__init__.py

@@ -98,9 +98,15 @@ def _esp32_config_schema() -> cv.All:
                 raise cv.Invalid(
                     "Connections can only be used if the proxy is set to active"
                 )
+            # Explicit entries claim slots like the generated ones; dev
+            # historically skipped this, letting an explicit-connections
+            # config evade the controller budget.
+            bluetooth_connection.consume_gatt_slot(
+                "bluetooth_proxy", len(config[CONF_CONNECTIONS])
+            )(config)
         elif config[CONF_ACTIVE]:
             connection_slots: int = config[CONF_CONNECTION_SLOTS]
-            esp32_ble.consume_connection_slots(connection_slots, "bluetooth_proxy")(
+            bluetooth_connection.consume_gatt_slot("bluetooth_proxy", connection_slots)(
                 config
             )
 
@@ -157,14 +163,14 @@ def _rp2_config_schema() -> cv.All:
     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)
+        bluetooth_connection.consume_gatt_slot("bluetooth_proxy", connection_slots)(
+            config
+        )
         return {
             **config,
             CONF_CONNECTIONS: [connection_schema({}) for _ in range(connection_slots)],
@@ -214,7 +220,9 @@ async def _connections_to_code(var: cg.MockObj, config: ConfigType) -> None:
         # sends those requests and their handlers and encoders are dead.
         cg.add_define("USE_BLUETOOTH_PROXY_CONNECTIONS")
     for connection_conf in connections:
-        backend = await bluetooth_connection.new_gatt_backend(connection_conf)
+        backend = await bluetooth_connection.new_gatt_backend(
+            connection_conf, service_table=False
+        )
         connection = cg.new_Pvariable(connection_conf[CONF_ID])
         cg.add(connection.set_backend(backend))
         cg.add(var.register_connection(connection))

esphome/core/defines.h

@@ -318,6 +318,11 @@
 #define USE_ESP32_BLE_SERVER_ON_DISCONNECT
 #define USE_ESP32_BLE_TRACKER
 #define USE_BLE_GATT_CLIENT
+#define USE_BLE_GATT_BACKEND_BLUEDROID
+#define USE_BLUEDROID_GATT_SERVICE_TABLE
+#define USE_BLE_CLIENT_GATT_NODES
+#define USE_BLE_CLIENT_LEGACY_ENGINE
+#define ESPHOME_BLE_CLIENT_MAX_NODES 1
 #define ESPHOME_BLE_GATT_CLIENT_COUNT 1
 #define ESPHOME_ESP32_BLE_TRACKER_LISTENER_COUNT 1
 #define ESPHOME_ESP32_BLE_TRACKER_CLIENT_COUNT 1
@@ -485,7 +490,10 @@
 #define ESPHOME_BLE_DEVICE_BASE_LISTENER_COUNT 1
 #define USE_BLE_SCAN_RESPONSE_MERGER
 #define USE_BLE_GATT_CLIENT
+#define USE_BLE_GATT_BACKEND_RP2
+#define USE_BLE_CLIENT_GATT_NODES
 #define ESPHOME_BLE_GATT_CLIENT_COUNT 3
+#define ESPHOME_BLE_CLIENT_MAX_NODES 1
 #define USE_RP2040_VARIANT_RP2040
 #define USE_SPI
 #ifndef USE_ETHERNET

tests/component_tests/ble_client/test_validation.py

@@ -19,9 +19,21 @@ from esphome.const import (
     CONF_NOTIFY,
     CONF_SERVICE_UUID,
     CONF_TYPE,
+    PlatformFramework,
 )
+from esphome.core import CORE
 from esphome.types import ConfigType
 
+from ..types import SetCoreConfigCallable
+
+
+@pytest.fixture(autouse=True)
+def esp32_platform(set_core_config: SetCoreConfigCallable) -> None:
+    # The raw-gattc node family gates through BLE_CLIENT_SCHEMA's
+    # _legacy_engine_only choke point; these schema tests exercise the esp32 arm.
+    set_core_config(PlatformFramework.ESP32_IDF)
+
+
 DESCRIPTOR_CONFIG: ConfigType = {
     CONF_NAME: "test",
     CONF_SERVICE_UUID: "6E400001-B5A3-F393-E0A9-E50E24DCCA9E",
@@ -84,3 +96,72 @@ def test_on_notify_implies_notify() -> None:
 def test_notify_unchanged_without_on_notify() -> None:
     config: ConfigType = {CONF_NOTIFY: False}
     assert notify_from_on_notify(config)[CONF_NOTIFY] is False
+
+
+def test_legacy_node_choke_point_rejects_other_platforms(
+    set_core_config: SetCoreConfigCallable,
+) -> None:
+    from esphome.components import ble_client
+    from esphome.core import ID
+
+    set_core_config(PlatformFramework.RP2_ARDUINO)
+    with pytest.raises(cv.Invalid, match="not been migrated"):
+        ble_client._legacy_engine_only(ID("x"))
+    # Through the public schema too, so removing the cv.All wiring fails here.
+    with pytest.raises(cv.Invalid, match="not been migrated"):
+        ble_client.BLE_CLIENT_SCHEMA({})
+
+
+def test_neutral_arm_rejects_esp32_only_keys(
+    set_core_config: SetCoreConfigCallable,
+) -> None:
+    # Pins the schema split's rejection side: the legacy-only keys must not
+    # leak into the neutral arm. The hub is registered so the extra key is
+    # the only error - without it the missing-tracker error would satisfy
+    # the raises vacuously.
+    from esphome.components import ble_client, ble_device_base
+
+    set_core_config(PlatformFramework.RP2_ARDUINO)
+    ble_device_base.register_hub_provider("rp2_ble_tracker")
+    CORE.loaded_integrations.add("rp2_ble_tracker")
+    for key in ("name", "on_passkey_request", "on_passkey_notification"):
+        with pytest.raises(cv.Invalid, match="extra keys not allowed"):
+            ble_client.CONFIG_SCHEMA({"mac_address": "AA:BB:CC:DD:EE:FF", key: "x"})
+
+
+def test_security_actions_reject_platforms_without_the_feature(
+    set_core_config: SetCoreConfigCallable,
+) -> None:
+    from esphome.components import ble_client
+
+    set_core_config(PlatformFramework.RP2_ARDUINO)
+    for schema in (
+        ble_client.BLE_PASSKEY_REPLY_ACTION_SCHEMA,
+        ble_client.BLE_NUMERIC_COMPARISON_REPLY_ACTION_SCHEMA,
+        ble_client.BLE_REMOVE_BOND_ACTION_SCHEMA,
+    ):
+        with pytest.raises(cv.Invalid, match="'security' feature, which rp2"):
+            schema({})
+
+
+def test_node_schema_passes_on_every_gatt_platform(
+    set_core_config: SetCoreConfigCallable,
+) -> None:
+    # The neutral node schema carries no engine gate: raw_gattc components
+    # stay choked, gatt_node components validate wherever ble_client does.
+    from esphome.components import ble_client
+
+    for pf in (PlatformFramework.ESP32_IDF, PlatformFramework.RP2_ARDUINO):
+        set_core_config(pf)
+        assert ble_client.NODE_BLE_CLIENT_SCHEMA({})
+
+
+def test_feature_error_names_the_available_features(
+    set_core_config: SetCoreConfigCallable,
+) -> None:
+    from esphome.components import ble_client
+    from esphome.core import ID
+
+    set_core_config(PlatformFramework.RP2_ARDUINO)
+    with pytest.raises(cv.Invalid, match="provides: gatt_node"):
+        ble_client._legacy_engine_only(ID("x"))

tests/component_tests/bluetooth_connection/test_slot_ledger.py

@@ -0,0 +1,79 @@
+"""Tests for the cross-component GATT slot ledger."""
+
+import pytest
+
+from esphome import config_validation as cv
+from esphome.components import (
+    ble_client,
+    ble_device_base,
+    bluetooth_connection,
+    bluetooth_proxy,
+    rp2040_ble,
+)
+from esphome.const import CONF_MAC_ADDRESS, PlatformFramework
+from esphome.core import CORE
+
+from ..types import SetCoreConfigCallable
+
+
+def test_gatt_slot_ledger_rejects_overcommit_on_rp2(
+    set_core_config: SetCoreConfigCallable,
+) -> None:
+    # rp2 owns its budget: the stack's validation reports the overcommit and
+    # the neutral cap check stays silent (one message per misconfiguration).
+    set_core_config(PlatformFramework.RP2_ARDUINO)
+    bluetooth_connection.consume_gatt_slot("bluetooth_proxy", 3)({})
+    bluetooth_connection.consume_gatt_slot("ble_client")({})
+    bluetooth_connection.FINAL_VALIDATE_SCHEMA({})
+    with pytest.raises(cv.Invalid, match="rp2 maximum is 3"):
+        rp2040_ble.validate_connection_slots()
+
+
+def test_gatt_slot_ledger_skipped_in_testing_mode(
+    set_core_config: SetCoreConfigCallable,
+) -> None:
+    # Grouped component builds merge fixtures past the cap; the check defers
+    # to testing mode like esp32_ble.validate_connection_slots.
+    set_core_config(PlatformFramework.RP2_ARDUINO)
+    bluetooth_connection.consume_gatt_slot("bluetooth_proxy", 3)({})
+    bluetooth_connection.consume_gatt_slot("ble_client")({})
+    CORE.testing_mode = True
+    try:
+        bluetooth_connection.FINAL_VALIDATE_SCHEMA({})
+        rp2040_ble.validate_connection_slots()
+    finally:
+        CORE.testing_mode = False
+
+
+def test_real_validators_charge_the_ledger_on_rp2(
+    set_core_config: SetCoreConfigCallable,
+) -> None:
+    # End to end through the component CONFIG_SCHEMAs (no hand charges):
+    # removing either consumer's consume_gatt_slot call fails this test.
+    set_core_config(PlatformFramework.RP2_ARDUINO)
+    ble_device_base.register_hub_provider("rp2_ble_tracker")
+    CORE.loaded_integrations.add("rp2_ble_tracker")
+    bluetooth_proxy.CONFIG_SCHEMA({})
+    ble_client.CONFIG_SCHEMA({CONF_MAC_ADDRESS: "AA:BB:CC:DD:EE:FF"})
+    # The proxy defaults to 3 slots on rp2; ble_client's claim overcommits
+    # and rp2's own budget names every claimant.
+    with pytest.raises(
+        cv.Invalid,
+        match="Components: bluetooth_proxy, bluetooth_proxy, bluetooth_proxy, "
+        "ble_client",
+    ):
+        rp2040_ble.validate_connection_slots()
+
+
+def test_neutral_cap_check_guards_future_hub_platforms(
+    set_core_config: SetCoreConfigCallable, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    # Both current platforms defer to their stack budgets; pin the message
+    # and boundary of the branch a future budget-less hub platform takes.
+    set_core_config(PlatformFramework.RP2_ARDUINO)
+    monkeypatch.setattr(bluetooth_connection, "_STACK_BUDGET_PLATFORMS", set())
+    bluetooth_connection.consume_gatt_slot("bluetooth_proxy", 3)({})
+    bluetooth_connection.FINAL_VALIDATE_SCHEMA({})
+    bluetooth_connection.consume_gatt_slot("ble_client")({})
+    with pytest.raises(cv.Invalid, match="supports at most 3 GATT client connection"):
+        bluetooth_connection.FINAL_VALIDATE_SCHEMA({})

tests/component_tests/bluetooth_proxy/test_platform_gates.py

@@ -186,6 +186,18 @@ def test_rp2_rejects_esp32_only_keys_by_name(
         bluetooth_proxy.CONFIG_SCHEMA({"connections": [{}]})
 
 
+def test_esp32_explicit_connections_claim_gatt_slots(
+    set_core_config: SetCoreConfigCallable,
+) -> None:
+    # Explicit `connections:` entries must charge the slot ledger like the
+    # generated ones; dev historically let them evade the budget.
+    set_core_config(PlatformFramework.ESP32_IDF)
+    bluetooth_proxy.CONFIG_SCHEMA({"active": True, "connections": [{}, {}]})
+    # Exact match (one entry per slot): catches a missed charge and a
+    # double charge alike.
+    assert bluetooth_connection._get_data().slot_consumers == ["bluetooth_proxy"] * 2
+
+
 def test_hub_source_filter_covers_every_hub_platform() -> None:
     # bluetooth_connection cannot import this module to derive the hub.cpp
     # framework set, so pin it here: a platform admitted to the proxy but
@@ -229,6 +241,12 @@ def test_every_registered_hub_platform_has_a_schema_arm() -> None:
     # Hub platforms must also be in the backend registry the shared codegen
     # helpers dispatch on.
     assert registered <= set(bluetooth_connection._PLATFORM_BACKENDS)
+    # Every non-esp32 backend platform must carry a slot cap: without one the
+    # ledger's FINAL_VALIDATE accepts unlimited claims silently (esp32's cap
+    # is the controller budget in esp32_ble).
+    assert set(bluetooth_connection._PLATFORM_BACKENDS) - {"esp32"} <= set(
+        bluetooth_connection.HUB_MAX_CONNECTIONS
+    )
     # The outer walkable schema's bound must stay the loosest platform cap.
     assert (
         max(bluetooth_connection.HUB_MAX_CONNECTIONS.values())

tests/components/ble_client/common-gatt.yaml

@@ -0,0 +1,24 @@
+ble_client:
+  - mac_address: 01:02:03:04:05:06
+    id: test_blec
+    on_connect:
+      then:
+        - ble_client.ble_write:
+            id: test_blec
+            service_uuid: '1802'
+            characteristic_uuid: '2a06'
+            value: [0x04, 0x05, 0x06]
+    on_disconnect:
+      then:
+        - ble_client.disconnect: test_blec
+
+button:
+  - platform: template
+    name: Connect button
+    on_press:
+      - ble_client.connect: test_blec
+      - ble_client.ble_write:
+          id: test_blec
+          service_uuid: '1802'
+          characteristic_uuid: '2a06'
+          value: !lambda return {0x01, 0x02};

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

@@ -0,0 +1,6 @@
+# The neutral engine: the BTstack backend and rp2040_ble come in through
+# bluetooth_connection's auto-load; the tracker hub supplies the sightings.
+packages:
+  common: !include common-gatt.yaml
+
+rp2_ble_tracker:

tests/components/ble_device_base/__init__.py

@@ -10,6 +10,9 @@ def override_manifest(manifest: ComponentManifestOverride) -> None:
     # and the listener vector it dispatches into (codegen-sized by consumers).
     async def to_code_testing(config):
         cg.add_define("USE_BLE_DEVICE_IRK")
+        # The gatt contract test exercises the gated lookup helpers; compile
+        # their definitions (ble_gatt_client.cpp) into the test build.
+        cg.add_define("USE_BLE_GATT_CLIENT")
         cg.add_define("USE_BLE_SCAN_RESPONSE_MERGER")
         cg.add_define("ESPHOME_BLE_DEVICE_BASE_LISTENER_COUNT", 4)
 

tests/components/ble_device_base/test_gatt_client_contract.cpp

@@ -79,4 +79,54 @@ TEST(BleGattClientContract, MinimalImplementerCompilesAndRoutesEvents) {
   EXPECT_EQ(table.descriptor_count, 0);
 }
 
+// A radon_eye_rd200-shaped table: two services, the second holding a
+// notifying characteristic with a CCCD and a bare write characteristic.
+class ServiceTableLookup : public ::testing::Test {
+ protected:
+  void SetUp() override {
+    this->services_[0] = {ESPBTUUID::from_uint16(0x1800), 0x0001, 0x0005, 0, 1};
+    this->services_[1] = {ESPBTUUID::from_uint16(0x1523), 0x0010, 0x0020, 1, 2};
+    this->characteristics_[0] = {ESPBTUUID::from_uint16(0x2A00), 0x0003, 0x0003, 0x02, 0, 0};
+    this->characteristics_[1] = {ESPBTUUID::from_uint16(0x1525), 0x0012, 0x0014, 0x10, 0, 1};
+    this->characteristics_[2] = {ESPBTUUID::from_uint16(0x1524), 0x0016, 0x0016, 0x04, 1, 0};
+    this->descriptors_[0] = {ESPBTUUID::from_uint16(0x2902), 0x0013};
+    this->table_ = {this->services_, this->characteristics_, this->descriptors_, 2, 3, 1};
+  }
+
+  GattService services_[2];
+  GattCharacteristic characteristics_[3];
+  GattDescriptor descriptors_[1];
+  GattServiceTable table_;
+};
+
+TEST_F(ServiceTableLookup, FindsServicesAndCharacteristicsByUuid) {
+  const GattService *service = find_service(this->table_, ESPBTUUID::from_uint16(0x1523));
+  ASSERT_NE(service, nullptr);
+  EXPECT_EQ(service->start_handle, 0x0010);
+  EXPECT_EQ(find_service(this->table_, ESPBTUUID::from_uint16(0xFFFF)), nullptr);
+
+  const GattCharacteristic *characteristic =
+      find_characteristic(this->table_, *service, ESPBTUUID::from_uint16(0x1525));
+  ASSERT_NE(characteristic, nullptr);
+  EXPECT_EQ(characteristic->value_handle, 0x0012);
+  // The lookup is scoped to the service: 0x2A00 lives in the other service.
+  EXPECT_EQ(find_characteristic(this->table_, *service, ESPBTUUID::from_uint16(0x2A00)), nullptr);
+}
+
+TEST_F(ServiceTableLookup, FindsTheCccdAndReportsItsAbsence) {
+  const GattService *service = find_service(this->table_, ESPBTUUID::from_uint16(0x1523));
+  const GattCharacteristic *notify_char = find_characteristic(this->table_, *service, ESPBTUUID::from_uint16(0x1525));
+  EXPECT_EQ(find_cccd(this->table_, *notify_char), 0x0013);
+  const GattCharacteristic *write_char = find_characteristic(this->table_, *service, ESPBTUUID::from_uint16(0x1524));
+  EXPECT_EQ(find_cccd(this->table_, *write_char), 0);
+}
+
+TEST_F(ServiceTableLookup, RejectsRangesThatOverrunTheTable) {
+  // A corrupt index range must fail the lookup, not walk out of bounds.
+  GattService bad_service = {ESPBTUUID::from_uint16(0x1523), 0x0010, 0x0020, 2, 5};
+  EXPECT_EQ(find_characteristic(this->table_, bad_service, ESPBTUUID::from_uint16(0x1524)), nullptr);
+  GattCharacteristic bad_char = {ESPBTUUID::from_uint16(0x1525), 0x0012, 0x0014, 0x10, 0, 9};
+  EXPECT_EQ(find_cccd(this->table_, bad_char), 0);
+}
+
 }  // namespace esphome::ble_device_base::testing

tests/components/bluetooth_proxy/test-passive.esp32-c6-idf.yaml

@@ -1,5 +1,6 @@
 # Advertisement-only proxy on esp32 by explicit choice: no GATT backend is
-# compiled (USE_BLE_GATT_CLIENT unset), which pins the HAS_GATT gating and the
+# compiled (USE_BLE_GATT_CLIENT unset), which pins the
+# USE_BLUETOOTH_PROXY_CONNECTIONS gating and the
 # address-scoped maintenance path that a connections build never exercises.
 # Under batch grouping the active default build is what runs; the standalone
 # compile of this fixture is what exercises the passive gating.