[core] Skip copying entity automation and filter sources when unused (#18602 )

[ota] Restore lazy flash erase for ESP32 OTA with 64 KiB block erase (#18580 )
[core] Replace a damaged existing file in write_file_if_changed (#18665 )
2026-08-23 14:46:20 +00:00 · 2026-08-23 09:05:04 -05:00 · 2026-08-23 09:04:48 -05:00 · 2026-08-23 09:04:24 -05:00 · 2026-08-23 09:04:06 -05:00 · 2026-08-22 22:21:04 -05:00
1606 changed files with 60370 additions and 9436 deletions
@@ -8,7 +8,7 @@ contact_links:
    url: https://github.com/esphome/esphome-webserver/issues/new/choose
    about: Report an issue with the ESPHome web server.
  - name: Report an issue with the ESPHome Builder / Dashboard
-    url: https://github.com/esphome/dashboard/issues/new/choose
+    url: https://github.com/esphome/device-builder/issues/new/choose
    about: Report an issue with the ESPHome Builder / Dashboard.
  - name: Report an issue with the ESPHome API client
    url: https://github.com/esphome/aioesphomeapi/issues/new/choose
@@ -32,7 +32,7 @@ runs:
      # detects the activated venv via ``VIRTUAL_ENV`` so the venv layout
      # downstream jobs rely on is preserved.
      if: steps.cache-venv.outputs.cache-hit != 'true'
-      uses: astral-sh/setup-uv@c771a70e6277c0a99b617c7a806ffedaca235ff9 # v9.0.0
+      uses: astral-sh/setup-uv@20cfd1bf945f4377ade1205e4dbc17946fc9a30d # v10.0.1
      with:
        enable-cache: true
        # Pull request saves land in per-PR scopes nothing else can
@@ -49,7 +49,7 @@ runs:
        python -m venv venv
        source venv/bin/activate
        python --version
-        uv pip install -r requirements.txt -r requirements_test.txt
+        uv pip install -r requirements.txt -r requirements_dev.txt -r requirements_test.txt
        uv pip install -e .
    - name: Create Python virtual environment
      if: steps.cache-venv.outputs.cache-hit != 'true' && runner.os == 'Windows'
@@ -58,5 +58,5 @@ runs:
        python -m venv venv
        source ./venv/Scripts/activate
        python --version
-        uv pip install -r requirements.txt -r requirements_test.txt
+        uv pip install -r requirements.txt -r requirements_dev.txt -r requirements_test.txt
        uv pip install -e .
@@ -70,6 +70,7 @@ async function isStackedPr(github, context) {
 async function detectMergeBranch(github, context) {
  const labels = new Set();
  const baseRef = context.payload.pull_request.base.ref;
+  const defaultBranch = context.payload.repository.default_branch;

  if (baseRef === 'release') {
    labels.add('merging-to-release');
@@ -78,7 +79,7 @@ async function detectMergeBranch(github, context) {
  } else if (await isStackedPr(github, context)) {
    // GitHub manages the merge order for a stack, so these are not blocked.
    labels.add('stacked-pr');
-  } else if (baseRef !== 'dev') {
+  } else if (baseRef !== defaultBranch) {
    // A chain built by hand: it must not merge until its base branch does.
    labels.add('chained-pr');
  }
@@ -43,14 +43,14 @@ const WITHOUT_SCHEMA = 'CODEOWNERS = ["@esphome/core"]';

 // Builds a fresh context for detectMergeBranch tests instead of mutating the
 // shared CONTEXT fixture above (which other describe blocks rely on).
-function makeMergeContext(baseRef, { stack } = {}) {
+function makeMergeContext(baseRef, { stack, defaultBranch = 'dev' } = {}) {
  const pull_request = { number: 1, base: { ref: baseRef } };
  if (stack !== undefined) {
    pull_request.stack = stack;
  }
  return {
    repo: { owner: 'esphome', repo: 'esphome' },
-    payload: { pull_request }
+    payload: { pull_request, repository: { default_branch: defaultBranch } }
  };
 }

@@ -136,6 +136,21 @@ describe('detectMergeBranch', () => {
    assert.deepEqual(Array.from(labels).sort(), ['chained-pr']);
    assert.equal(state.calls, 1);
  });
+
+  it('base ref matches default branch adds no labels', async () => {
+    const { github } = makeStackGithub({ stack: null });
+    const context = makeMergeContext('other', { defaultBranch: 'other' });
+    const labels = await detectMergeBranch(github, context);
+    assert.deepEqual(Array.from(labels).sort(), []);
+  });
+
+  it('base ref dev when the default branch is main adds chained-pr', async () => {
+    const { github } = makeStackGithub({ stack: null });
+    const context = makeMergeContext('dev', { defaultBranch: 'main' });
+    const labels = await detectMergeBranch(github, context);
+    assert.deepEqual(Array.from(labels).sort(), ['chained-pr']);
+  });
+
 });

 // ---------------------------------------------------------------------------
@@ -29,7 +29,7 @@ jobs:
      - name: Set up uv
        # ``--system`` (below) installs into the setup-python interpreter;
        # no venv is created or restored by this workflow.
-        uses: astral-sh/setup-uv@c771a70e6277c0a99b617c7a806ffedaca235ff9 # v9.0.0
+        uses: astral-sh/setup-uv@20cfd1bf945f4377ade1205e4dbc17946fc9a30d # v10.0.1
        with:
          enable-cache: true
          # Pull-request-only workflow: a save could never be shared and
@@ -41,10 +41,32 @@ jobs:
          version: "0.11.15"

      - name: Install apt dependencies
+        # PR-only workflow, so nothing on dev could seed a shared apt cache
+        # entry; the cached apt action would save one copy per PR. Plain apt
+        # with every call bounded: the apt.conf.d timeouts make a dead
+        # mirror fail over in seconds, and timeout runs under sudo so it can
+        # kill apt-get itself. Install without update first: image lists are
+        # fresh, and the index refresh is what a congested mirror makes slow.
+        timeout-minutes: 15
        run: |
-          sudo apt update
-          sudo apt-cache show protobuf-compiler
-          sudo apt install -y protobuf-compiler
+          sudo tee /etc/apt/apt.conf.d/99ci-acquire-timeouts >/dev/null <<'EOF'
+          Acquire::Retries "1";
+          Acquire::http::Timeout "15";
+          Acquire::https::Timeout "15";
+          EOF
+          # Common path: the image's package lists are fresh enough.
+          if sudo DEBIAN_FRONTEND=noninteractive timeout -k 15 90 \
+              apt-get install -y protobuf-compiler; then
+            protoc --version
+            exit 0
+          fi
+          # Rescue path: refresh the lists once with a generous bound; the
+          # apt config already fails a stalled mirror over quickly.
+          sudo DEBIAN_FRONTEND=noninteractive timeout -k 10 30 \
+            dpkg --configure -a || true
+          sudo timeout -k 15 300 apt-get update
+          sudo DEBIAN_FRONTEND=noninteractive timeout -k 15 300 \
+            apt-get install -y protobuf-compiler
          protoc --version
      - name: Install python dependencies
        run: uv pip install --system aioesphomeapi -c requirements.txt -r requirements_dev.txt
@@ -67,7 +67,7 @@ jobs:
        with:
          python-version: "3.12"
      - name: Set up Docker Buildx
-        uses: docker/setup-buildx-action@bb05f3f5519dd87d3ba754cc423b652a5edd6d2c # v4.2.0
+        uses: docker/setup-buildx-action@37fe631027851001ddb9b187196cc803df7f5f0e # v4.3.0

      - name: Determine tag and whether to push
        id: tag
@@ -153,7 +153,7 @@ jobs:
        with:
          python-version: "3.12"
      - name: Set up Docker Buildx
-        uses: docker/setup-buildx-action@bb05f3f5519dd87d3ba754cc423b652a5edd6d2c # v4.2.0
+        uses: docker/setup-buildx-action@37fe631027851001ddb9b187196cc803df7f5f0e # v4.3.0

      - name: Log in to the GitHub container registry
        uses: docker/login-action@dbcb813823bdd20940b903addbd779551569679f # v4.6.0
@@ -182,8 +182,8 @@ jobs:
      contents: read  # actions/checkout to load the test configs
    strategy:
      fail-fast: false
-      # Cap concurrency so this smoke test doesn't hog all the shared runners.
-      max-parallel: 2
+      # Modest cap so this smoke test leaves room on the shared runner pool.
+      max-parallel: 8
      matrix:
        # One entry per distinct toolchain. ESP32 variants (c3/c6/s2/s3/p4)
        # share a toolchain bundle, so esp32 is exercised on the base variant
@@ -49,7 +49,7 @@ jobs:
        # detects the activated venv via ``VIRTUAL_ENV`` so downstream jobs
        # that ``. venv/bin/activate`` see an identical layout.
        if: steps.cache-venv.outputs.cache-hit != 'true'
-        uses: astral-sh/setup-uv@c771a70e6277c0a99b617c7a806ffedaca235ff9 # v9.0.0
+        uses: astral-sh/setup-uv@20cfd1bf945f4377ade1205e4dbc17946fc9a30d # v10.0.1
        with:
          enable-cache: true
          # Pull request saves land in per-PR scopes nothing else can
@@ -65,9 +65,25 @@ jobs:
          python -m venv venv
          . venv/bin/activate
          python --version
-          uv pip install -r requirements.txt -r requirements_dev.txt -r requirements_test.txt pre-commit
+          uv pip install -r requirements.txt -r requirements_dev.txt -r requirements_test.txt
          uv pip install -e .

+  seed-apt-cache:
+    name: Seed apt package cache
+    runs-on: ubuntu-24.04
+    # PR-branch cache saves are invisible to other PRs, so dev/beta/release
+    # pushes seed the one shared entry PR jobs restore. The key is derived
+    # only from the package list and version; keep both identical in every
+    # step that restores it. In ci-status needs so a broken seed fails dev.
+    if: github.event_name == 'push'
+    timeout-minutes: 10
+    steps:
+      - name: Install apt packages (cached)
+        uses: awalsh128/cache-apt-pkgs-action@553a35bb8ebd9fcabcb1c9451aa4c98e1b4ca8a9 # v1.6.3
+        with:
+          packages: libsdl2-dev ccache
+          version: 1.1
+
  determine-jobs:
    name: Determine which jobs to run
    runs-on: ubuntu-24.04
@@ -179,6 +195,7 @@ jobs:
          . venv/bin/activate
          script/ci-custom.py
          script/build_codeowners.py --check
+          script/build_alias_registry.py --check
          script/build_language_schema.py --check
          script/generate-esp32-boards.py --check
          script/generate-rp2-boards.py --check
@@ -208,69 +225,34 @@ jobs:
        run: script/ci-suggest-changes
        if: always()

-  pre-commit-ci-lite:
-    name: pre-commit.ci lite
+  lint-format:
+    name: Check lint and formatting
    runs-on: ubuntu-latest
    needs:
-      - common
      - determine-jobs
    if: github.event_name == 'pull_request' && !startsWith(github.base_ref, 'beta') && !startsWith(github.base_ref, 'release') && needs.determine-jobs.outputs.core-ci == 'true'
    steps:
      - name: Check out code from GitHub
        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
-      - name: Restore Python
-        uses: ./.github/actions/restore-python
+      - name: Run prek
+        uses: j178/prek-action@4e14d07f9231acabce116ccfca13b13dd9755ece # v3.0.0
        with:
-          python-version: ${{ env.DEFAULT_PYTHON }}
-          cache-key: ${{ needs.common.outputs.cache-key }}
-      # Inlined from esphome/pre-commit-action with a restore-only cache
-      # step: the pre-commit-seed-cache job owns saving this cache, so
-      # pull request runs never write per-PR copies.
-      - name: Restore pre-commit cache
-        uses: actions/cache/restore@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6.1.0
-        with:
-          path: ~/.cache/pre-commit
-          # Must match the key pre-commit-seed-cache saves
-          # yamllint disable-line rule:line-length
-          key: pre-commit-3|${{ env.pythonLocation }}|${{ hashFiles('.pre-commit-config.yaml') }}
-      - name: Run pre-commit
+          # Keep in sync with requirements_test.txt.
+          prek-version: "0.4.11"
+          # This job only runs on pull requests, so nothing ever populates
+          # the cache on dev. Every run would miss and then write a per-pull
+          # request copy, which is what the old seed-cache job existed to
+          # avoid. Building the hooks from scratch takes seconds, so skip it.
+          cache: false
        env:
-          SKIP: pylint,ci-custom
-        run: |
-          python -m pip install pre-commit
-          pre-commit run --show-diff-on-failure --color=always --all-files
+          PREK_SKIP: pylint,ci-custom
+      # Pushes any fixes the hooks made back to the pull request. This step
+      # must keep its default name: the GitHub App that performs the push
+      # locates the workflow run by that name.
      - uses: pre-commit-ci/lite-action@5d6cc0eb514c891a40562a58a8e71576c5c7fb43 # v1.1.0
        if: always()
-
-  pre-commit-seed-cache:
-    name: Seed pre-commit cache
-    runs-on: ubuntu-latest
-    needs:
-      - common
-    # Saves a dev-scoped pre-commit cache that pull request runs can
-    # restore, since pre-commit.ci lite itself never runs on dev pushes.
-    if: github.event_name == 'push' && github.ref == 'refs/heads/dev'
-    steps:
-      - name: Check out code from GitHub
-        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
-      - name: Restore Python
-        uses: ./.github/actions/restore-python
        with:
-          python-version: ${{ env.DEFAULT_PYTHON }}
-          cache-key: ${{ needs.common.outputs.cache-key }}
-      - name: Cache pre-commit environments
-        id: cache-pre-commit
-        uses: actions/cache@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6.1.0
-        with:
-          path: ~/.cache/pre-commit
-          # Must match the restore key in pre-commit-ci-lite
-          # yamllint disable-line rule:line-length
-          key: pre-commit-3|${{ env.pythonLocation }}|${{ hashFiles('.pre-commit-config.yaml') }}
-      - name: Install pre-commit hook environments
-        if: steps.cache-pre-commit.outputs.cache-hit != 'true'
-        run: |
-          python -m pip install pre-commit
-          pre-commit install-hooks
+          msg: apply automatic formatting fixes

  pytest:
    name: Run pytest
@@ -357,7 +339,8 @@ jobs:

  integration-tests:
    name: Run integration tests (${{ matrix.bucket.name }})
-    runs-on: ubuntu-latest
+    # Must match seed-apt-cache's image: the apt cache key has no OS in it.
+    runs-on: ubuntu-24.04
    needs:
      - common
      - determine-jobs
@@ -369,24 +352,16 @@ jobs:
    steps:
      - name: Check out code from GitHub
        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
-      - name: Install ccache
-        # Speeds up the host compiles: tests in a bucket compile overlapping
-        # component sets, so later tests reuse earlier tests' objects.
-        run: |
-          sudo apt-get update -qq
-          sudo apt-get install -y --no-install-recommends ccache
-      - name: Restore ccache (restore-only)
-        # esphome stores the PlatformIO ccache under the machine-global cache
-        # dir (see _ccache_env() in esphome/platformio/toolchain.py). The
-        # bucket-name prefix prefers a same-bucket seed; the bare prefix falls
-        # back to any seed when the bucket layout differs from dev.
-        uses: actions/cache/restore@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6.1.0
+      - name: Install apt packages (cached)
+        # ccache speeds up the host compiles. A cache hit never touches apt
+        # (mirror outages cannot hang the job); the timeout bounds the cold
+        # path. Packages and version must match seed-apt-cache exactly;
+        # libsdl2-dev is unused here and carried only for cache-key parity.
+        timeout-minutes: 10
+        uses: awalsh128/cache-apt-pkgs-action@553a35bb8ebd9fcabcb1c9451aa4c98e1b4ca8a9 # v1.6.3
        with:
-          path: ~/.cache/esphome/platformio-ccache
-          key: integration-ccache-${{ matrix.bucket.name }}-${{ github.sha }}
-          restore-keys: |
-            integration-ccache-${{ matrix.bucket.name }}-
-            integration-ccache-
+          packages: libsdl2-dev ccache
+          version: 1.1
      - name: Set up Python 3.13
        id: python
        uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0
@@ -401,7 +376,7 @@ jobs:
      - name: Set up uv
        # Only needed on cache miss to populate the venv.
        if: steps.cache-venv.outputs.cache-hit != 'true'
-        uses: astral-sh/setup-uv@c771a70e6277c0a99b617c7a806ffedaca235ff9 # v9.0.0
+        uses: astral-sh/setup-uv@20cfd1bf945f4377ade1205e4dbc17946fc9a30d # v10.0.1
        with:
          enable-cache: true
          # Pull request saves land in per-PR scopes nothing else can
@@ -435,14 +410,6 @@ jobs:
        # esphome stores the PlatformIO ccache under the machine-global cache
        # dir (see _ccache_env() in esphome/platformio/toolchain.py).
        run: CCACHE_DIR="$HOME/.cache/esphome/platformio-ccache" ccache -s
-      - name: Save ccache
-        # Pull request saves land in per-PR scopes nothing else can reuse;
-        # dev pushes seed the shared copy instead.
-        if: github.event_name != 'pull_request'
-        uses: actions/cache/save@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6.1.0
-        with:
-          path: ~/.cache/esphome/platformio-ccache
-          key: integration-ccache-${{ matrix.bucket.name }}-${{ github.sha }}

  import-time:
    name: Check import esphome.__main__ time
@@ -475,12 +442,17 @@ jobs:
  benchmarks:
    name: Run CodSpeed benchmarks
    runs-on: ubuntu-24.04
+    timeout-minutes: 30
    needs:
      - common
      - determine-jobs
    if: >-
-      (github.event_name == 'push' && github.ref_name == 'dev') ||
-      (github.event_name == 'pull_request' && needs.determine-jobs.outputs.benchmarks == 'true')
+      github.repository == 'esphome/esphome' && (
+        (github.event_name == 'push' && github.ref_name == 'dev') ||
+        (github.event_name == 'pull_request' && needs.determine-jobs.outputs.benchmarks == 'true')
+      )
+    # CodSpeed benchmarks require a CodSpeed account linked to the repository to run
+    # (https://codspeed.io) -- disabled on forks that aren't esphome/esphome itself.
    steps:
      - name: Check out code from GitHub
        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
@@ -494,14 +466,60 @@ jobs:
      - name: Build benchmarks
        id: build
        run: |
+          # pipefail: without it a failed build is masked by the grep/cut
+          # pipeline below, leaving BINARY empty and silently dropping every
+          # C++ benchmark from the run while the job still reports success.
+          set -o pipefail
          . venv/bin/activate
-          export BENCHMARK_LIB_CONFIG=$(python script/setup_codspeed_lib.py)
-          # --build-only prints BUILD_BINARY=<path> to stdout
-          BINARY=$(script/cpp_benchmark.py --all --build-only | grep '^BUILD_BINARY=' | tail -1 | cut -d= -f2-)
+          BENCHMARK_LIB_CONFIG=$(python script/setup_codspeed_lib.py)
+          export BENCHMARK_LIB_CONFIG
+          # --build-only prints BUILD_BINARY=<path> to stdout; the grep is
+          # non-fatal so a missing marker reaches the check below instead of
+          # tripping errexit at this assignment
+          BINARY=$(script/cpp_benchmark.py --all --build-only | { grep '^BUILD_BINARY=' || true; } | tail -1 | cut -d= -f2-)
+          if [ -z "$BINARY" ]; then
+            echo "::error::Benchmark build did not report a binary path"
+            exit 1
+          fi
          echo "binary=$BINARY" >> $GITHUB_OUTPUT

+      - name: Bound apt fetches and pre-install libc6-dbg
+        # The CodSpeed runner installs valgrind + libc6-dbg via its own
+        # unbounded apt-get update; per-invocation apt options cannot reach
+        # it. The apt.conf.d timeouts below bound every later apt call in
+        # this job, the runner's included. Pre-installing libc6-dbg lets the
+        # runner skip apt once its valgrind cache is restored (it checks
+        # ``dpkg -s libc6-dbg``, so the cache action's unregistered restores
+        # would not count). Install without update first: image lists are
+        # fresh, and the index refresh is what a congested mirror makes
+        # slow. Best effort; the job timeout is the last backstop.
+        timeout-minutes: 15
+        continue-on-error: true
+        run: |
+          sudo tee /etc/apt/apt.conf.d/99ci-acquire-timeouts >/dev/null <<'EOF'
+          Acquire::Retries "1";
+          Acquire::http::Timeout "15";
+          Acquire::https::Timeout "15";
+          EOF
+          if dpkg -s libc6-dbg >/dev/null 2>&1; then
+            echo "libc6-dbg already installed"
+            exit 0
+          fi
+          # Common path: the image's package lists are fresh enough.
+          if sudo DEBIAN_FRONTEND=noninteractive timeout -k 15 90 \
+              apt-get install -y libc6-dbg; then
+            exit 0
+          fi
+          # Rescue path: refresh the lists once with a generous bound; the
+          # apt config already fails a stalled mirror over quickly.
+          sudo DEBIAN_FRONTEND=noninteractive timeout -k 10 30 \
+            dpkg --configure -a || true
+          sudo timeout -k 15 300 apt-get update
+          sudo DEBIAN_FRONTEND=noninteractive timeout -k 15 300 \
+            apt-get install -y libc6-dbg
+
      - name: Run CodSpeed benchmarks
-        uses: CodSpeedHQ/action@88472375d0a4572cf70a9f1fe3a4e0ab8da1b924 # v5.0.1
+        uses: CodSpeedHQ/action@4296e51e7041e24dadb86d1d6e8b9320d223dbe8 # v5.0.3
        with:
          run: |
            . venv/bin/activate
@@ -551,7 +569,6 @@ jobs:
      ESPHOME_SDK_NRF_PREFIX: ~/.esphome-sdk-nrf
    strategy:
      fail-fast: false
-      max-parallel: 2
      matrix:
        include:
          - id: clang-tidy
@@ -585,24 +602,29 @@ jobs:
          fetch-depth: 2

      - name: Restore Python
+        id: restore-python
        uses: ./.github/actions/restore-python
        with:
          python-version: ${{ env.DEFAULT_PYTHON }}
          cache-key: ${{ needs.common.outputs.cache-key }}

+      # Key on the exact Python version as well: LibreTiny creates a venv under
+      # ~/.platformio/penv whose interpreter is a symlink into the runner's
+      # hosted toolcache, so a cache saved on an older runner image breaks once
+      # a new image ships a newer patch release and drops the old interpreter.
      - name: Cache platformio
        if: github.ref == 'refs/heads/dev' && matrix.pio_cache_key
        uses: actions/cache@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6.1.0
        with:
          path: ~/.platformio
-          key: platformio-${{ matrix.pio_cache_key }}-${{ hashFiles('platformio.ini') }}
+          key: platformio-${{ matrix.pio_cache_key }}-${{ steps.restore-python.outputs.python-version }}-${{ hashFiles('platformio.ini') }}

      - name: Cache platformio
        if: github.ref != 'refs/heads/dev' && matrix.pio_cache_key
        uses: actions/cache/restore@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6.1.0
        with:
          path: ~/.platformio
-          key: platformio-${{ matrix.pio_cache_key }}-${{ hashFiles('platformio.ini') }}
+          key: platformio-${{ matrix.pio_cache_key }}-${{ steps.restore-python.outputs.python-version }}-${{ hashFiles('platformio.ini') }}

      - name: Cache ESP-IDF install
        if: matrix.cache_idf
@@ -742,7 +764,6 @@ jobs:
      ESPHOME_ESP_IDF_PREFIX: ~/.esphome-idf
    strategy:
      fail-fast: false
-      max-parallel: 3
      matrix:
        include:
          - id: clang-tidy
@@ -822,7 +843,6 @@ jobs:
      ESPHOME_ESP_IDF_PREFIX: ~/.esphome-idf
    strategy:
      fail-fast: false
-      max-parallel: 3
      matrix:
        include:
          - id: clang-tidy
@@ -909,7 +929,7 @@ jobs:
      ESPHOME_SDK_NRF_PREFIX: ~/.esphome-sdk-nrf
    strategy:
      fail-fast: false
-      max-parallel: ${{ (startsWith(github.base_ref, 'beta') || startsWith(github.base_ref, 'release')) && 8 || 4 }}
+      max-parallel: ${{ (startsWith(github.base_ref, 'beta') || startsWith(github.base_ref, 'release')) && 32 || 16 }}
      matrix:
        batch: ${{ fromJson(needs.determine-jobs.outputs.component-test-batches) }}
    steps:
@@ -921,12 +941,17 @@ jobs:
      - name: List components
        run: echo ${{ matrix.batch.components }}

-      - name: Install apt packages
-        # Not cached: this job is pull-request-only, so a cache save could
-        # never be shared and would only consume quota.
-        run: |
-          sudo apt-get update -qq
-          sudo apt-get install -y --no-install-recommends libsdl2-dev ccache
+      - name: Install apt packages (cached)
+        # A cache hit (seeded on dev by seed-apt-cache) never touches apt,
+        # so mirror outages cannot hang this PR-only job; the timeout bounds
+        # the cold path. Packages and version must match seed-apt-cache
+        # exactly. The action has no --no-install-recommends; same package
+        # set this job used before #17463.
+        timeout-minutes: 10
+        uses: awalsh128/cache-apt-pkgs-action@553a35bb8ebd9fcabcb1c9451aa4c98e1b4ca8a9 # v1.6.3
+        with:
+          packages: libsdl2-dev ccache
+          version: 1.1

      - name: Check out code from GitHub
        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
@@ -1128,7 +1153,7 @@ jobs:
        # install step (order-of-magnitude faster on cold boots,
        # with its own wheel cache). actions/setup-python still
        # provides the interpreter.
-        uses: astral-sh/setup-uv@c771a70e6277c0a99b617c7a806ffedaca235ff9 # v9.0.0
+        uses: astral-sh/setup-uv@20cfd1bf945f4377ade1205e4dbc17946fc9a30d # v10.0.1
        with:
          enable-cache: true
          # Pull request saves land in per-PR scopes nothing else can
@@ -1456,16 +1481,16 @@ jobs:
  ci-status:
    name: CI Status
    runs-on: ubuntu-24.04
-    # Listed in the same order the jobs are defined above. Two jobs are
+    # Listed in the same order the jobs are defined above. One job is
    # deliberately left out: "benchmarks" reports through CodSpeed rather than
-    # this check, and "pre-commit-seed-cache" only populates a cache on pushes
-    # to dev.
+    # this check.
    needs:
      - common
+      - seed-apt-cache
      - determine-jobs
      - ci-custom
      - pylint
-      - pre-commit-ci-lite
+      - lint-format
      - pytest
      - codecov-empty-upload
      - integration-tests
@@ -56,7 +56,7 @@ jobs:

      # Initializes the CodeQL tools for scanning.
      - name: Initialize CodeQL
-        uses: github/codeql-action/init@e4fba868fa4b1b91e1fdab776edc8cfbe6e9fb81 # v4.37.3
+        uses: github/codeql-action/init@ff2f1c621b7f889edc0d3c761ac2e6a3f8cdb0dd # v4.37.7
        with:
          languages: ${{ matrix.language }}
          build-mode: ${{ matrix.build-mode }}
@@ -84,6 +84,6 @@ jobs:
          exit 1

      - name: Perform CodeQL Analysis
-        uses: github/codeql-action/analyze@e4fba868fa4b1b91e1fdab776edc8cfbe6e9fb81 # v4.37.3
+        uses: github/codeql-action/analyze@ff2f1c621b7f889edc0d3c761ac2e6a3f8cdb0dd # v4.37.7
        with:
          category: "/language:${{matrix.language}}"
@@ -14,4 +14,4 @@ jobs:
    permissions:
      issues: write         # issues.lock on closed issues
      pull-requests: write  # issues.lock on closed pull requests
-    uses: esphome/workflows/.github/workflows/lock.yml@9f6577fd37b5cf773ab1b9be929714a0dcd15661  # 2026.7.0
+    uses: esphome/workflows/.github/workflows/lock.yml@0fdd5e311b7e744069166696072a1a9cbc5fbeb6  # 2026.8.1
@@ -0,0 +1,38 @@
+---
+name: Nightly Dev Release
+
+# Works out the dated dev tag and starts the release workflow with it, so that
+# the release run is named after the tag it builds. A workflow run name is
+# fixed when the run starts and cannot read a file or the current date.
+
+on:
+  schedule:
+    - cron: "0 2 * * *"
+
+permissions:
+  contents: read  # actions/checkout to read the version from esphome/const.py
+
+jobs:
+  trigger:
+    name: Start release build
+    if: github.repository == 'esphome/esphome'
+    runs-on: ubuntu-latest
+    permissions:
+      contents: read  # actions/checkout to read the version from esphome/const.py
+      actions: write  # gh workflow run starts release.yml
+    steps:
+      - name: Check out the repository
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+
+      - name: Start the release workflow
+        env:
+          GH_TOKEN: ${{ github.token }}
+        run: |
+          VERSION=$(sed -n -E "s/^__version__\s+=\s+\"(.+)\"$/\1/p" esphome/const.py)
+          if [[ -z "$VERSION" ]]; then
+            echo "::error::Could not read __version__ from esphome/const.py"
+            exit 1
+          fi
+          TAG="${VERSION}$(date --utc '+%Y%m%d')"
+          echo "Starting release build for ${TAG}"
+          gh workflow run release.yml --ref "${GITHUB_REF_NAME}" --field tag="${TAG}"
@@ -1,12 +1,23 @@
 ---
 name: Publish Release

+# Releases (production and beta) are named after the version they publish.
+# Dev builds are named after the dated dev tag, which is passed in by the
+# nightly workflow because a run name cannot compute it itself.
+run-name: ${{ github.event.inputs.tag || github.event.release.tag_name || format('Manual build ({0})', github.ref_name) }}
+
 on:
  workflow_dispatch:
+    inputs:
+      tag:
+        description: >-
+          Tag to build. Only supported on dev, where the nightly workflow
+          uses it. Leave empty to build the version from esphome/const.py
+          with today's date appended.
+        required: false
+        default: ""
  release:
    types: [published]
-  schedule:
-    - cron: "0 2 * * *"

 permissions:
  contents: read  # actions/checkout for all jobs; deploy jobs add their own scopes when they need to write
@@ -23,6 +34,8 @@ jobs:
      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
      - name: Get tag
        id: tag
+        env:
+          INPUT_TAG: ${{ github.event.inputs.tag }}
        # yamllint disable rule:line-length
        run: |
          if [[ "${{ github.event_name }}" = "release" ]]; then
@@ -34,12 +47,23 @@ jobs:
              ENVIRONMENT="production"
            fi
          else
-            TAG=$(cat esphome/const.py | sed -n -E "s/^__version__\s+=\s+\"(.+)\"$/\1/p")
-            today="$(date --utc '+%Y%m%d')"
-            TAG="${TAG}${today}"
            BRANCH=${GITHUB_REF#refs/heads/}
+            # The nightly workflow passes the finished tag so that the run name
+            # matches what is built. Without it, work it out here.
+            TAG="${INPUT_TAG}"
+            if [[ -n "$TAG" && "$BRANCH" != "dev" ]]; then
+              echo "::error::The tag input is only supported on dev. A build from ${BRANCH} has to use the tag worked out here, which carries the branch name, so that it cannot publish over the dev, beta, latest or stable images."
+              exit 1
+            fi
+            if [[ -z "$TAG" ]]; then
+              TAG=$(cat esphome/const.py | sed -n -E "s/^__version__\s+=\s+\"(.+)\"$/\1/p")
+              today="$(date --utc '+%Y%m%d')"
+              TAG="${TAG}${today}"
+              if [[ "$BRANCH" != "dev" ]]; then
+                TAG="${TAG}-${BRANCH}"
+              fi
+            fi
            if [[ "$BRANCH" != "dev" ]]; then
-              TAG="${TAG}-${BRANCH}"
              BRANCH_BUILD="true"
              ENVIRONMENT=""
            else
@@ -99,7 +123,7 @@ jobs:
          python-version: "3.12"

      - name: Set up Docker Buildx
-        uses: docker/setup-buildx-action@bb05f3f5519dd87d3ba754cc423b652a5edd6d2c # v4.2.0
+        uses: docker/setup-buildx-action@37fe631027851001ddb9b187196cc803df7f5f0e # v4.3.0

      - name: Log in to docker hub
        uses: docker/login-action@dbcb813823bdd20940b903addbd779551569679f # v4.6.0
@@ -178,7 +202,7 @@ jobs:
          merge-multiple: true

      - name: Set up Docker Buildx
-        uses: docker/setup-buildx-action@bb05f3f5519dd87d3ba754cc423b652a5edd6d2c # v4.2.0
+        uses: docker/setup-buildx-action@37fe631027851001ddb9b187196cc803df7f5f0e # v4.3.0

      - name: Log in to docker hub
        if: matrix.registry == 'dockerhub'
@@ -16,7 +16,7 @@ jobs:
    # No GITHUB_TOKEN permissions: the reusable workflow mints an ESPHome
    # GitHub App token so the labels, comments and closures come from
    # esphome[bot] instead of github-actions[bot].
-    uses: esphome/workflows/.github/workflows/stale.yml@203cea60ebfd18e2b966e57750750e0417a9feec  # main
+    uses: esphome/workflows/.github/workflows/stale.yml@a1c1485ab46ef41a84a6a9d8abd7fa4b7628fd70  # main
    secrets:
      ESPHOME_GITHUB_APP_PRIVATE_KEY: ${{ secrets.ESPHOME_GITHUB_APP_PRIVATE_KEY }}
    with:
@@ -44,10 +44,10 @@ jobs:
      - name: Set up uv
        # An order of magnitude faster than pip on cold boots, with its
        # own wheel cache. ``--system`` (below) installs into the
-        # setup-python interpreter so subsequent ``pre-commit`` /
+        # setup-python interpreter so subsequent ``prek`` /
        # ``script/run-in-env.py`` steps find the deps without a
        # ``uv run`` prefix.
-        uses: astral-sh/setup-uv@c771a70e6277c0a99b617c7a806ffedaca235ff9 # v9.0.0
+        uses: astral-sh/setup-uv@20cfd1bf945f4377ade1205e4dbc17946fc9a30d # v10.0.1
        with:
          enable-cache: true
          # Pin uv version so the action does not have to fetch the
@@ -58,19 +58,19 @@ jobs:
      - name: Install Home Assistant
        run: |
          uv pip install --system -e lib/home-assistant
-          uv pip install --system -r requirements.txt -r requirements_test.txt pre-commit
+          uv pip install --system -r requirements.txt -r requirements_test.txt

      - name: Sync
        run: |
          python ./script/sync-device_class.py

-      - name: Apply pre-commit auto-fixes
+      - name: Apply prek auto-fixes
        # First pass: let formatters (ruff, end-of-file-fixer, etc.) modify
-        # files. pre-commit exits non-zero whenever a hook touches anything,
+        # files. prek exits non-zero whenever a hook touches anything,
        # which would otherwise abort the workflow before the auto-fixes
        # can flow into the sync PR.
        #
-        # SKIP:
+        # PREK_SKIP:
        #   - no-commit-to-branch is a local guard against committing on
        #     dev/release/beta; CI runs on dev by definition, and
        #     peter-evans/create-pull-request creates the branch itself.
@@ -79,18 +79,18 @@ jobs:
        #     the runtime deps (HA + requirements*.txt); main CI already
        #     gates pylint on real PRs.
        env:
-          SKIP: pylint,no-commit-to-branch
-        run: python script/run-in-env.py pre-commit run --all-files || true
+          PREK_SKIP: pylint,no-commit-to-branch
+        run: python script/run-in-env.py prek run --all-files || true

-      - name: Verify pre-commit clean
+      - name: Verify prek clean
        # Second pass: re-run all hooks against the now-fixed tree.
        # Auto-fixers exit 0 (nothing to change); any remaining failure
        # from a check-only hook (flake8 / yamllint / ci-custom) is a
-        # real issue and fails the workflow loudly. Same SKIP list as
+        # real issue and fails the workflow loudly. Same PREK_SKIP list as
        # above for the same reasons.
        env:
-          SKIP: pylint,no-commit-to-branch
-        run: python script/run-in-env.py pre-commit run --all-files
+          PREK_SKIP: pylint,no-commit-to-branch
+        run: python script/run-in-env.py prek run --all-files

      - name: Commit changes
        uses: peter-evans/create-pull-request@5f6978faf089d4d20b00c7766989d076bb2fc7f1 # v8.1.1
@@ -133,6 +133,8 @@ CTestTestfile.cmake
 .gcc-flags.json

 config/
+# Test fixture config/ directories are tracked (the rule above is the dashboard dir)
+!tests/component_tests/**/config/
 tests/build/
 tests/.esphome/
 /.temp-clang-tidy.cpp
@@ -11,7 +11,7 @@ ci:
 repos:
  - repo: https://github.com/astral-sh/ruff-pre-commit
    # Ruff version.
-    rev: v0.16.0
+    rev: v0.16.3
    hooks:
      # Run the linter.
      - id: ruff
@@ -57,6 +57,12 @@ This document provides essential context for AI models interacting with this pro
        - Function-local constants: `lower_snake_case`
        - Protected/private fields: `lower_snake_case_with_trailing_underscore_`
        - Favor descriptive names over abbreviations
+        - Enumerator names: prefix every value of an `enum class` with the enum name converted to
+          `UPPER_SNAKE_CASE` (e.g. `UARTFlushResult::UART_FLUSH_RESULT_SUCCESS`). Never use bare
+          names like `SUCCESS`, `FAILURE`, `OK`, or `FAIL`: platform SDK headers define macros with
+          these common names (for example the Realtek SDKs used by LibreTiny define
+          `#define SUCCESS 0` in `basic_types.h`), and the preprocessor replaces the enumerator
+          before the compiler sees it, breaking the build and clang-tidy on those platforms.

 *   **Python Idioms:**
    *   **Assignment expressions (PEP 572):** Prefer the walrus operator (`:=`) wherever it removes a redundant lookup or a throwaway temporary. The most common case in component code is presence-checking a config key and then indexing it separately — fetch once with `.get()` and bind in the condition instead:
@@ -412,7 +418,7 @@ This document provides essential context for AI models interacting with this pro
 *   **Configuration:**
    *   `pyproject.toml`: Defines the Python project metadata and dependencies.
    *   `platformio.ini`: Configures the PlatformIO build environments for different microcontrollers.
-    *   `.pre-commit-config.yaml`: Configures the pre-commit hooks for linting and formatting.
+    *   `.pre-commit-config.yaml`: Configures the lint and format hooks, run by `prek`.
 *   **CI/CD Pipeline:** Defined in `.github/workflows`.
 *   **Static Analysis & Development:**
    *   `esphome/core/defines.h`: A comprehensive header file containing all `#define` directives that can be added by components using `cg.add_define()` in Python. This file is used exclusively for development, static analysis tools, and CI testing - it is not used during runtime compilation. When developing components that add new defines, they must be added to this file to ensure proper IDE support and static analysis coverage. The file includes feature flags, build configurations, and platform-specific defines that help static analyzers understand the complete codebase without needing to compile for specific platforms.
@@ -420,7 +426,7 @@ This document provides essential context for AI models interacting with this pro
 ## 6. Development & Testing Workflow

 *   **Local Development Environment:** Use the provided Docker container or create a Python virtual environment and install dependencies from `requirements_dev.txt`.
-*   **Running Commands:** Use the `script/run-in-env.py` script to execute commands within the project's virtual environment. For example, to run the linter: `python3 script/run-in-env.py pre-commit run`.
+*   **Running Commands:** Use the `script/run-in-env.py` script to execute commands within the project's virtual environment. For example, to run the linter: `python3 script/run-in-env.py prek run`.
 *   **Testing:**
    *   **Python:** Run unit tests with `pytest`.
    *   **C++:** Use `clang-tidy` for static analysis.
@@ -493,7 +499,7 @@ This document provides essential context for AI models interacting with this pro
    1.  **Fork & Branch:** Create a new branch based on the `dev` branch (always use `git checkout -b <branch-name> dev` to ensure you're branching from `dev`, not the currently checked out branch).
    2.  **Make Changes:** Adhere to all coding conventions and patterns.
    3.  **Test:** Create component tests for all supported platforms and run the full test suite locally.
-    4.  **Lint:** Run `pre-commit` to ensure code is compliant.
+    4.  **Lint:** Run `prek` to ensure code is compliant.
    5.  **Commit:** Commit your changes. There is no strict format for commit messages.
    6.  **Pull Request:** Submit a PR against the `dev` branch. The Pull Request title must start with a `[tag]` prefix. For component work, use the component name (e.g., `[display] Fix bug`, `[abc123] Add new component`); for changes to shared/core code that isn't tied to a single component, use `[core]` (e.g., `[core] Add validator`). Update documentation, examples, and add `CODEOWNERS` entries as needed. Pull requests should always be made using the `.github/PULL_REQUEST_TEMPLATE.md` template - fill out all sections completely without removing any parts of the template.

@@ -757,3 +763,13 @@ The project uses English for non-code content. When drafting documentation, code
 PR descriptions, and similar text, avoid technical jargon. Instead, express concepts in plain English,
 using standard technical terms only when required. Ensure the text is readily comprehensible to a wide
 audience, including non-native English speakers.
+
+## 10. Code Comments
+
+Code comments on individual lines should be used only where necessary to flag issues that may not be obvious
+on a simple reading of the code. Keep them short (e.g. 1 or 2 lines).
+
+Function and method comment blocks may include more detail as required to make
+calling contracts clear and document parameter usage, but should still be kept concise.
+
+Avoid redundancy and repetition; comments should never simply restate what the code already says.
@@ -78,6 +78,7 @@ esphome/components/bl0942/* @dbuezas @dwmw2
 esphome/components/ble_client/* @buxtronix @clydebarrow
 esphome/components/ble_device_base/* @Bl00d-B0b
 esphome/components/ble_nus/* @tomaszduda23
+esphome/components/bluetooth_connection/* @bdraco @jesserockz
 esphome/components/bluetooth_proxy/* @bdraco @jesserockz
 esphome/components/bm8563/* @abmantis
 esphome/components/bme280_base/* @esphome/core
@@ -237,6 +238,7 @@ esphome/components/hlw8032/* @rici4kubicek
 esphome/components/hm3301/* @freekode
 esphome/components/hmac_md5/* @dwmw2
 esphome/components/hmac_sha256/* @dwmw2
+esphome/components/hoermann_hcp/* @zweckj
 esphome/components/homeassistant/* @esphome/core @OttoWinter
 esphome/components/homeassistant/number/* @landonr
 esphome/components/homeassistant/switch/* @Links2004
@@ -288,6 +290,7 @@ esphome/components/ld2412/* @Rihan9
 esphome/components/ld2420/* @descipher
 esphome/components/ld2450/* @hareeshmu
 esphome/components/ld24xx/* @kbx81
+esphome/components/ld6002b/* @hepter
 esphome/components/ledc/* @OttoWinter
 esphome/components/libretiny/* @kuba2k2
 esphome/components/libretiny_pwm/* @kuba2k2
@@ -296,6 +299,7 @@ esphome/components/lightwaverf/* @max246
 esphome/components/lilygo_t5_47/touchscreen/* @jesserockz
 esphome/components/lm75b/* @beormund
 esphome/components/ln882h_ble/* @Bl00d-B0b
+esphome/components/ln882h_ble_tracker/* @Bl00d-B0b
 esphome/components/ln882x/* @lamauny
 esphome/components/lock/* @esphome/core
 esphome/components/logger/* @esphome/core
@@ -350,6 +354,7 @@ esphome/components/mlx90393/* @functionpointer
 esphome/components/mlx90614/* @jesserockz
 esphome/components/mmc5603/* @benhoff
 esphome/components/mmc5983/* @agoode
+esphome/components/modbus_client/* @exciton
 esphome/components/modbus_controller/* @martgras
 esphome/components/modbus_controller/binary_sensor/* @martgras
 esphome/components/modbus_controller/number/* @martgras
@@ -438,6 +443,7 @@ esphome/components/rp2/* @jesserockz
 esphome/components/rp2040_ble/* @bdraco
 esphome/components/rp2040_pio_led_strip/* @Papa-DMan
 esphome/components/rp2040_pwm/* @jesserockz
+esphome/components/rp2_ble_tracker/* @bdraco
 esphome/components/rpi_dpi_rgb/* @clydebarrow
 esphome/components/rtl87xx/* @kuba2k2
 esphome/components/rtttl/* @glmnet @ximex
@@ -460,6 +466,7 @@ esphome/components/sen21231/* @shreyaskarnik
 esphome/components/sen5x/* @martgras
 esphome/components/sen6x/* @martgras @mebner86 @tuct
 esphome/components/sendspin/* @kahrendt
+esphome/components/sendspin/image/* @kahrendt
 esphome/components/sendspin/media_player/* @kahrendt
 esphome/components/sendspin/media_source/* @kahrendt
 esphome/components/sendspin/sensor/* @kahrendt
@@ -48,7 +48,7 @@ PROJECT_NAME           = ESPHome
 # could be handy for archiving the generated documentation or if some version
 # control system is used.

-PROJECT_NUMBER         = 2026.8.0-dev
+PROJECT_NUMBER         = 2026.9.0-dev

 # Using the PROJECT_BRIEF tag one can provide an optional one line description
 # for a project that appears at the top of each page and should give viewer a
@@ -22,7 +22,7 @@ RUN \
    -r /requirements.txt

 # Install the ESPHome Device Builder dashboard.
-RUN uv pip install --no-cache-dir esphome-device-builder==1.9.0
+RUN uv pip install --no-cache-dir esphome-device-builder==1.12.4

 RUN \
    platformio settings set enable_telemetry No \
@@ -2,35 +2,33 @@
 import argparse
 from collections.abc import Callable
 from contextlib import suppress
-from datetime import datetime
 import functools
-import getpass
 import importlib
 import logging
 import os
 from pathlib import Path
 import re
-import shutil
-import subprocess
 import sys
 import time
-from typing import Protocol
+from typing import TYPE_CHECKING, Protocol

 # Note: Do not import modules from esphome.components here, as this would
 # cause them to be loaded before external components are processed, resulting
 # in the built-in version being used instead of the external component one.
-from esphome import const
+from esphome import const, platform_hooks
 from esphome.const import (
    ALLOWED_NAME_CHARS,
    ARGUMENT_HELP_DEVICE,
+    BUNDLE_EXTENSION,
    CONF_API,
-    CONF_AUTH,
    CONF_BAUD_RATE,
    CONF_BROKER,
    CONF_DEASSERT_RTS_DTR,
    CONF_DISABLED,
+    CONF_DISCOVER_IP,
    CONF_ESPHOME,
    CONF_LEVEL,
+    CONF_LOG,
    CONF_LOG_TOPIC,
    CONF_LOGGER,
    CONF_MDNS,
@@ -44,10 +42,12 @@ from esphome.const import (
    CONF_PORT,
    CONF_SUBSTITUTIONS,
    CONF_TOPIC,
-    CONF_USERNAME,
+    CONF_VERSION,
    CONF_WEB_SERVER,
    CONF_WIFI,
    ENV_NOGITIGNORE,
+    KEY_ESP32,
+    KEY_VARIANT,
    SECRETS_FILES,
    Toolchain,
 )
@@ -55,6 +55,7 @@ from esphome.core import CORE, EsphomeError, coroutine
 from esphome.enum import StrEnum
 from esphome.helpers import get_bool_env, indent, is_ip_address
 from esphome.log import AnsiFore, color, setup_log
+from esphome.stacktrace import LogLineProcessor
 from esphome.types import ConfigType
 from esphome.upload_targets import PortType, get_port_type
 from esphome.util import (
@@ -70,6 +71,9 @@ from esphome.util import (
    safe_print,
 )

+if TYPE_CHECKING:
+    import threading
+
 # Keep expensive imports (zeroconf, writer, yaml_util, etc.) out of this
 # module's top level. Every `esphome` invocation — including fast paths
 # like `esphome version` — pays the cost of what's imported here before
@@ -272,8 +276,8 @@ def _unresolved_default_error(purpose: Purpose, defaults: list[str]) -> str:
        if purpose == Purpose.LOGGING and not has_api():
            return (
                "Cannot view logs over the network: no 'api:' component is "
-                "configured. Network log streaming requires the native API; add "
-                "an 'api:' component, enable MQTT logging, or view logs over USB."
+                "configured. Add an 'api:' component, enable MQTT logging, add a "
+                "'web_server:' component, or view logs over USB."
            )
        if purpose == Purpose.UPLOADING and not has_ota():
            return (
@@ -313,9 +317,12 @@ def choose_upload_log_host(
                ]
                resolved.append(choose_prompt(options, purpose=purpose))
            elif device == "OTA":
+                # Logs can stream over a network transport via the native API
+                # or the web_server HTTP SSE feed.
+                network_logging = has_api() or has_web_server_logging()
                # ensure IP adresses are used first
                if is_ip_address(CORE.address) and (
-                    (purpose == Purpose.LOGGING and has_api())
+                    (purpose == Purpose.LOGGING and network_logging)
                    or (purpose == Purpose.UPLOADING and has_ota())
                ):
                    resolved.extend(_resolve_with_cache(CORE.address, purpose))
@@ -327,7 +334,11 @@ def choose_upload_log_host(
                    if has_mqtt_logging():
                        resolved.append("MQTT")

-                    if has_api() and has_non_ip_address() and has_resolvable_address():
+                    if (
+                        network_logging
+                        and has_non_ip_address()
+                        and has_resolvable_address()
+                    ):
                        resolved.extend(_ota_hostnames_for_default(purpose))

                elif purpose == Purpose.UPLOADING:
@@ -389,7 +400,7 @@ def choose_upload_log_host(
            mqtt_config = CORE.config[CONF_MQTT]
            options.append((f"MQTT ({mqtt_config[CONF_BROKER]})", "MQTT"))

-        if has_api():
+        if has_api() or has_web_server_logging():
            add_ota_options()

    elif purpose == Purpose.UPLOADING and has_ota():
@@ -482,10 +493,23 @@ def has_web_server_ota() -> bool:
    )


+def has_web_server_logging() -> bool:
+    """Check if logs can be streamed over the web_server HTTP SSE endpoint.
+
+    The ``web_server`` component exposes a ``/events`` Server-Sent Events
+    stream that carries ``event: log`` frames. This requires version 2+ (the
+    v1 UI has no ``/events`` endpoint) and the ``log`` option enabled (default).
+    """
+    web_conf = CORE.config.get(CONF_WEB_SERVER)
+    if web_conf is None:
+        return False
+    if web_conf.get(CONF_VERSION, 2) == 1:
+        return False
+    return web_conf.get(CONF_LOG, True)
+
+
 def has_mqtt_ip_lookup() -> bool:
    """Check if MQTT is available and IP lookup is supported."""
-    from esphome.components.mqtt import CONF_DISCOVER_IP
-
    if CONF_MQTT not in CORE.config:
        return False
    # Default Enabled
@@ -546,11 +570,48 @@ def has_name_add_mac_suffix() -> bool:


 def mqtt_get_ip(
-    config: ConfigType, username: str, password: str, client_id: str
+    config: ConfigType,
+    username: str,
+    password: str,
+    client_id: str,
+    stop_event: "threading.Event | None" = None,
 ) -> list[str]:
    from esphome import mqtt

-    return mqtt.get_esphome_device_ip(config, username, password, client_id)
+    return mqtt.get_esphome_device_ip(
+        config, username, password, client_id, stop_event=stop_event
+    )
+
+
+def _add_network_device(device: str, network_devices: list[str]) -> None:
+    """Append a device to the list, expanding it through ``CORE.address_cache``.
+
+    If the hostname is already in the address cache (e.g. populated by mDNS
+    discovery), substitute the cached IPs so aioesphomeapi doesn't open its
+    own Zeroconf to re-resolve it. Duplicates are dropped.
+    """
+    if CORE.address_cache and (cached := CORE.address_cache.get_addresses(device)):
+        network_devices.extend(addr for addr in cached if addr not in network_devices)
+    elif device not in network_devices:
+        network_devices.append(device)
+
+
+def _split_network_devices(devices: list[str]) -> tuple[list[str], bool]:
+    """Split the device list into direct addresses and an MQTT-lookup flag.
+
+    Direct addresses are expanded through ``CORE.address_cache`` and deduped
+    the same way ``_resolve_network_devices`` does; MQTT/MQTTIP magic strings
+    are not resolved, only reported via the returned bool so the caller can
+    defer the broker lookup.
+    """
+    network_devices: list[str] = []
+    has_mqtt_lookup = False
+    for device in devices:
+        if get_port_type(device) in _MQTT_PORT_TYPES:
+            has_mqtt_lookup = True
+        else:
+            _add_network_device(device, network_devices)
+    return network_devices, has_mqtt_lookup


 def _resolve_network_devices(
@@ -583,41 +644,47 @@ def _resolve_network_devices(
        if port_type in _MQTT_PORT_TYPES:
            # Only resolve MQTT once, even if multiple MQTT entries
            if not mqtt_resolved:
-                try:
-                    mqtt_ips = mqtt_get_ip(
-                        config, args.username, args.password, args.client_id
-                    )
-                    # pylint can't infer mqtt_get_ip's return through its
-                    # lazy ``from esphome import mqtt`` import, so it flags
-                    # the genexpr below.
-                    network_devices.extend(
-                        addr
-                        for addr in mqtt_ips  # pylint: disable=not-an-iterable
-                        if addr not in network_devices
-                    )
-                except EsphomeError as err:
-                    _LOGGER.warning(
-                        "MQTT IP discovery failed (%s), will try other devices if available",
-                        err,
-                    )
+                mqtt_ips = _mqtt_get_ip_or_warn(
+                    config, args.username, args.password, args.client_id
+                )
+                network_devices.extend(
+                    addr for addr in mqtt_ips if addr not in network_devices
+                )
                mqtt_resolved = True
            continue

-        # If the hostname is already in the address cache (e.g. populated by
-        # mDNS discovery), substitute the cached IPs so aioesphomeapi doesn't
-        # open its own Zeroconf to re-resolve it.
-        if CORE.address_cache and (cached := CORE.address_cache.get_addresses(device)):
-            network_devices.extend(
-                addr for addr in cached if addr not in network_devices
-            )
-        elif device not in network_devices:
-            # Regular network address or IP - add if not already present
-            network_devices.append(device)
+        _add_network_device(device, network_devices)

    return network_devices


+def _mqtt_get_ip_or_warn(
+    config: ConfigType,
+    username: str,
+    password: str,
+    client_id: str,
+    stop_event: "threading.Event | None" = None,
+) -> list[str]:
+    """Look up the device IP via MQTT, returning [] with a warning on failure.
+
+    This owns the failure policy for MQTT IP discovery on paths that have
+    other addresses to fall back on: a broker problem must not abort the
+    operation. Also used as the deferred resolver handed to ``run_logs``,
+    where it runs in a worker thread.
+    """
+    try:
+        return mqtt_get_ip(config, username, password, client_id, stop_event=stop_event)
+    except EsphomeError as err:
+        _LOGGER.warning(
+            "MQTT IP discovery failed (%s), will try other devices if available",
+            err,
+        )
+        return []
+
+
 def run_miniterm(config: ConfigType, port: str, args) -> int:
+    from datetime import datetime
+
    from aioesphomeapi import LogParser
    import serial

@@ -630,18 +697,9 @@ def run_miniterm(config: ConfigType, port: str, args) -> int:
        return 1
    _LOGGER.info("Starting log output from %s with baud rate %s", port, baud_rate)

-    process_stacktrace = None
-
-    try:
-        module = importlib.import_module("esphome.components." + CORE.target_platform)
-        process_stacktrace = module.process_stacktrace
-    except (AttributeError, ImportError):
-        _LOGGER.info(
-            'Stacktrace analysis is unavailable: no compatible analyzer found for target platform "%s".',
-            CORE.target_platform,
-        )
-
-    backtrace_state = False
+    # Decoder resolution, crash isolation, and disable-after-failure
+    # all live in LogLineProcessor, shared with the API log path.
+    processor = LogLineProcessor(config, CORE.target_platform)
    ser = serial.Serial()
    ser.baudrate = baud_rate
    ser.port = port
@@ -681,11 +739,7 @@ def run_miniterm(config: ConfigType, port: str, args) -> int:
                                "utf8", "backslashreplace"
                            )
                            safe_print(parser.parse_line(line, time_str))
-
-                            if process_stacktrace is not None:
-                                backtrace_state = process_stacktrace(
-                                    config, line, backtrace_state
-                                )
+                            processor.process_line(line)
                    except serial.SerialException:
                        _LOGGER.error("Serial port closed!")
                        return 0
@@ -708,9 +762,11 @@ def _wrap_to_code(name, comp, yaml_util):
    async def wrapped(conf):
        cg.add(cg.LineComment(f"{name}:"))
        if comp.config_schema is not None:
-            conf_str = yaml_util.dump(conf)
+            # sort_keys: voluptuous fills defaults in set order, so an
+            # unsorted dump would churn main.cpp and relink every run
+            conf_str = yaml_util.dump(conf, sort_keys=True)
            conf_str = conf_str.replace("//", "")
-            # remove tailing \ to avoid multi-line comment warning
+            # remove trailing \ to avoid multi-line comment warning
            conf_str = conf_str.replace("\\\n", "\n")
            cg.add(cg.LineComment(indent(conf_str)))
        await coro(conf)
@@ -930,9 +986,10 @@ def upload_using_esptool(

    mcu = "esp8266"
    if CORE.is_esp32:
-        from esphome.components.esp32 import get_esp32_variant
-
-        mcu = get_esp32_variant().lower()
+        # Same lookup as esp32.get_esp32_variant(), read directly so the
+        # serial upload path does not import the esp32 package; both the
+        # validator and the warm-cache apply_to_core populate this key.
+        mcu = CORE.data[KEY_ESP32][KEY_VARIANT].lower()

    line_callbacks: list[Callable[[str], str | None]] = []
    if (
@@ -986,6 +1043,8 @@ def upload_using_esptool(


 def upload_using_platformio(config: ConfigType, port: str) -> int:
+    import shutil
+
    from esphome.platformio import toolchain

    # RP2040 platform-raspberrypi build recipe expects firmware.bin.signed for
@@ -1023,6 +1082,8 @@ def upload_using_picotool(config: ConfigType) -> int:
    the mass storage copy approach that causes "disk not ejected properly"
    warnings on macOS.
    """
+    import subprocess
+
    from esphome.platformio import toolchain

    idedata = toolchain.get_idedata(config)
@@ -1129,6 +1190,8 @@ def check_permissions(port: str):
                "the USB cable can be used for data and is not a power-only cable."
            )
        if not (os.access(port, os.R_OK | os.W_OK)):
+            import getpass
+
            raise EsphomeError(
                "You do not have read or write permission on the selected serial port. "
                "To resolve this issue, you can add your user to the dialout group "
@@ -1141,12 +1204,11 @@ def upload_program(
    config: ConfigType, args: ArgsProtocol, devices: list[str]
 ) -> tuple[int, str | None]:
    host = devices[0]
-    try:
-        module = importlib.import_module("esphome.components." + CORE.target_platform)
-        if module.upload_program(config, args, host):
-            return 0, host
-    except AttributeError:
-        pass
+    platform_upload = platform_hooks.get_platform_hook(
+        CORE.target_platform, "upload_program"
+    )
+    if platform_upload is not None and platform_upload(config, args, host):
+        return 0, host

    port_type = get_port_type(host)

@@ -1297,25 +1359,23 @@ def _upload_via_native_api(
 def _upload_via_web_server(
    config: ConfigType, network_devices: list[str], binary: Path
 ) -> tuple[int, str | None]:
-    web_conf = config.get(CONF_WEB_SERVER)
-    if not web_conf:
-        raise EsphomeError(
-            f"Cannot upload via web_server OTA: the {CONF_WEB_SERVER} component "
-            f"is not configured."
-        )
-
-    remote_port = int(web_conf[CONF_PORT])
-    auth = web_conf.get(CONF_AUTH) or {}
-    username = auth.get(CONF_USERNAME)
-    password = auth.get(CONF_PASSWORD)
-
    from esphome import web_server_ota
+    from esphome.web_server_helpers import get_web_server_connection

+    remote_port, username, password = get_web_server_connection(config)
    return web_server_ota.run_ota(
        network_devices, remote_port, username, password, binary
    )


+def _show_logs_via_web_server(config: ConfigType, network_devices: list[str]) -> int:
+    from esphome import web_server_logs
+    from esphome.web_server_helpers import get_web_server_connection
+
+    port, username, password = get_web_server_connection(config)
+    return web_server_logs.run_logs(network_devices, port, username, password)
+
+
 # Layout of esp_partition_info_t on flash. Each entry is 32 bytes, leading with a
 # 16-bit little-endian magic. ESP-IDF defines ESP_PARTITION_MAGIC = 0x50AA (stored as
 # bytes 0xAA, 0x50) for partition entries and ESP_PARTITION_MAGIC_MD5 = 0xEBEB for the
@@ -1406,12 +1466,11 @@ def _should_subscribe_states(args: ArgsProtocol) -> bool:


 def show_logs(config: ConfigType, args: ArgsProtocol, devices: list[str]) -> int | None:
-    try:
-        module = importlib.import_module("esphome.components." + CORE.target_platform)
-        if module.show_logs(config, args, devices):
-            return 0
-    except AttributeError:
-        pass
+    platform_show_logs = platform_hooks.get_platform_hook(
+        CORE.target_platform, "show_logs"
+    )
+    if platform_show_logs is not None and platform_show_logs(config, args, devices):
+        return 0

    if "logger" not in config:
        raise EsphomeError("Logger is not configured!")
@@ -1425,17 +1484,37 @@ def show_logs(config: ConfigType, args: ArgsProtocol, devices: list[str]) -> int
        return run_miniterm(config, port, args)

    # Check if we should use API for logging
-    # Resolve MQTT magic strings to actual IP addresses
-    if has_api() and (
-        network_devices := _resolve_network_devices(devices, config, args)
-    ):
-        from esphome.components.api.client import run_logs
+    if has_api():
+        network_devices, has_mqtt_lookup = _split_network_devices(devices)
+        mqtt_resolver = None
+        if has_mqtt_lookup:
+            if network_devices:
+                # Addresses are already known, so don't block startup on the
+                # MQTT broker lookup; hand it to run_logs as a deferred
+                # resolver that runs in the background and feeds discovered
+                # addresses into the running log client, keeping MQTT as a
+                # fallback for when the known addresses are stale (e.g. DHCP
+                # reassigned the IP).
+                mqtt_resolver = functools.partial(
+                    _mqtt_get_ip_or_warn,
+                    config,
+                    args.username,
+                    args.password,
+                    args.client_id,
+                )
+            else:
+                # The MQTT lookup is the only way to find the device; resolve
+                # it up front since the client needs an address to start with.
+                network_devices = _resolve_network_devices(devices, config, args)
+        if network_devices:
+            from esphome.api_client import run_logs

-        return run_logs(
-            config,
-            network_devices,
-            subscribe_states=_should_subscribe_states(args),
-        )
+            return run_logs(
+                config,
+                network_devices,
+                subscribe_states=_should_subscribe_states(args),
+                mqtt_resolver=mqtt_resolver,
+            )

    if port_type in (PortType.NETWORK, PortType.MQTT) and has_mqtt_logging():
        from esphome import mqtt
@@ -1444,6 +1523,13 @@ def show_logs(config: ConfigType, args: ArgsProtocol, devices: list[str]) -> int
            config, args.topic, args.username, args.password, args.client_id
        )

+    # Fall back to the web_server HTTP SSE log stream for devices that have
+    # web_server: but no api: (the logging counterpart to web_server OTA).
+    if has_web_server_logging() and (
+        network_devices := _resolve_network_devices(devices, config, args)
+    ):
+        return _show_logs_via_web_server(config, network_devices)
+
    raise EsphomeError("No remote or local logging method configured (api/mqtt/logger)")


@@ -1713,7 +1799,7 @@ def command_clean(args: ArgsProtocol, config: ConfigType) -> int | None:


 def command_bundle(args: ArgsProtocol, config: ConfigType) -> int | None:
-    from esphome.bundle import BUNDLE_EXTENSION, ConfigBundleCreator
+    from esphome.bundle import ConfigBundleCreator

    creator = ConfigBundleCreator(config)

@@ -1948,7 +2034,7 @@ def command_rename(args: ArgsProtocol, config: ConfigType) -> int | None:
    new_name = args.name
    for c in new_name:
        if c not in ALLOWED_NAME_CHARS:
-            print(
+            safe_print(
                color(
                    AnsiFore.BOLD_RED,
                    f"'{c}' is an invalid character for names. Valid characters are: "
@@ -1961,7 +2047,7 @@ def command_rename(args: ArgsProtocol, config: ConfigType) -> int | None:

    yaml = yaml_util.load_yaml(CORE.config_path)
    if CONF_ESPHOME not in yaml or CONF_NAME not in yaml[CONF_ESPHOME]:
-        print(
+        safe_print(
            color(
                AnsiFore.BOLD_RED, "Complex YAML files cannot be automatically renamed."
            )
@@ -2008,7 +2094,9 @@ def command_rename(args: ArgsProtocol, config: ConfigType) -> int | None:
            )
            > 1
        ):
-            print(color(AnsiFore.BOLD_RED, "Too many matches in YAML to safely rename"))
+            safe_print(
+                color(AnsiFore.BOLD_RED, "Too many matches in YAML to safely rename")
+            )
            return 1

        new_raw = re.sub(
@@ -2026,7 +2114,7 @@ def command_rename(args: ArgsProtocol, config: ConfigType) -> int | None:
    # ``kitchen``; running ``esphome rename weird-file.yaml kitchen``
    # would otherwise just re-flash the same hostname).
    if new_name == old_name:
-        print(
+        safe_print(
            color(
                AnsiFore.BOLD_RED,
                f"'{new_name}' is already the device's name.",
@@ -2036,7 +2124,7 @@ def command_rename(args: ArgsProtocol, config: ConfigType) -> int | None:

    new_path: Path = CORE.config_dir / (new_name + ".yaml")
    if new_path.resolve() == CORE.config_path.resolve():
-        print(
+        safe_print(
            color(
                AnsiFore.BOLD_RED,
                f"'{new_name}' is already the device's name.",
@@ -2044,7 +2132,7 @@ def command_rename(args: ArgsProtocol, config: ConfigType) -> int | None:
        )
        return 1
    if new_path.exists():
-        print(
+        safe_print(
            color(
                AnsiFore.BOLD_RED,
                f"Cannot rename: {new_path} already exists. "
@@ -2052,7 +2140,7 @@ def command_rename(args: ArgsProtocol, config: ConfigType) -> int | None:
            )
        )
        return 1
-    print(
+    safe_print(
        f"Updating {color(AnsiFore.CYAN, str(CORE.config_path))} to {color(AnsiFore.CYAN, str(new_path))}"
    )
    print()
@@ -2061,7 +2149,7 @@ def command_rename(args: ArgsProtocol, config: ConfigType) -> int | None:

    rc = run_external_process(*ESPHOME_COMMAND, "config", str(new_path))
    if rc != 0:
-        print(color(AnsiFore.BOLD_RED, "Rename failed. Reverting changes."))
+        safe_print(color(AnsiFore.BOLD_RED, "Rename failed. Reverting changes."))
        new_path.unlink()
        return 1

@@ -2087,7 +2175,7 @@ def command_rename(args: ArgsProtocol, config: ConfigType) -> int | None:
    if CORE.config_path != new_path:
        CORE.config_path.unlink()

-    print(color(AnsiFore.BOLD_GREEN, "SUCCESS"))
+    safe_print(color(AnsiFore.BOLD_GREEN, "SUCCESS"))
    print()
    return 0

@@ -2514,6 +2602,49 @@ def parse_args(argv):
    return parser.parse_args(arguments)


+def _warn_if_source_tree_mismatch() -> None:
+    """Warn when the checkout the user is standing in is not the one being run.
+
+    An editable install records one absolute path, so a venv shared between git
+    worktrees (or reused after a checkout is copied or renamed) keeps importing
+    the tree it was installed from. Every command then silently runs, and
+    compiles, sources the user is not looking at. Only fires inside a checkout,
+    so ordinary installs never see it.
+    """
+    try:
+        cwd = Path.cwd()
+    except OSError:
+        return  # working directory is gone; a diagnostic must not break startup
+    for candidate in (cwd, *cwd.parents):
+        if (candidate / "esphome" / "__main__.py").is_file():
+            standing_in = candidate.resolve()
+            break
+    else:
+        return  # not inside a checkout; nothing to compare against
+
+    running = Path(__file__).resolve().parent.parent
+    # Both sides are resolved, so on a case-sensitive filesystem this matches
+    # plain equality. samefile() compares device and inode, which additionally
+    # covers a case-insensitive filesystem (macOS) reaching one directory by
+    # differently cased paths. Falls back to equality if either path is gone.
+    try:
+        same = standing_in.samefile(running)
+    except OSError:
+        same = standing_in == running
+    if same:
+        return
+
+    _LOGGER.warning(
+        "Running ESPHome from a different checkout than the one you are in:\n"
+        "  running from: %s\n"
+        "  you are in:   %s\n"
+        "The installed esphome resolves to the first, so its sources are used.\n"
+        "Run 'python -m esphome' from the second to use that one instead.",
+        running,
+        standing_in,
+    )
+
+
 def run_esphome(argv):
    from esphome.address_cache import AddressCache

@@ -2532,6 +2663,7 @@ def run_esphome(argv):
        args.log_level = "CRITICAL"

    setup_log(log_level=args.log_level)
+    _warn_if_source_tree_mismatch()

    if args.command in PRE_CONFIG_ACTIONS:
        try:
@@ -2563,10 +2695,11 @@ def run_esphome(argv):
        return 0

    # Bundle support: if the configuration is a .esphomebundle, extract it
-    # and rewrite conf_path to the extracted YAML config.
-    from esphome.bundle import is_bundle_path, prepare_bundle_for_compile
+    # and rewrite conf_path to the extracted YAML config. The suffix check
+    # stays inline so the ordinary run never imports esphome.bundle.
+    if conf_path.name.lower().endswith(BUNDLE_EXTENSION):
+        from esphome.bundle import prepare_bundle_for_compile

-    if is_bundle_path(conf_path):
        _LOGGER.info("Extracting config bundle %s...", conf_path)
        conf_path = prepare_bundle_for_compile(conf_path)
        # Update the argument so downstream code sees the extracted path
@@ -2601,33 +2734,35 @@ def run_esphome(argv):
                conf_path.name,
            )

-    if config is None:
+    cache_missed = config is None
+    if cache_missed:
        from esphome.config import read_config

        config = read_config(
            command_line_substitutions,
            skip_external_update=skip_external,
+            # Snapshot only needed by `esphome config --no-defaults`.
+            snapshot_user_config=getattr(args, "no_defaults", False),
        )
-        # Refresh the cache so the next upload/logs hits the fast path
-        # instead of re-running read_config. Skip when the storage
-        # sidecar is absent (no compile has run): the cache would
-        # never be loaded back, so writing secrets to disk is wasted.
-        if cache_eligible and config is not None:
-            from esphome.compiled_config import save_compiled_config
-            from esphome.storage_json import ext_storage_path
-
-            if ext_storage_path(conf_path.name).exists():
-                save_compiled_config(config)
-    if config is None:
-        return 2
+        if config is None:
+            return 2
    CORE.config = config

    # Fallback for platforms whose validators didn't set the toolchain
    # (only the esp32 component reads esp32.framework.toolchain). All
-    # other platforms only support PlatformIO today.
+    # other platforms only support PlatformIO today. Must run before the
+    # cache refresh below so its sidecar records the same toolchain a
+    # compile would.
    if CORE.toolchain is None:
        CORE.toolchain = Toolchain.PLATFORMIO

+    # Refresh the cache so the next upload/logs hits the fast path
+    # instead of re-running read_config.
+    if cache_eligible and cache_missed:
+        from esphome.compiled_config import save_compiled_config_and_sidecar
+
+        save_compiled_config_and_sidecar(config)
+
    if args.command not in POST_CONFIG_ACTIONS:
        safe_print(f"Unknown command {args.command}")
        return 1
@@ -0,0 +1,198 @@
+from __future__ import annotations
+
+import asyncio
+from contextlib import suppress
+import logging
+import threading
+from typing import TYPE_CHECKING, Any
+import warnings
+
+# Suppress protobuf version warnings
+with warnings.catch_warnings():
+    warnings.filterwarnings(
+        "ignore", category=UserWarning, message=".*Protobuf gencode version.*"
+    )
+    from aioesphomeapi import APIClient, parse_log_message
+    from aioesphomeapi.log_runner import async_run
+
+from esphome.const import CONF_ENCRYPTION, CONF_KEY, CONF_PORT, __version__
+from esphome.core import CORE
+from esphome.stacktrace import LogLineProcessor
+from esphome.util import safe_print
+
+if TYPE_CHECKING:
+    from collections.abc import Callable
+
+    from aioesphomeapi.api_pb2 import (
+        SubscribeLogsResponse,  # pylint: disable=no-name-in-module
+    )
+
+
+_LOGGER = logging.getLogger(__name__)
+
+
+async def async_run_logs(
+    config: dict[str, Any],
+    addresses: list[str],
+    subscribe_states: bool = True,
+    mqtt_resolver: Callable[[threading.Event], list[str]] | None = None,
+) -> None:
+    """Run the logs command in the event loop.
+
+    If ``mqtt_resolver`` is given, it is called in a worker thread (paho-mqtt
+    has no asyncio support on Windows) concurrently with the connection
+    attempts to ``addresses``, and any addresses it discovers are fed into
+    the running client. It owns its own failure handling (returning [] when
+    discovery fails) and must honor the ``threading.Event`` it is passed so
+    teardown is not delayed by the lookup's wait window; the initial broker
+    connect itself is only bounded by the socket timeout.
+    """
+    from datetime import datetime
+
+    conf = config["api"]
+    name = config["esphome"]["name"]
+    port: int = int(conf[CONF_PORT])
+    noise_psk: str | None = None
+    if (encryption := conf.get(CONF_ENCRYPTION)) and (key := encryption.get(CONF_KEY)):
+        noise_psk = key
+
+    _LOGGER.info(
+        "Starting log output from %s using esphome API", " or ".join(addresses)
+    )
+
+    cli = APIClient(
+        addresses[0],  # Primary address for compatibility
+        port,
+        "",  # Password auth removed in 2026.1.0
+        client_info=f"ESPHome Logs {__version__}",
+        noise_psk=noise_psk,
+        addresses=addresses,  # Pass all addresses for automatic retry
+        provide_time=False,
+    )
+
+    # Decoder resolution policy lives in LogLineProcessor.
+    processor = LogLineProcessor(config, CORE.target_platform)
+
+    mqtt_task: asyncio.Task[None] | None = None
+    mqtt_stop_event = threading.Event()
+
+    def _cancel_mqtt_discovery() -> None:
+        """Stop the broker lookup once a connection has been established.
+
+        Its answer is only useful while still disconnected: after that it
+        either duplicates the connected address or arrives too late to
+        matter, so don't keep an idle broker session open for it.
+        """
+        mqtt_stop_event.set()
+        if mqtt_task is not None and not mqtt_task.done():
+            mqtt_task.cancel()
+
+    async def _resolve_mqtt_addresses() -> None:
+        """Discover the device address via the MQTT broker in the background."""
+        try:
+            mqtt_ips = await asyncio.to_thread(mqtt_resolver, mqtt_stop_event)
+            if not mqtt_ips:
+                _LOGGER.debug(
+                    "MQTT discovery %s",
+                    "aborted" if mqtt_stop_event.is_set() else "found no addresses",
+                )
+                return
+            if cli.add_addresses(mqtt_ips):
+                _LOGGER.info("Discovered address(es) via MQTT: %s", ", ".join(mqtt_ips))
+            else:
+                _LOGGER.debug(
+                    "MQTT-discovered address(es) already known: %s", ", ".join(mqtt_ips)
+                )
+        except Exception:  # pylint: disable=broad-except
+            # A background task failure would otherwise stay invisible for
+            # the whole session and only re-raise at teardown
+            _LOGGER.exception("MQTT address discovery failed")
+
+    def on_log(msg: SubscribeLogsResponse) -> None:
+        """Handle a new log message."""
+        time_ = datetime.now().astimezone()
+        message: bytes = msg.message
+        text = message.decode("utf8", "backslashreplace")
+        nanoseconds = time_.microsecond // 1000
+        timestamp = (
+            f"[{time_.hour:02}:{time_.minute:02}:{time_.second:02}.{nanoseconds:03}]"
+        )
+        for parsed_msg in parse_log_message(text, timestamp):
+            # safe_print handles the dashboard \033 escaping and falls back
+            # to backslashreplace encoding on stdouts that can't represent
+            # the wifi signal-bar block characters (Windows redirected
+            # cp1252 pipe).
+            safe_print(parsed_msg)
+        for raw_line in text.splitlines():
+            processor.process_line(raw_line)
+
+    # Safe to fall back to plaintext here only for this diagnostics use
+    # case: the stream is one-way from device to client, and this code
+    # never accepts commands or acts on any message the device sends.
+    # An on-path attacker could still both inject fabricated log lines
+    # and passively read the device's log output (and any state data
+    # delivered when subscribe_states is enabled), so this does lose
+    # confidentiality as well as authentication/integrity. That tradeoff
+    # is acceptable for operator-visible logs, which aioesphomeapi also
+    # warns may come from an unverified device. Never mirror this opt-in
+    # for any connection that sends data to the device or uses Home
+    # Assistant actions.
+    stop = await async_run(
+        cli,
+        on_log,
+        name=name,
+        subscribe_states=subscribe_states,
+        allow_plaintext_fallback=True,
+        # A top-level ``deep_sleep:`` block means the device is only awake
+        # briefly; cap the reconnect backoff so a wake window is not missed.
+        deep_sleep="deep_sleep" in config,
+        on_connect=_cancel_mqtt_discovery if mqtt_resolver is not None else None,
+    )
+    try:
+        # Don't start (or keep) the broker lookup if a connection already
+        # succeeded; the stop event doubles as the not-needed-anymore latch
+        # and get_esphome_device_ip returns immediately when it is set.
+        if mqtt_resolver is not None and not mqtt_stop_event.is_set():
+            mqtt_task = asyncio.create_task(_resolve_mqtt_addresses())
+        await asyncio.Event().wait()
+    finally:
+        try:
+            if mqtt_task is not None:
+                # Unblock the worker thread first so it can't hold up
+                # loop.shutdown_default_executor() for the full lookup timeout.
+                mqtt_stop_event.set()
+                # Give the worker a moment to exit through its own error
+                # handling; cancelling first would race out a late failure.
+                done, _ = await asyncio.wait([mqtt_task], timeout=1.0)
+                if not done:
+                    mqtt_task.cancel()
+                # return_exceptions keeps a CancelledError from the cancel()
+                # above from re-raising here and jumping over the stop() below.
+                # The task handles Exception itself, so only a BaseException
+                # escape (e.g. SystemExit from the worker) can land here.
+                (result,) = await asyncio.gather(mqtt_task, return_exceptions=True)
+                if isinstance(result, BaseException) and not isinstance(
+                    result, asyncio.CancelledError
+                ):
+                    _LOGGER.error("MQTT address discovery failed", exc_info=result)
+        finally:
+            # Must run even if a second cancellation lands mid-cleanup above
+            await stop()
+
+
+def run_logs(
+    config: dict[str, Any],
+    addresses: list[str],
+    subscribe_states: bool = True,
+    mqtt_resolver: Callable[[threading.Event], list[str]] | None = None,
+) -> None:
+    """Run the logs command."""
+    with suppress(KeyboardInterrupt):
+        asyncio.run(
+            async_run_logs(
+                config,
+                addresses,
+                subscribe_states=subscribe_states,
+                mqtt_resolver=mqtt_resolver,
+            )
+        )
@@ -11,43 +11,136 @@ from __future__ import annotations

 import asyncio
 from collections.abc import Awaitable, Callable
+from itertools import count
+import logging
 import threading
+from typing import cast
+
+_LOGGER = logging.getLogger(__name__)
+
+# How long the orphan watcher waits for an abandoned coroutine before giving
+# up, so a hung operation does not park a watcher thread forever.
+ORPHAN_WAIT_TIMEOUT = 300.0
+
+
+_runner_ids = count(1)
+
+
+class AsyncDispatchTimeout(TimeoutError):
+    """The caller stopped waiting; the coroutine was abandoned.
+
+    A subclass so callers can tell the dispatcher's own expiry apart from a
+    ``TimeoutError`` raised inside the coroutine, while existing
+    ``except TimeoutError`` handlers keep working.
+    """


 class AsyncThreadRunner[T](threading.Thread):
    """Run an async coroutine in a daemon thread and expose its result.

-    The runner catches all exceptions from the coroutine and stores them in
-    ``exception`` so ``event`` is always set — this prevents callers waiting
-    on ``event`` from hanging forever when the coroutine crashes.
-
-    Typical usage::
-
-        runner = AsyncThreadRunner(lambda: my_coro(arg))
-        runner.start()
-        if not runner.event.wait(timeout=5.0):
-            ...  # timed out
-        if runner.exception is not None:
-            raise runner.exception
-        result = runner.result
+    ``event`` is always set, even when the coroutine crashes, so waiters
+    never hang; ``completed`` distinguishes a delivered result (even a
+    legitimate ``None``) from a coroutine that never finished. Prefer
+    :func:`run_async`; use this class directly only when a failure should
+    degrade to a default value instead of raising.
    """

    def __init__(self, coro_factory: Callable[[], Awaitable[T]]) -> None:
-        super().__init__(daemon=True)
+        super().__init__(daemon=True, name=f"async-thread-runner-{next(_runner_ids)}")
        self._coro_factory = coro_factory
        self.result: T | None = None
        self.exception: BaseException | None = None
+        self.completed = False
        self.event = threading.Event()

    async def _runner(self) -> None:
        try:
            self.result = await self._coro_factory()
-        except Exception as exc:  # noqa: BLE001  # pylint: disable=broad-except
-            # Capture all exceptions so ``event`` is always set — otherwise a
-            # crash would hang the waiter forever.
+            # Distinguishes a delivered result from "never ran", since None
+            # is a valid result value.
+            self.completed = True
+        except BaseException as exc:  # noqa: BLE001  # pylint: disable=broad-except
+            # Capture everything, including BaseException — otherwise a
+            # cancellation or SystemExit would leave ``exception`` unset and
+            # waiters would mistake the empty ``result`` for success.
            self.exception = exc
        finally:
            self.event.set()

    def run(self) -> None:
-        asyncio.run(self._runner())
+        try:
+            asyncio.run(self._runner())
+        except BaseException as exc:  # noqa: BLE001  # pylint: disable=broad-except
+            # asyncio.run itself can fail before _runner executes (e.g. loop
+            # creation under fd exhaustion); record it so waiters never hang.
+            # A failure during loop cleanup after the coroutine completed
+            # must not clobber the delivered result, hence the guard.
+            if self.exception is None and not self.completed:
+                self.exception = exc
+            else:
+                _LOGGER.debug(
+                    "Event loop teardown failed after outcome recorded",
+                    exc_info=True,
+                )
+        finally:
+            self.event.set()
+
+
+def run_async[T](
+    coro_factory: Callable[[], Awaitable[T]],
+    timeout: float | None = None,
+    on_orphan: Callable[[T], None] | None = None,
+) -> T:
+    """Run a coroutine in a daemon-thread event loop and return its result.
+
+    Raises :class:`AsyncDispatchTimeout` if the coroutine does not finish
+    within ``timeout`` seconds; the thread is abandoned and exits with the
+    interpreter. If the abandoned coroutine later produces a result,
+    ``on_orphan`` (if given) is called with it so resources such as a
+    connected socket can be released; delivery is best effort and bounded
+    by ``ORPHAN_WAIT_TIMEOUT``.
+    """
+    runner: AsyncThreadRunner[T] = AsyncThreadRunner(coro_factory)
+    runner.start()
+    if not runner.event.wait(timeout):
+
+        def _cleanup() -> None:
+            if not runner.event.wait(ORPHAN_WAIT_TIMEOUT):
+                # The one state where a resource can genuinely leak; leave
+                # a trace so a recurring hang is attributable.
+                _LOGGER.info(
+                    "Orphan watcher gave up after %.0fs; a late result may leak",
+                    ORPHAN_WAIT_TIMEOUT,
+                )
+                return
+            if not runner.completed:
+                # The only place an abandoned thread's real error surfaces;
+                # without it a late failure hides behind the TimeoutError.
+                # INFO, not DEBUG: it fires at most once per abandoned
+                # operation and the cause may not reproduce on a rerun.
+                _LOGGER.info(
+                    "Abandoned async operation failed",
+                    exc_info=runner.exception,
+                )
+                return
+            if (result := runner.result) is None:
+                return
+            if on_orphan is None:
+                _LOGGER.debug("Discarding late result; no on_orphan handler")
+                return
+            try:
+                on_orphan(result)
+            except Exception:  # pylint: disable=broad-except
+                # INFO, not DEBUG: a failed release means a real leak, and
+                # it fires at most once per abandoned operation.
+                _LOGGER.info("Error releasing orphaned result", exc_info=True)
+
+        threading.Thread(
+            target=_cleanup, daemon=True, name="async-orphan-cleanup"
+        ).start()
+        raise AsyncDispatchTimeout("Timed out waiting for async operation")
+    if (exc := runner.exception) is not None:
+        raise exc
+    if not runner.completed:
+        raise RuntimeError("Async operation finished without a result or an exception")
+    return cast("T", runner.result)
@@ -3,7 +3,12 @@
 import json
 from pathlib import Path

-from esphome.components.esp32 import get_esp32_variant, idf_version
+from esphome.components.esp32 import (
+    get_esp32_variant,
+    get_excluded_builtin_components,
+    get_managed_component_require_names,
+    idf_version,
+)
 import esphome.config_validation as cv
 from esphome.core import CORE
 from esphome.framework_helpers import (
@@ -67,6 +72,13 @@ def has_discovered_components() -> bool:
    return get_available_components() is not None


+def _cmake_quote(value: str) -> str:
+    """Quote a cmake arg value for a set() line. add_cmake_arg rejects
+    whitespace, quotes, and '$', so only backslashes need escaping."""
+    escaped = value.replace("\\", "\\\\")
+    return f'"{escaped}"'
+
+
 def get_project_cmakelists(minimal: bool = False) -> str:
    """Generate the top-level CMakeLists.txt for ESP-IDF project.

@@ -109,6 +121,15 @@ def get_project_cmakelists(minimal: bool = False) -> str:
        else ""
    )

+    # CMake variables registered via cg.add_cmake_arg(). Emitted before
+    # include(project.cmake) so values like EXCLUDE_COMPONENTS are already
+    # set when project.cmake seeds the component list, and on minimal
+    # (discovery) writes too so excluded components never register.
+    cmake_args = "\n".join(
+        f"set({name} {_cmake_quote(value)})"
+        for name, value in sorted(CORE.cmake_args.items())
+    )
+
    # Per-project list exposed as a CMake variable so converted PIO libs
    # can reference ${ESPHOME_PROJECT_MANAGED_COMPONENTS} without baking
    # project-specific names into their cached CMakeLists.
@@ -119,8 +140,6 @@ def get_project_cmakelists(minimal: bool = False) -> str:
    # runs as a separate CMake script invocation that doesn't load the
    # project's top-level CMakeLists; without this, ${ESPHOME_PROJECT_
    # MANAGED_COMPONENTS} in a converted-lib REQUIRES expands to empty).
-    from esphome.components.esp32 import get_managed_component_require_names
-
    managed_components_property = "\n".join(
        f"idf_build_set_property(ESPHOME_PROJECT_MANAGED_COMPONENTS {name} APPEND)"
        for name in get_managed_component_require_names()
@@ -131,12 +150,22 @@ def get_project_cmakelists(minimal: bool = False) -> str:
    # component's REQUIRES including real IDF components). Referenced by
    # src/CMakeLists and by each converted PIO lib's CMakeLists. Skipped
    # on minimal writes because project_description.json may be stale.
+    # Excluded components are dropped here as well: a stale
+    # project_description.json from a build without exclusions may still
+    # list them, and requiring an excluded component pulls it back into
+    # the build (IDF requirement expansion overrides EXCLUDE_COMPONENTS).
+    # Derived from the EXCLUDE_COMPONENTS cmake arg emitted above so the
+    # two can never disagree within one generated file.
    builtin_components_property = (
        ""
        if minimal
        else "\n".join(
            f"idf_build_set_property(ESPHOME_PROJECT_BUILTIN_COMPONENTS {name} APPEND)"
-            for name in sorted(get_available_components() or [])
+            for name in sorted(
+                set(get_available_components() or []).difference(
+                    CORE.cmake_args.get("EXCLUDE_COMPONENTS", "").split(";")
+                )
+            )
        )
    )

@@ -163,6 +192,8 @@ set(CMAKE_NINJA_FORCE_RESPONSE_FILE 1)
 set(IDF_TARGET {idf_target})
 set(EXTRA_COMPONENT_DIRS ${{CMAKE_SOURCE_DIR}}/src)

+{cmake_args}
+
 include($ENV{{IDF_PATH}}/tools/cmake/project.cmake)

 {cpp_standard_options}
@@ -264,3 +295,13 @@ def write_project(minimal: bool = False) -> None:
        CORE.relative_src_path("CMakeLists.txt"),
        get_component_cmakelists(),
    )
+
+    # Snapshot the exclusion set so has_outdated_files() can trigger a
+    # discovery reconfigure when it changes. Excluded components never
+    # register in project_description.json, so re-including one (e.g. a
+    # config gains mqtt) requires a fresh discovery pass before the
+    # ESPHOME_PROJECT_BUILTIN_COMPONENTS property can list it.
+    write_file_if_changed(
+        CORE.relative_build_path("exclude_components.esphomeinternal"),
+        ";".join(get_excluded_builtin_components()),
+    )
@@ -63,6 +63,17 @@ def get_ini_content():
    # Add extra script for C++ flags
    CORE.add_platformio_option("extra_scripts", [f"pre:{CXX_FLAGS_FILE_NAME}"])

+    # Add CMake args. A user-supplied value (str or list) is deliberately
+    # replaced; this option was always overwritten at FINAL priority.
+    if CORE.cmake_args:
+        CORE.add_platformio_option(
+            "board_build.cmake_extra_args",
+            " ".join(
+                f"-D{name}={value}" for name, value in sorted(CORE.cmake_args.items())
+            ),
+            replace=True,
+        )
+
    content = "[platformio]\n"
    content += f"description = ESPHome {__version__}\n"

@@ -20,6 +20,7 @@ from typing import Any

 from esphome import const, yaml_util
 from esphome.const import (
+    BUNDLE_EXTENSION,
    CONF_ESPHOME,
    CONF_EXTERNAL_COMPONENTS,
    CONF_INCLUDES,
@@ -29,12 +30,12 @@ from esphome.const import (
    CONF_TYPE,
 )
 from esphome.core import CORE, EsphomeError
+from esphome.util import filter_yaml_files

 _LOGGER = logging.getLogger(__name__)

 DOMAIN = "bundle"

-BUNDLE_EXTENSION = ".esphomebundle.tar.gz"
 MANIFEST_FILENAME = "manifest.json"
 CURRENT_MANIFEST_VERSION = 1
 MAX_DECOMPRESSED_SIZE = 500 * 1024 * 1024  # 500 MB
@@ -128,6 +129,9 @@ class BundleData:
    """Files components asked to include, keyed under DOMAIN in CORE.data."""

    extra_files: list[Path] = field(default_factory=list)
+    # Directories whose YAML files are scanned for !secret references but
+    # never bundled, e.g. git package checkouts the builder re-fetches.
+    secret_scan_dirs: set[Path] = field(default_factory=set)
    # Original config dir parsed from an extracted bundle's manifest.json,
    # kept in the path flavor of the machine the bundle was created on.
    # The checked flag makes the manifest lookup happen at most once per run;
@@ -155,6 +159,30 @@ def add_bundle_file(path: Path) -> None:
    _get_data().extra_files.append(CORE.relative_config_path(path))


+def add_secret_scan_dir(path: Path) -> None:
+    """Register a directory to scan for ``!secret`` references when bundling.
+
+    The directory's files are not added to the bundle. Components call this
+    for YAML the build consumes without bundling it — such as git-fetched
+    packages, which the builder re-fetches — so the secrets those files
+    reference are still shipped in the filtered secrets file.
+
+    A relative path is taken as relative to the config directory.
+    """
+    if not path.is_absolute():
+        path = CORE.relative_config_path(path)
+    _get_data().secret_scan_dirs.add(path)
+
+
+def _secret_scan_yaml_files() -> list[Path]:
+    """Return the YAML files inside registered secret-scan directories."""
+    return filter_yaml_files(
+        f
+        for scan_dir in _get_data().secret_scan_dirs
+        for f in yaml_util.find_files(scan_dir, "*")
+    )
+
+
 # Windows paths start with a drive letter or contain backslashes; POSIX
 # paths do neither in practice, so this is how the flavor of a recorded
 # path string is recognized on any host.
@@ -310,6 +338,7 @@ class ConfigBundleCreator:
        yaml_sources = [
            bf.source for bf in files if bf.source.suffix in (".yaml", ".yml")
        ]
+        yaml_sources.extend(_secret_scan_yaml_files())
        used_secret_keys = _find_used_secret_keys(yaml_sources)
        filtered_secrets = self._build_filtered_secrets(used_secret_keys)

@@ -394,6 +423,13 @@ class ConfigBundleCreator:
        """
        discovered = yaml_util.discover_user_yaml_files(self._config_path)
        self._secrets_paths.update(discovered.secrets)
+        # A !secret inside a file this re-parse does not reach (for example
+        # a git-fetched package the builder re-fetches) still resolves
+        # against the config-dir secrets.yaml at build time, so always
+        # consider that file; filtering no-ops when no key matches.
+        default_secrets = self._config_dir / yaml_util.SECRET_YAML
+        if default_secrets.is_file():
+            self._secrets_paths.add(default_secrets.resolve())
        config_resolved = self._config_path.resolve()
        for fpath in discovered.files:
            if fpath == config_resolved:
@@ -719,11 +755,6 @@ def _validate_tar_members(tar: tarfile.TarFile, target_dir: Path) -> None:
            )


-def is_bundle_path(path: Path) -> bool:
-    """Check if a path looks like a bundle file."""
-    return path.name.lower().endswith(BUNDLE_EXTENSION)
-
-
 def _add_bytes_to_tar(tar: tarfile.TarFile, name: str, data: bytes) -> None:
    """Add in-memory bytes to a tar archive with deterministic metadata."""
    info = tarfile.TarInfo(name=name)
@@ -25,6 +25,7 @@ from esphome.cpp_generator import (  # noqa: F401
    add,
    add_build_flag,
    add_build_unflag,
+    add_cmake_arg,
    add_cxx_build_flag,
    add_define,
    add_global,
@@ -49,11 +50,13 @@ from esphome.cpp_helpers import (  # noqa: F401
    build_registry_entry,
    build_registry_list,
    extract_registry_entry_config,
+    get_slot_count,
    gpio_pin_expression,
    past_safe_mode,
    register_component,
    register_parented,
    set_setup_priority,
+    slot_counter,
 )
 from esphome.cpp_types import (  # noqa: F401
    NAN,
@@ -1,26 +1,178 @@
 """Validated-config cache for the upload/logs fast path.

-compile dumps the validated config to <data_dir>/storage/<file>.validated.yaml;
+compile dumps the validated config to <data_dir>/storage/<file>.validated.json;
 the next upload/logs for that YAML reuses it instead of running the full
-read_config pipeline. YAML round-trip (yaml_util.dump/load_yaml) keeps
-!lambda/!include/IDs/paths intact; mtime gates staleness.
+read_config pipeline. The cache is deliberately lossy: only ``!lambda``
+bodies survive typed (``Lambda``); IDs, time periods, MAC/IP addresses,
+paths, UUIDs and enums store the same string form the YAML dumper
+produced for them. JSON additionally coerces non-str dict keys to
+strings; validated configs only use string keys (every schema key
+validator is ``cv.string``). mtime gates staleness.
 """

 from __future__ import annotations

+import json
 import logging
 from pathlib import Path
+from typing import Any

-from esphome.core import CORE
+from esphome.const import __version__ as ESPHOME_VERSION
+from esphome.core import CORE, EsphomeError, Lambda
 from esphome.helpers import write_file
-from esphome.storage_json import StorageJSON, ext_storage_path
+from esphome.storage_json import StorageJSON, ext_storage_path, storage_path
 from esphome.types import ConfigType

 _LOGGER = logging.getLogger(__name__)

+# Bump when the on-disk shape changes; a mismatched version falls back
+# to read_config. The envelope also stamps the writing esphome version:
+# after an upgrade the cache holds the previous release's validation, so
+# it falls back once and the re-save self-heals.
+_CACHE_VERSION = 1
+_LAMBDA_KEY = "__esphome_lambda__"
+

 def compiled_config_path(config_filename: str) -> Path:
    """Path to the cached validated config alongside the storage sidecar."""
+    return CORE.data_dir / "storage" / f"{config_filename}.validated.json"
+
+
+def save_compiled_config(config: ConfigType) -> None:
+    """Write the validated-config cache. Always-write so mtime stays fresh.
+
+    Mode 0600 because config validation resolved !secret inline.
+    Failures are non-fatal: the fast path falls back to read_config.
+    """
+    try:
+        # The legacy YAML cache holds inline-resolved secrets and nothing
+        # reads it anymore; drop it even when the write below fails. A
+        # failed removal leaves resolved secrets on disk, so it warns.
+        try:
+            _legacy_compiled_config_path(CORE.config_filename).unlink(missing_ok=True)
+        except OSError as err:
+            _LOGGER.warning(
+                "Could not remove the legacy validated-config cache: %s", err
+            )
+        rendered = json.dumps(
+            {"v": _CACHE_VERSION, "esphome": ESPHOME_VERSION, "config": config},
+            separators=(",", ":"),
+            default=_json_default,
+        )
+        write_file(compiled_config_path(CORE.config_filename), rendered, private=True)
+    except TypeError as err:
+        # Structural, not transient: this config can never cache (e.g. a
+        # non-basic dict key), so every upload/logs pays the slow path.
+        _LOGGER.warning("Cannot cache the validated config: %s", err)
+    except Exception as err:  # noqa: BLE001  # pylint: disable=broad-except
+        # Likely persistent (permissions, full disk): every upload/logs
+        # pays the slow path until it clears, so surface it.
+        _LOGGER.warning("Skipping compiled config cache write: %s", err)
+
+
+def save_compiled_config_and_sidecar(config: ConfigType) -> None:
+    """Refresh the cache from the upload/logs fallback (CORE.config must be set).
+
+    The cache is only written when a complete sidecar is on disk:
+    load_compiled_config can't use it otherwise, and it holds resolved
+    secrets.
+    """
+    if _refresh_sidecar():
+        save_compiled_config(config)
+
+
+def _refresh_sidecar() -> bool:
+    """Ensure a complete sidecar is on disk; True when one is.
+
+    Writes one (without claiming a build) when missing or wizard-only.
+    Failures are non-fatal; the next upload/logs pays the slow path again.
+    """
+    try:
+        path = storage_path()
+        try:
+            old = StorageJSON.load_strict(path)
+        except Exception as err:  # noqa: BLE001  # pylint: disable=broad-except
+            # Present but unreadable: it may hold a real build's metadata,
+            # and a fresh rewrite would also stop the next compile from
+            # cleaning a possibly incoherent build tree.
+            _LOGGER.warning(
+                "Not caching: storage sidecar %s is unreadable (%s)", path, err
+            )
+            return False
+        if old is not None and old.can_apply_to_core():
+            # Compile-written; nothing to refresh.
+            return True
+        if CORE.build_path is not None and CORE.build_path.exists():
+            # An unvalidated build tree: its absent or mismatched sidecar
+            # is what makes the next compile wipe it, so don't vouch for
+            # a build this run never saw.
+            _LOGGER.warning(
+                "Not caching: build tree %s has no matching sidecar; "
+                "'esphome compile' will settle it",
+                CORE.build_path,
+            )
+            return False
+        new = StorageJSON.from_esphome_core(CORE, old, claim_build=False)
+        if not new.can_apply_to_core():
+            _LOGGER.warning("Not caching: rebuilt storage sidecar is still incomplete")
+            return False
+        new.save(path)
+        return True
+    except (OSError, EsphomeError) as err:
+        # write_file wraps OSError into EsphomeError. Persistent
+        # (unwritable storage dir), so surface that every upload/logs
+        # pays the slow path.
+        _LOGGER.warning("Could not refresh the storage sidecar: %s", err)
+    except Exception:  # noqa: BLE001  # pylint: disable=broad-except
+        # A structural bug; keep the traceback so it isn't mistaken
+        # for the I/O failure above.
+        _LOGGER.warning(
+            "Unexpected error refreshing the storage sidecar", exc_info=True
+        )
+    return False
+
+
+def load_compiled_config(conf_path: Path) -> ConfigType | None:
+    """Load the cached validated config and apply storage metadata to CORE.
+
+    Returns None (caller falls back to read_config) when the cache is
+    missing, older than the source YAML, unparseable, a different cache
+    version, or the sidecar is incomplete. The loaded config carries no
+    source ranges; callers must not feed it into read_config/write_cpp.
+    """
+    cache_path = compiled_config_path(conf_path.name)
+    if not _cache_is_fresh(cache_path, conf_path):
+        return None
+
+    try:
+        envelope = json.loads(
+            cache_path.read_text(encoding="utf-8"), object_hook=_decode_object
+        )
+    except (OSError, ValueError) as err:
+        _LOGGER.debug("Ignoring unreadable compiled config cache: %s", err)
+        return None
+
+    if (
+        not isinstance(envelope, dict)
+        or envelope.get("v") != _CACHE_VERSION
+        or envelope.get("esphome") != ESPHOME_VERSION
+        or not isinstance(config := envelope.get("config"), dict)
+    ):
+        _LOGGER.debug("Ignoring compiled config cache with a foreign envelope")
+        return None
+
+    storage = StorageJSON.load(ext_storage_path(conf_path.name))
+    if storage is None or not storage.can_apply_to_core():
+        _LOGGER.debug("Ignoring compiled config cache: sidecar missing or incomplete")
+        return None
+    storage.apply_to_core()
+    return config
+
+
+# Remove before 2027.8: by then every maintained install has saved the
+# JSON cache at least once and dropped its legacy YAML file.
+def _legacy_compiled_config_path(config_filename: str) -> Path:
+    """Path of the pre-JSON YAML cache; only ever removed."""
    return CORE.data_dir / "storage" / f"{config_filename}.validated.yaml"


@@ -32,45 +184,21 @@ def _cache_is_fresh(cache_path: Path, source_path: Path) -> bool:
        return False


-def save_compiled_config(config: ConfigType) -> None:
-    """Write the validated-config cache. Always-write so mtime stays fresh.
+def _json_default(value: Any) -> Any:
+    """Mirror ESPHomeDumper's representers: Lambda stays typed, the rest
+    stringify (IDs, time periods, MAC/IP addresses, paths, UUIDs, enums).

-    Mode 0600 because show_secrets=True resolves !secret inline.
-    Failures are non-fatal: the fast path falls back to read_config.
+    IncludeFile/Extend/Remove have no JSON mirror and would stringify
+    wrong, but none survive validation (config.py's packages merge and
+    the substitution pass consume them) so no guard is spent on them.
    """
-    from esphome import yaml_util
-
-    try:
-        rendered = yaml_util.dump(config, show_secrets=True)
-        write_file(compiled_config_path(CORE.config_filename), rendered, private=True)
-    except Exception as err:  # noqa: BLE001  # pylint: disable=broad-except
-        _LOGGER.debug("Skipping compiled config cache write: %s", err)
+    if isinstance(value, Lambda):
+        return {_LAMBDA_KEY: value.value}
+    return str(value)


-def load_compiled_config(conf_path: Path) -> ConfigType | None:
-    """Load the cached validated config and apply storage metadata to CORE.
-
-    Returns None (caller falls back to read_config) when the cache is
-    missing, older than the source YAML, unparseable, or the sidecar
-    is incomplete.
-    """
-    cache_path = compiled_config_path(conf_path.name)
-    if not _cache_is_fresh(cache_path, conf_path):
-        return None
-
-    from esphome import yaml_util
-
-    try:
-        config = yaml_util.load_yaml(cache_path, clear_secrets=False)
-    except Exception:  # noqa: BLE001  # pylint: disable=broad-except
-        return None
-
-    storage = StorageJSON.load(ext_storage_path(conf_path.name))
-    if storage is None:
-        return None
-    # apply_to_core assumes a real compile wrote the sidecar; wizard-only
-    # sidecars leave both of these unset and can't drive upload/logs.
-    if not storage.core_platform and not storage.target_platform:
-        return None
-    storage.apply_to_core()
-    return config
+def _decode_object(obj: dict[str, Any]) -> Any:
+    """Revive the Lambda sentinel; every other mapping passes through."""
+    if len(obj) == 1 and isinstance(value := obj.get(_LAMBDA_KEY), str):
+        return Lambda(value)
+    return obj
@@ -0,0 +1,10 @@
+"""Component alias registry.
+
+Generated by script/build_alias_registry.py - do not edit manually.
+See the component-alias section of esphome/loader.py.
+"""
+
+# alias -> (canonical component, removal version or None)
+COMPONENT_ALIASES: dict[str, tuple[str, str | None]] = {
+    "rp2040": ("rp2", "2027.7.0"),
+}
@@ -49,6 +49,12 @@ CONFIG_SCHEMA = cv.All(


 async def to_code(config):
+    if CORE.is_esp32:
+        from esphome.components.esp32 import include_builtin_idf_component
+
+        # Re-enable the gptimer driver (excluded by default to save compile time)
+        include_builtin_idf_component("esp_driver_gptimer")
+
    if CORE.is_esp8266:
        # ac_dimmer uses setTimer1Callback which requires the waveform generator
        from esphome.components.esp8266.const import require_waveform
@@ -1,3 +1,5 @@
+from typing import Any
+
 from esphome import pins
 import esphome.codegen as cg
 from esphome.components.esp32 import (
@@ -16,6 +18,7 @@ from esphome.components.esp32 import (
 import esphome.config_validation as cv
 from esphome.const import CONF_ANALOG, CONF_INPUT, CONF_NUMBER, PLATFORM_ESP8266
 from esphome.core import CORE
+from esphome.types import ConfigType

 CODEOWNERS = ["@esphome/core"]

@@ -225,12 +228,13 @@ ESP32_VARIANT_ADC2_PIN_TO_CHANNEL = {
 }


-def validate_adc_pin(value):
+def validate_adc_pin(value: Any) -> ConfigType | str:
    if str(value).upper() == "VCC":
        if CORE.is_rp2:
            return pins.internal_gpio_input_pin_schema(29)
        return cv.only_on([PLATFORM_ESP8266])("VCC")

+    # Deprecated in favour of the `internal_temperature` platform, remove before 2027.2.0
    if str(value).upper() == "TEMPERATURE":
        return cv.only_on_rp2("TEMPERATURE")

@@ -3,7 +3,7 @@

 namespace esphome::adc {

-static const char *const TAG = "adc.common";
+static const char *const TAG = "adc";

 const LogString *sampling_mode_to_str(SamplingMode mode) {
  switch (mode) {
@@ -6,7 +6,7 @@

 namespace esphome::adc {

-static const char *const TAG = "adc.esp32";
+static const char *const TAG = "adc";

 adc_oneshot_unit_handle_t ADCSensor::shared_adc_handles[2] = {nullptr, nullptr};

@@ -13,7 +13,7 @@ ADC_MODE(ADC_VCC)

 namespace esphome::adc {

-static const char *const TAG = "adc.esp8266";
+static const char *const TAG = "adc";

 void ADCSensor::setup() {
 #ifndef USE_ADC_SENSOR_VCC
@@ -5,7 +5,7 @@

 namespace esphome::adc {

-static const char *const TAG = "adc.libretiny";
+static const char *const TAG = "adc";

 void ADCSensor::setup() {
 #ifndef USE_ADC_SENSOR_VCC
@@ -17,7 +17,26 @@

 namespace esphome::adc {

-static const char *const TAG = "adc.rp2";
+static const char *const TAG = "adc";
+
+// The on-die temperature sensor sits on the last ADC channel: input 4 on RP2040
+// and RP2350A, but input 8 on RP2350B, which has eight external channels rather
+// than four.
+//
+// This deliberately does not use the SDK's ADC_TEMPERATURE_CHANNEL_NUM. That
+// derives from NUM_ADC_CHANNELS, which <pico.h> settles from a board header, and
+// arduino-pico supplies a fixed B-die one for every RP2350 build. The real die
+// is only declared later, by the variant's pins_arduino.h, so the SDK constant
+// reads 8 on A-die boards. PICO_RP2350A itself is correct by the time this file
+// is compiled, on both arduino-pico and pico-sdk builds.
+#if defined(PICO_RP2350) && !defined(PICO_RP2350A)
+#error "PICO_RP2350A is not defined, so the RP2350 die is unknown and the temperature ADC channel cannot be chosen"
+#endif
+#if defined(PICO_RP2350) && !PICO_RP2350A
+static constexpr uint8_t TEMPERATURE_ADC_INPUT = 8;
+#else
+static constexpr uint8_t TEMPERATURE_ADC_INPUT = 4;
+#endif

 void ADCSensor::setup() {
  static bool initialized = false;
@@ -52,7 +71,7 @@ float ADCSensor::sample() {
  if (this->is_temperature_) {
    adc_set_temp_sensor_enabled(true);
    delay(1);
-    adc_select_input(4);
+    adc_select_input(TEMPERATURE_ADC_INPUT);

    for (uint8_t sample = 0; sample < this->sample_count_; sample++) {
      raw = adc_read();
@@ -7,7 +7,7 @@

 namespace esphome::adc {

-static const char *const TAG = "adc.zephyr";
+static const char *const TAG = "adc";

 void ADCSensor::setup() {
  if (!adc_is_ready_dt(this->channel_)) {
@@ -52,7 +52,7 @@ _attenuation = cv.enum(ATTENUATION_MODES, lower=True)
 _sampling_mode = cv.enum(SAMPLING_MODES, lower=True)


-def validate_config(config):
+def validate_config(config: ConfigType) -> ConfigType:
    if config[CONF_RAW] and config.get(CONF_ATTENUATION, None) == "auto":
        raise cv.Invalid("Automatic attenuation cannot be used when raw output is set")

@@ -67,6 +67,13 @@ def validate_config(config):
        # Alter value here so `config` command prints the recommended change
        config[CONF_ATTENUATION] = _attenuation("12db")

+    # Remove before 2027.2.0
+    if config[CONF_PIN] == "TEMPERATURE":
+        _LOGGER.warning(
+            "[adc] `pin: TEMPERATURE` is deprecated, use the `internal_temperature` "
+            "sensor platform instead. Will be removed in 2027.2.0"
+        )
+
    return config


@@ -113,7 +120,7 @@ CONFIG_SCHEMA = cv.All(
 CONF_ADC_CHANNEL_ID = "adc_channel_id"


-def _overlay_io_channels():
+def _overlay_io_channels() -> str:
    channel_count = CORE.data[CONF_ADC_CHANNEL_ID]
    entries = ", ".join(f"<&adc {channel_id}>" for channel_id in range(channel_count))
    return f"""
@@ -125,7 +132,7 @@ def _overlay_io_channels():
            """


-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
    var = cg.new_Pvariable(config[CONF_ID])
    await cg.register_component(var, config)
    await sensor.register_sensor(var, config)
@@ -133,6 +140,7 @@ async def to_code(config):
    if config[CONF_PIN] == "VCC":
        cg.add_define("USE_ADC_SENSOR_VCC")
    elif config[CONF_PIN] == "TEMPERATURE":
+        # Remove before 2027.2.0
        cg.add(var.set_is_temperature())
    elif not CORE.is_nrf52 or config[CONF_PIN][CONF_NUMBER] not in EXTRA_ADC:
        pin = await cg.gpio_pin_expression(config[CONF_PIN])
@@ -17,6 +17,9 @@ from esphome.const import (
    UNIT_OHM,
    UNIT_PARTS_PER_BILLION,
 )
+from esphome.core import ID
+from esphome.cpp_generator import MockObj, TemplateArgsType
+from esphome.types import ConfigType

 CONF_RESISTANCE = "resistance"

@@ -62,7 +65,7 @@ CONFIG_SCHEMA = (
 FINAL_VALIDATE_SCHEMA = i2c.final_validate_device_schema("ags10", max_frequency="15khz")


-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
    var = cg.new_Pvariable(config[CONF_ID])
    await cg.register_component(var, config)
    await i2c.register_i2c_device(var, config)
@@ -94,7 +97,12 @@ AGS10_NEW_I2C_ADDRESS_SCHEMA = cv.maybe_simple_value(
    AGS10_NEW_I2C_ADDRESS_SCHEMA,
    synchronous=True,
 )
-async def ags10newi2caddress_to_code(config, action_id, template_arg, args):
+async def ags10newi2caddress_to_code(
+    config: ConfigType,
+    action_id: ID,
+    template_arg: cg.TemplateArguments,
+    args: TemplateArgsType,
+) -> MockObj:
    var = cg.new_Pvariable(action_id, template_arg)
    await cg.register_parented(var, config[CONF_ID])
    address = await cg.templatable(config[CONF_ADDRESS], args, cg.uint8)
@@ -126,7 +134,12 @@ AGS10_SET_ZERO_POINT_SCHEMA = cv.Schema(
    AGS10_SET_ZERO_POINT_SCHEMA,
    synchronous=True,
 )
-async def ags10setzeropoint_to_code(config, action_id, template_arg, args):
+async def ags10setzeropoint_to_code(
+    config: ConfigType,
+    action_id: ID,
+    template_arg: cg.TemplateArguments,
+    args: TemplateArgsType,
+) -> MockObj:
    var = cg.new_Pvariable(action_id, template_arg)
    await cg.register_parented(var, config[CONF_ID])
    mode = await cg.templatable(
@@ -4,6 +4,9 @@ from esphome.components import i2c
 from esphome.components.audio_dac import AudioDac
 import esphome.config_validation as cv
 from esphome.const import CONF_ID, CONF_MODE
+from esphome.core import ID
+from esphome.cpp_generator import MockObj, TemplateArgsType
+from esphome.types import ConfigType

 CODEOWNERS = ["@kbx81"]
 DEPENDENCIES = ["i2c"]
@@ -39,7 +42,12 @@ SET_AUTO_MUTE_ACTION_SCHEMA = cv.maybe_simple_value(
    SET_AUTO_MUTE_ACTION_SCHEMA,
    synchronous=True,
 )
-async def aic3204_set_volume_to_code(config, action_id, template_arg, args):
+async def aic3204_set_volume_to_code(
+    config: ConfigType,
+    action_id: ID,
+    template_arg: cg.TemplateArguments,
+    args: TemplateArgsType,
+) -> MockObj:
    paren = await cg.get_variable(config[CONF_ID])
    var = cg.new_Pvariable(action_id, template_arg, paren)

@@ -49,7 +57,7 @@ async def aic3204_set_volume_to_code(config, action_id, template_arg, args):
    return var


-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
    var = cg.new_Pvariable(config[CONF_ID])
    await cg.register_component(var, config)
    await i2c.register_i2c_device(var, config)
@@ -1,23 +1,26 @@
 import esphome.codegen as cg
-from esphome.components import esp32_ble_tracker
+from esphome.components import ble_device_base
 import esphome.config_validation as cv
 from esphome.const import CONF_ID

-DEPENDENCIES = ["esp32_ble_tracker"]
+AUTO_LOAD = ["ble_device_base"]
 CODEOWNERS = ["@jeromelaban"]

 airthings_ble_ns = cg.esphome_ns.namespace("airthings_ble")
 AirthingsListener = airthings_ble_ns.class_(
-    "AirthingsListener", esp32_ble_tracker.ESPBTDeviceListener
+    "AirthingsListener", ble_device_base.ESPBTDeviceListener
 )

-CONFIG_SCHEMA = cv.Schema(
-    {
-        cv.GenerateID(): cv.declare_id(AirthingsListener),
-    }
-).extend(esp32_ble_tracker.ESP_BLE_DEVICE_SCHEMA)
+CONFIG_SCHEMA = cv.All(
+    ble_device_base.rename_legacy_hub_id("airthings_ble"),
+    cv.Schema(
+        {
+            cv.GenerateID(): cv.declare_id(AirthingsListener),
+        }
+    ).extend(ble_device_base.BLE_DEVICE_SCHEMA),
+)


 async def to_code(config):
    var = cg.new_Pvariable(config[CONF_ID])
-    await esp32_ble_tracker.register_ble_device(var, config)
+    await ble_device_base.register_ble_device(var, config)
@@ -2,15 +2,13 @@
 #include "esphome/core/log.h"
 #include <cinttypes>

-#ifdef USE_ESP32
-
 namespace esphome::airthings_ble {

 static const char *const TAG = "airthings_ble";

-bool AirthingsListener::parse_device(const esp32_ble_tracker::ESPBTDevice &device) {
+bool AirthingsListener::parse_device(const ble_device_base::ESPBTDevice &device) {
  for (auto &it : device.get_manufacturer_datas()) {
-    if (it.uuid == esp32_ble_tracker::ESPBTUUID::from_uint32(0x0334)) {
+    if (it.uuid == ble_device_base::ESPBTUUID::from_uint32(0x0334)) {
      if (it.data.size() < 4)
        continue;

@@ -29,5 +27,3 @@ bool AirthingsListener::parse_device(const esp32_ble_tracker::ESPBTDevice &devic
 }

 }  // namespace esphome::airthings_ble
-
-#endif
@@ -1,17 +1,13 @@
 #pragma once

-#ifdef USE_ESP32
-
 #include "esphome/core/component.h"
-#include "esphome/components/esp32_ble_tracker/esp32_ble_tracker.h"
+#include "esphome/components/ble_device_base/ble_device.h"

 namespace esphome::airthings_ble {

-class AirthingsListener final : public esp32_ble_tracker::ESPBTDeviceListener {
+class AirthingsListener final : public ble_device_base::ESPBTDeviceListener {
 public:
-  bool parse_device(const esp32_ble_tracker::ESPBTDevice &device) override;
+  bool parse_device(const ble_device_base::ESPBTDevice &device) override;
 };

 }  // namespace esphome::airthings_ble
-
-#endif
@@ -13,8 +13,13 @@
 import esphome.components.image as espImage
 import esphome.config_validation as cv

+from . import image as animation_image
 from .image import ANIMATION_CONFIG_SCHEMA, setup_animation

+# The deprecated top-level `animation:` shim gets the same batched
+# downloads as the `image:` platform form.
+PREFETCH_FILES = animation_image.PREFETCH_FILES
+
 AUTO_LOAD = ["image", "file"]
 CODEOWNERS = ["@syndlex"]
 DEPENDENCIES = ["display"]
@@ -1,13 +1,20 @@
 from esphome import automation
 import esphome.codegen as cg
 from esphome.components.const import CONF_LOOP
+from esphome.components.file import image as file_image
 from esphome.components.file.image import image_schema, write_image
 from esphome.components.image import Image_, validate_settings
 import esphome.config_validation as cv
 from esphome.const import CONF_ID, CONF_REPEAT
+from esphome.core import ID
+from esphome.cpp_generator import MockObj, TemplateArgsType
 from esphome.types import ConfigType

 CODEOWNERS = ["@syndlex"]
+
+# The animation platform shares the file platform's remote file handling,
+# including its batch-download hook.
+PREFETCH_FILES = file_image.PREFETCH_FILES
 AUTO_LOAD = ["file"]
 DEPENDENCIES = ["display"]

@@ -74,7 +81,12 @@ SET_FRAME_SCHEMA = cv.Schema(
@automation.register_action(
    "animation.set_frame", SetFrameAction, SET_FRAME_SCHEMA, synchronous=True
 )
-async def animation_action_to_code(config, action_id, template_arg, args):
+async def animation_action_to_code(
+    config: ConfigType,
+    action_id: ID,
+    template_arg: cg.TemplateArguments,
+    args: TemplateArgsType,
+) -> MockObj:
    paren = await cg.get_variable(config[CONF_ID])
    var = cg.new_Pvariable(action_id, template_arg, paren)

@@ -2,6 +2,7 @@ import esphome.codegen as cg
 from esphome.components import i2c
 import esphome.config_validation as cv
 from esphome.const import CONF_ID
+from esphome.types import ConfigType

 DEPENDENCIES = ["i2c"]
 MULTI_CONF = True
@@ -57,7 +58,7 @@ CONFIG_SCHEMA = (
 )


-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
    var = cg.new_Pvariable(config[CONF_ID])
    await cg.register_component(var, config)
    await i2c.register_i2c_device(var, config)
@@ -2,6 +2,7 @@ import esphome.codegen as cg
 from esphome.components import binary_sensor
 import esphome.config_validation as cv
 from esphome.const import CONF_DIRECTION, DEVICE_CLASS_MOVING
+from esphome.types import ConfigType

 from . import APDS9960, CONF_APDS9960_ID

@@ -19,7 +20,7 @@ CONFIG_SCHEMA = binary_sensor.binary_sensor_schema(
 )


-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
    hub = await cg.get_variable(config[CONF_APDS9960_ID])
    var = await binary_sensor.new_binary_sensor(config)
    func = getattr(hub, f"set_{config[CONF_DIRECTION]}_direction_binary_sensor")
@@ -7,6 +7,7 @@ from esphome.const import (
    STATE_CLASS_MEASUREMENT,
    UNIT_PERCENT,
 )
+from esphome.types import ConfigType

 from . import APDS9960, CONF_APDS9960_ID

@@ -27,7 +28,7 @@ CONFIG_SCHEMA = sensor.sensor_schema(
 )


-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
    hub = await cg.get_variable(config[CONF_APDS9960_ID])
    var = await sensor.new_sensor(config)
    func = getattr(hub, f"set_{config[CONF_TYPE]}_sensor")
@@ -1,5 +1,6 @@
 import base64
 import logging
+from typing import Any

 from esphome import automation
 from esphome.automation import Condition
@@ -13,6 +14,7 @@ from esphome.const import (
    CONF_CAPTURE_RESPONSE,
    CONF_DATA,
    CONF_DATA_TEMPLATE,
+    CONF_ENCRYPTION,
    CONF_EVENT,
    CONF_ID,
    CONF_KEY,
@@ -102,7 +104,6 @@ SERVICE_ARG_FALLBACK_TYPES: dict[str, MockObj] = {
        for name, t in _SERVICE_ARG_SCALAR_TYPES.items()
    },
 }
-CONF_ENCRYPTION = "encryption"
 CONF_BATCH_DELAY = "batch_delay"
 CONF_CUSTOM_SERVICES = "custom_services"
 CONF_HOMEASSISTANT_SERVICES = "homeassistant_services"
@@ -129,7 +130,7 @@ def _register_provisioning_source(config: ConfigType) -> ConfigType:
    return config


-def validate_encryption_key(value):
+def validate_encryption_key(value: Any) -> str:
    value = cv.string_strict(value)
    try:
        decoded = base64.b64decode(value, validate=True)
@@ -217,7 +218,7 @@ def _auto_detect_supports_response(config: ConfigType) -> ConfigType:
    return config


-def _validate_supports_response(value):
+def _validate_supports_response(value: Any) -> str:
    """Validate supports_response after auto-detection has set the value."""
    return cv.enum(SUPPORTS_RESPONSE_OPTIONS, lower=True)(value)

@@ -256,7 +257,7 @@ ENCRYPTION_SCHEMA = cv.Schema(
 )


-def _encryption_schema(config):
+def _encryption_schema(config: ConfigType | None) -> ConfigType:
    if config is None:
        config = {}
    return ENCRYPTION_SCHEMA(config)
@@ -393,7 +394,7 @@ async def to_code(config: ConfigType) -> None:

    if actions := config.get(CONF_ACTIONS, []):
        # Collect all triggers first, then register all at once with initializer_list
-        triggers: list[cg.Pvariable] = []
+        triggers: list[cg.MockObj] = []
        for conf in actions:
            func_args: list[tuple[MockObj, str]] = []
            service_template_args: list[MockObj] = []  # User service argument types
@@ -497,7 +498,7 @@ async def to_code(config: ConfigType) -> None:
            # and plaintext disabled. Only a factory reset can remove it.
            cg.add_define("USE_API_PLAINTEXT")
        cg.add_define("USE_API_NOISE")
-        cg.add_library("esphome/noise-c", "0.1.11")
+        cg.add_library("esphome/noise-c", "0.1.21")
        # Enable optimized memzero/memcmp in libsodium instead of volatile byte loops
        cg.add_build_flag("-DHAVE_WEAK_SYMBOLS=1")
        cg.add_build_flag("-DHAVE_INLINE_ASM=1")
@@ -581,7 +582,7 @@ async def homeassistant_service_to_code(
    action_id: ID,
    template_arg: cg.TemplateArguments,
    args: TemplateArgsType,
-):
+) -> MockObj:
    cg.add_define("USE_API_HOMEASSISTANT_SERVICES")
    serv = await cg.get_variable(config[CONF_ID])
    var = cg.new_Pvariable(action_id, template_arg, serv, False)
@@ -647,7 +648,7 @@ async def homeassistant_service_to_code(
    return var


-def validate_homeassistant_event(value):
+def validate_homeassistant_event(value: Any) -> str:
    value = cv.string(value)
    if not value.startswith("esphome."):
        raise cv.Invalid(
@@ -676,7 +677,12 @@ HOMEASSISTANT_EVENT_ACTION_SCHEMA = cv.Schema(
    HOMEASSISTANT_EVENT_ACTION_SCHEMA,
    synchronous=True,
 )
-async def homeassistant_event_to_code(config, action_id, template_arg, args):
+async def homeassistant_event_to_code(
+    config: ConfigType,
+    action_id: ID,
+    template_arg: cg.TemplateArguments,
+    args: TemplateArgsType,
+) -> MockObj:
    cg.add_define("USE_API_HOMEASSISTANT_SERVICES")
    serv = await cg.get_variable(config[CONF_ID])
    var = cg.new_Pvariable(action_id, template_arg, serv, True)
@@ -724,7 +730,12 @@ HOMEASSISTANT_TAG_SCANNED_ACTION_SCHEMA = cv.maybe_simple_value(
    HOMEASSISTANT_TAG_SCANNED_ACTION_SCHEMA,
    synchronous=True,
 )
-async def homeassistant_tag_scanned_to_code(config, action_id, template_arg, args):
+async def homeassistant_tag_scanned_to_code(
+    config: ConfigType,
+    action_id: ID,
+    template_arg: cg.TemplateArguments,
+    args: TemplateArgsType,
+) -> MockObj:
    cg.add_define("USE_API_HOMEASSISTANT_SERVICES")
    serv = await cg.get_variable(config[CONF_ID])
    var = cg.new_Pvariable(action_id, template_arg, serv, True)
@@ -740,7 +751,7 @@ CONF_SUCCESS = "success"
 CONF_ERROR_MESSAGE = "error_message"


-def _validate_api_respond_data(config):
+def _validate_api_respond_data(config: ConfigType) -> ConfigType:
    """Set flag during validation so AUTO_LOAD can include json component."""
    if CONF_DATA in config:
        CORE.data.setdefault(DOMAIN, {})[CONF_CAPTURE_RESPONSE] = True
@@ -824,7 +835,12 @@ API_CONNECTED_CONDITION_SCHEMA = cv.Schema(
@automation.register_condition(
    "api.connected", APIConnectedCondition, API_CONNECTED_CONDITION_SCHEMA
 )
-async def api_connected_to_code(config, condition_id, template_arg, args):
+async def api_connected_to_code(
+    config: ConfigType,
+    condition_id: ID,
+    template_arg: cg.TemplateArguments,
+    args: TemplateArgsType,
+) -> MockObj:
    var = cg.new_Pvariable(condition_id, template_arg)
    templ = await cg.templatable(config[CONF_STATE_SUBSCRIPTION_ONLY], args, cg.bool_)
    cg.add(var.set_state_subscription_only(templ))
@@ -19,6 +19,7 @@ service APIConnection {
  rpc device_info (DeviceInfoRequest) returns (DeviceInfoResponse) {
    option (needs_authentication) = false;
  }
+  rpc device_capabilities (DeviceCapabilitiesRequest) returns (DeviceCapabilitiesResponse) {}
  rpc list_entities (ListEntitiesRequest) returns (void) {}
  rpc subscribe_states (SubscribeStatesRequest) returns (void) {}
  rpc subscribe_logs (SubscribeLogsRequest) returns (void) {}
@@ -243,6 +244,12 @@ message SerialProxyInfo {
 //   model = 127 (core/config.BOARD_MAX_LENGTH, validated in platform schemas)
 //   project_name/project_version = 127 (core/config.PROJECT_MAX_LENGTH)
 //   suggested_area = 120 (core/config.FRIENDLY_NAME_MAX_LEN via AREA_SCHEMA)
+//
+// Some fields below are marked "Superseded by DeviceCapabilitiesResponse". They
+// have moved to that message as of API 1.15, but are still sent here so that
+// older clients keep working. Do NOT mark them (deprecated) until the removal
+// release: in this repo (deprecated) makes the generator drop the field
+// entirely, so the device would stop sending it.
 message DeviceInfoResponse {
  option (id) = 10;
  option (source) = SOURCE_SERVER;
@@ -280,6 +287,8 @@ message DeviceInfoResponse {

  // Deprecated in API version 1.9
  uint32 legacy_bluetooth_proxy_version = 11 [deprecated=true, (field_ifdef) = "USE_BLUETOOTH_PROXY"];
+
+  // Superseded by DeviceCapabilitiesResponse.bluetooth_proxy as of API 1.15.
  uint32 bluetooth_proxy_feature_flags = 15 [(field_ifdef) = "USE_BLUETOOTH_PROXY"];

  string manufacturer = 12 [(max_data_length) = 20, (force) = true];
@@ -288,11 +297,14 @@ message DeviceInfoResponse {

  // Deprecated in API version 1.10
  uint32 legacy_voice_assistant_version = 14 [deprecated=true, (field_ifdef) = "USE_VOICE_ASSISTANT"];
+
+  // Superseded by DeviceCapabilitiesResponse.voice_assistant as of API 1.15.
  uint32 voice_assistant_feature_flags = 17 [(field_ifdef) = "USE_VOICE_ASSISTANT"];

  string suggested_area = 16 [(max_data_length) = 120, (force) = true, (field_ifdef) = "USE_AREAS"];

  // The Bluetooth mac address of the device. For example "AC:BC:32:89:0E:AA"
+  // Superseded by DeviceCapabilitiesResponse.bluetooth_proxy.mac_address as of API 1.15.
  string bluetooth_mac_address = 18 [(max_data_length) = 17, (force) = true, (field_ifdef) = "USE_BLUETOOTH_PROXY"];

  // Supports receiving and saving api encryption key
@@ -305,10 +317,13 @@ message DeviceInfoResponse {
  AreaInfo area = 22 [(field_ifdef) = "USE_AREAS"];

  // Indicates if Z-Wave proxy support is available and features supported
+  // Superseded by DeviceCapabilitiesResponse.zwave_proxy as of API 1.15.
  uint32 zwave_proxy_feature_flags = 23 [(field_ifdef) = "USE_ZWAVE_PROXY"];
+  // Superseded by DeviceCapabilitiesResponse.zwave_proxy as of API 1.15.
  uint32 zwave_home_id = 24 [(field_ifdef) = "USE_ZWAVE_PROXY"];

  // Serial proxy instance metadata
+  // Superseded by DeviceCapabilitiesResponse.serial_proxies as of API 1.15.
  repeated SerialProxyInfo serial_proxies = 25 [(field_ifdef) = "USE_SERIAL_PROXY", (fixed_array_size_define) = "SERIAL_PROXY_COUNT"];

  // Device is unprovisioned and accepts Noise handshakes with the well-known
@@ -317,6 +332,63 @@ message DeviceInfoResponse {
  bool api_encryption_provisionable = 26 [(field_ifdef) = "USE_API_NOISE"];
 }

+// ==================== DEVICE CAPABILITIES ====================
+
+// Asks the device which optional features it supports.
+//
+// This message exists so that DeviceInfoResponse does not have to keep growing
+// a flat list of feature flags. DeviceInfoResponse is served before
+// authentication, so it is limited to identity information. Capabilities are
+// only served on an authenticated connection (encrypted as well, when
+// encryption is configured).
+//
+// Clients that see api_version >= 1.15 should read these values from
+// DeviceCapabilitiesResponse and ignore the matching DeviceInfoResponse fields.
+// Older clients keep reading DeviceInfoResponse, which still carries the same
+// values, so this is not a breaking change.
+message DeviceCapabilitiesRequest {
+  option (id) = 149;
+  option (source) = SOURCE_CLIENT;
+  // Empty
+}
+
+// Each feature gets its own sub-message so that it can gain fields over time
+// without crowding the top-level field numbering.
+//
+// Note: a sub-message whose fields are all at their default value is not sent
+// at all, so the presence of a sub-message is not a reliable test for "this
+// feature is compiled in". Clients should test a value inside it, for example
+// a non-zero feature_flags, exactly as they do today with DeviceInfoResponse.
+
+message BluetoothProxyCapabilities {
+  // Bitmask of the features this proxy supports
+  uint32 feature_flags = 1;
+  // The Bluetooth mac address of the device. For example "AC:BC:32:89:0E:AA"
+  string mac_address = 2 [(max_data_length) = 17, (force) = true];
+}
+
+message VoiceAssistantCapabilities {
+  // Bitmask of the features this voice assistant supports
+  uint32 feature_flags = 1;
+}
+
+message ZWaveProxyCapabilities {
+  // Bitmask of the features this proxy supports
+  uint32 feature_flags = 1;
+  uint32 home_id = 2;
+}
+
+message DeviceCapabilitiesResponse {
+  option (id) = 150;
+  option (source) = SOURCE_SERVER;
+
+  BluetoothProxyCapabilities bluetooth_proxy = 1 [(field_ifdef) = "USE_BLUETOOTH_PROXY"];
+  VoiceAssistantCapabilities voice_assistant = 2 [(field_ifdef) = "USE_VOICE_ASSISTANT"];
+  ZWaveProxyCapabilities zwave_proxy = 3 [(field_ifdef) = "USE_ZWAVE_PROXY"];
+  repeated SerialProxyInfo serial_proxies = 4
+      [(field_ifdef) = "USE_SERIAL_PROXY", (fixed_array_size_define) = "SERIAL_PROXY_COUNT"];
+}
+
 message ListEntitiesRequest {
  option (id) = 11;
  option (source) = SOURCE_CLIENT;
@@ -1688,7 +1760,7 @@ enum BluetoothDeviceRequestType {
 message BluetoothDeviceRequest {
  option (id) = 68;
  option (source) = SOURCE_CLIENT;
-  option (ifdef) = "USE_BLUETOOTH_PROXY";
+  option (ifdef) = "USE_BLUETOOTH_PROXY_CONNECTIONS";

  uint64 address = 1;
  BluetoothDeviceRequestType request_type = 2;
@@ -1699,7 +1771,7 @@ message BluetoothDeviceRequest {
 message BluetoothDeviceConnectionResponse {
  option (id) = 69;
  option (source) = SOURCE_SERVER;
-  option (ifdef) = "USE_BLUETOOTH_PROXY";
+  option (ifdef) = "USE_BLUETOOTH_PROXY_CONNECTIONS";

  uint64 address = 1;
  bool connected = 2;
@@ -1710,7 +1782,7 @@ message BluetoothDeviceConnectionResponse {
 message BluetoothGATTGetServicesRequest {
  option (id) = 70;
  option (source) = SOURCE_CLIENT;
-  option (ifdef) = "USE_BLUETOOTH_PROXY";
+  option (ifdef) = "USE_BLUETOOTH_PROXY_CONNECTIONS";

  uint64 address = 1;
 }
@@ -1754,7 +1826,7 @@ message BluetoothGATTService {
 message BluetoothGATTGetServicesResponse {
  option (id) = 71;
  option (source) = SOURCE_SERVER;
-  option (ifdef) = "USE_BLUETOOTH_PROXY";
+  option (ifdef) = "USE_BLUETOOTH_PROXY_CONNECTIONS";

  uint64 address = 1;
  repeated BluetoothGATTService services = 2;
@@ -1763,7 +1835,7 @@ message BluetoothGATTGetServicesResponse {
 message BluetoothGATTGetServicesDoneResponse {
  option (id) = 72;
  option (source) = SOURCE_SERVER;
-  option (ifdef) = "USE_BLUETOOTH_PROXY";
+  option (ifdef) = "USE_BLUETOOTH_PROXY_CONNECTIONS";

  uint64 address = 1;
 }
@@ -1771,7 +1843,7 @@ message BluetoothGATTGetServicesDoneResponse {
 message BluetoothGATTReadRequest {
  option (id) = 73;
  option (source) = SOURCE_CLIENT;
-  option (ifdef) = "USE_BLUETOOTH_PROXY";
+  option (ifdef) = "USE_BLUETOOTH_PROXY_CONNECTIONS";

  uint64 address = 1;
  uint32 handle = 2;
@@ -1780,7 +1852,7 @@ message BluetoothGATTReadRequest {
 message BluetoothGATTReadResponse {
  option (id) = 74;
  option (source) = SOURCE_SERVER;
-  option (ifdef) = "USE_BLUETOOTH_PROXY";
+  option (ifdef) = "USE_BLUETOOTH_PROXY_CONNECTIONS";

  uint64 address = 1;
  uint32 handle = 2;
@@ -1792,7 +1864,7 @@ message BluetoothGATTReadResponse {
 message BluetoothGATTWriteRequest {
  option (id) = 75;
  option (source) = SOURCE_CLIENT;
-  option (ifdef) = "USE_BLUETOOTH_PROXY";
+  option (ifdef) = "USE_BLUETOOTH_PROXY_CONNECTIONS";

  uint64 address = 1;
  uint32 handle = 2;
@@ -1804,7 +1876,7 @@ message BluetoothGATTWriteRequest {
 message BluetoothGATTReadDescriptorRequest {
  option (id) = 76;
  option (source) = SOURCE_CLIENT;
-  option (ifdef) = "USE_BLUETOOTH_PROXY";
+  option (ifdef) = "USE_BLUETOOTH_PROXY_CONNECTIONS";

  uint64 address = 1;
  uint32 handle = 2;
@@ -1813,7 +1885,7 @@ message BluetoothGATTReadDescriptorRequest {
 message BluetoothGATTWriteDescriptorRequest {
  option (id) = 77;
  option (source) = SOURCE_CLIENT;
-  option (ifdef) = "USE_BLUETOOTH_PROXY";
+  option (ifdef) = "USE_BLUETOOTH_PROXY_CONNECTIONS";

  uint64 address = 1;
  uint32 handle = 2;
@@ -1824,7 +1896,7 @@ message BluetoothGATTWriteDescriptorRequest {
 message BluetoothGATTNotifyRequest {
  option (id) = 78;
  option (source) = SOURCE_CLIENT;
-  option (ifdef) = "USE_BLUETOOTH_PROXY";
+  option (ifdef) = "USE_BLUETOOTH_PROXY_CONNECTIONS";

  uint64 address = 1;
  uint32 handle = 2;
@@ -1834,7 +1906,7 @@ message BluetoothGATTNotifyRequest {
 message BluetoothGATTNotifyDataResponse {
  option (id) = 79;
  option (source) = SOURCE_SERVER;
-  option (ifdef) = "USE_BLUETOOTH_PROXY";
+  option (ifdef) = "USE_BLUETOOTH_PROXY_CONNECTIONS";

  uint64 address = 1;
  uint32 handle = 2;
@@ -1845,13 +1917,13 @@ message BluetoothGATTNotifyDataResponse {
 message SubscribeBluetoothConnectionsFreeRequest {
  option (id) = 80;
  option (source) = SOURCE_CLIENT;
-  option (ifdef) = "USE_BLUETOOTH_PROXY";
+  option (ifdef) = "USE_BLUETOOTH_PROXY_CONNECTIONS";
 }

 message BluetoothConnectionsFreeResponse {
  option (id) = 81;
  option (source) = SOURCE_SERVER;
-  option (ifdef) = "USE_BLUETOOTH_PROXY";
+  option (ifdef) = "USE_BLUETOOTH_PROXY_CONNECTIONS";

  uint32 free = 1;
  uint32 limit = 2;
@@ -1864,7 +1936,7 @@ message BluetoothConnectionsFreeResponse {
 message BluetoothGATTErrorResponse {
  option (id) = 82;
  option (source) = SOURCE_SERVER;
-  option (ifdef) = "USE_BLUETOOTH_PROXY";
+  option (ifdef) = "USE_BLUETOOTH_PROXY_CONNECTIONS";

  uint64 address = 1;
  uint32 handle = 2;
@@ -1874,7 +1946,7 @@ message BluetoothGATTErrorResponse {
 message BluetoothGATTWriteResponse {
  option (id) = 83;
  option (source) = SOURCE_SERVER;
-  option (ifdef) = "USE_BLUETOOTH_PROXY";
+  option (ifdef) = "USE_BLUETOOTH_PROXY_CONNECTIONS";

  uint64 address = 1;
  uint32 handle = 2;
@@ -1883,7 +1955,7 @@ message BluetoothGATTWriteResponse {
 message BluetoothGATTNotifyResponse {
  option (id) = 84;
  option (source) = SOURCE_SERVER;
-  option (ifdef) = "USE_BLUETOOTH_PROXY";
+  option (ifdef) = "USE_BLUETOOTH_PROXY_CONNECTIONS";

  uint64 address = 1;
  uint32 handle = 2;
@@ -1892,7 +1964,7 @@ message BluetoothGATTNotifyResponse {
 message BluetoothDevicePairingResponse {
  option (id) = 85;
  option (source) = SOURCE_SERVER;
-  option (ifdef) = "USE_BLUETOOTH_PROXY";
+  option (ifdef) = "USE_BLUETOOTH_PROXY_CONNECTIONS";

  uint64 address = 1;
  bool paired = 2;
@@ -1902,7 +1974,7 @@ message BluetoothDevicePairingResponse {
 message BluetoothDeviceUnpairingResponse {
  option (id) = 86;
  option (source) = SOURCE_SERVER;
-  option (ifdef) = "USE_BLUETOOTH_PROXY";
+  option (ifdef) = "USE_BLUETOOTH_PROXY_CONNECTIONS";

  uint64 address = 1;
  bool success = 2;
@@ -1918,7 +1990,7 @@ message UnsubscribeBluetoothLEAdvertisementsRequest {
 message BluetoothDeviceClearCacheResponse {
  option (id) = 88;
  option (source) = SOURCE_SERVER;
-  option (ifdef) = "USE_BLUETOOTH_PROXY";
+  option (ifdef) = "USE_BLUETOOTH_PROXY_CONNECTIONS";

  uint64 address = 1;
  bool success = 2;
@@ -2735,7 +2807,7 @@ message SerialProxyRequestResponse {
 message BluetoothSetConnectionParamsRequest {
  option (id) = 145;
  option (source) = SOURCE_CLIENT;
-  option (ifdef) = "USE_BLUETOOTH_PROXY";
+  option (ifdef) = "USE_BLUETOOTH_PROXY_CONNECTIONS";

  uint64 address = 1;
  uint32 min_interval = 2;  // units of 1.25ms
@@ -2747,7 +2819,7 @@ message BluetoothSetConnectionParamsRequest {
 message BluetoothSetConnectionParamsResponse {
  option (id) = 146;
  option (source) = SOURCE_SERVER;
-  option (ifdef) = "USE_BLUETOOTH_PROXY";
+  option (ifdef) = "USE_BLUETOOTH_PROXY_CONNECTIONS";

  uint64 address = 1;
  int32 error = 2;
@@ -23,6 +23,7 @@
 #include "esphome/core/application.h"
 #include "esphome/core/entity_base.h"
 #include "esphome/core/hal.h"
+#include "esphome/core/helpers.h"
 #include "esphome/core/log.h"
 #include "esphome/core/version.h"
 #ifdef USE_PROVISIONING
@@ -88,6 +89,13 @@ static_assert(ESPHOME_DEVICE_NAME_MAX_LEN <= 31, "Update max_data_length for nam
 static_assert(ESPHOME_FRIENDLY_NAME_MAX_LEN <= 120, "Update max_data_length for friendly_name in api.proto");

 static const char *const TAG = "api.connection";
+
+#if ESPHOME_LOG_LEVEL >= ESPHOME_LOG_LEVEL_WARN
+void log_dropped_message(const char *tag, int line, const LogString *what) {
+  esp_log_printf_(ESPHOME_LOG_LEVEL_WARN, tag, line, ESPHOME_LOG_FORMAT("%s dropped, TCP buffer full"),
+                  LOG_STR_ARG(what));
+}
+#endif
 #ifdef USE_CAMERA
 static const int CAMERA_STOP_STREAM = 5000;
 #endif
@@ -152,11 +160,6 @@ APIConnection::APIConnection(std::unique_ptr<socket::Socket> sock, APIServer *pa
 #else
 #error "No frame helper defined"
 #endif
-#ifdef USE_CAMERA
-  if (camera::Camera::instance() != nullptr) {
-    this->image_reader_ = std::unique_ptr<camera::CameraImageReader>{camera::Camera::instance()->create_image_reader()};
-  }
-#endif
 }

 void APIConnection::start() {
@@ -1132,6 +1135,7 @@ void APIConnection::try_send_camera_image_() {
  if (!this->image_reader_)
    return;

+  const auto *cam = camera::Camera::instance();
  // Send as many chunks as possible without blocking
  while (this->image_reader_->available()) {
    if (!this->helper_->can_write_without_blocking())
@@ -1141,11 +1145,11 @@ void APIConnection::try_send_camera_image_() {
    bool done = this->image_reader_->available() == to_send;

    CameraImageResponse msg;
-    msg.key = camera::Camera::instance()->get_object_id_hash();
+    msg.key = cam->get_object_id_hash();
    msg.set_data(this->image_reader_->peek_data_buffer(), to_send);
    msg.done = done;
 #ifdef USE_DEVICES
-    msg.device_id = camera::Camera::instance()->get_device_id();
+    msg.device_id = cam->get_device_id();
 #endif

    if (!this->send_message(msg)) {
@@ -1161,15 +1165,19 @@ void APIConnection::try_send_camera_image_() {
 void APIConnection::set_camera_state(std::shared_ptr<camera::CameraImage> image) {
  if (!this->flags_.state_subscription)
    return;
-  if (!this->image_reader_)
+  if (this->image_reader_ && this->image_reader_->available())
    return;
-  if (this->image_reader_->available())
+  if (!image->was_requested_by(esphome::camera::API_REQUESTER) && !image->was_requested_by(esphome::camera::IDLE))
    return;
-  if (image->was_requested_by(esphome::camera::API_REQUESTER) || image->was_requested_by(esphome::camera::IDLE)) {
-    this->image_reader_->set_image(std::move(image));
-    // Try to send immediately to reduce latency
-    this->try_send_camera_image_();
+  if (!this->image_reader_) {
+    // Created on the first image this connection will send, so connections
+    // that never receive one never pay for a reader. Only a registered
+    // camera's listener can reach this, so instance() is non-null here.
+    this->image_reader_ = std::unique_ptr<camera::CameraImageReader>{camera::Camera::instance()->create_image_reader()};
  }
+  this->image_reader_->set_image(std::move(image));
+  // Try to send immediately to reduce latency
+  this->try_send_camera_image_();
 }
 uint16_t APIConnection::try_send_camera_info(EntityBase *entity, APIConnection *conn, uint32_t remaining_size) {
  auto *camera = static_cast<camera::Camera *>(entity);
@@ -1235,6 +1243,7 @@ void APIConnection::on_subscribe_bluetooth_le_advertisements_request(
 void APIConnection::on_unsubscribe_bluetooth_le_advertisements_request() {
  bluetooth_proxy::global_bluetooth_proxy->unsubscribe_api_connection(this);
 }
+#ifdef USE_BLUETOOTH_PROXY_CONNECTIONS
 void APIConnection::on_bluetooth_device_request(const BluetoothDeviceRequest &msg) {
  bluetooth_proxy::global_bluetooth_proxy->bluetooth_device_request(msg);
 }
@@ -1268,13 +1277,15 @@ void APIConnection::on_subscribe_bluetooth_connections_free_request() {
  }
 }

+void APIConnection::on_bluetooth_set_connection_params_request(const BluetoothSetConnectionParamsRequest &msg) {
+  bluetooth_proxy::global_bluetooth_proxy->bluetooth_set_connection_params(msg);
+}
+#endif
+
 void APIConnection::on_bluetooth_scanner_set_mode_request(const BluetoothScannerSetModeRequest &msg) {
  bluetooth_proxy::global_bluetooth_proxy->bluetooth_scanner_set_mode(
      msg.mode == enums::BluetoothScannerMode::BLUETOOTH_SCANNER_MODE_ACTIVE);
 }
-void APIConnection::on_bluetooth_set_connection_params_request(const BluetoothSetConnectionParamsRequest &msg) {
-  bluetooth_proxy::global_bluetooth_proxy->bluetooth_set_connection_params(msg);
-}
 #endif

 #ifdef USE_VOICE_ASSISTANT
@@ -1532,7 +1543,13 @@ void APIConnection::on_infrared_rf_transmit_raw_timings_request(const InfraredRF
 #endif

 #if defined(USE_IR_RF) || defined(USE_RADIO_FREQUENCY)
-void APIConnection::send_infrared_rf_receive_event(const InfraredRFReceiveEvent &msg) { this->send_message(msg); }
+void APIConnection::send_infrared_rf_receive_event(const InfraredRFReceiveEvent &msg) {
+  if (!this->send_message(msg)) {
+    // V: fires per decoded frame with no subscription gate, so a warning
+    // would flood the congested link it reports on.
+    ESP_LOGV(TAG, "IR/RF event dropped, TCP buffer full");
+  }
+}
 #endif

 #ifdef USE_SERIAL_PROXY
@@ -1574,7 +1591,9 @@ void APIConnection::on_serial_proxy_get_modem_pins_request(const SerialProxyGetM
  SerialProxyGetModemPinsResponse resp{};
  resp.instance = msg.instance;
  resp.line_states = proxies[msg.instance]->get_modem_pins();
-  this->send_message(resp);
+  if (!this->send_message(resp)) {
+    API_LOG_MSG_DROPPED(TAG, "Serial proxy response");
+  }
 }

 void APIConnection::on_serial_proxy_request(const SerialProxyRequest &msg) {
@@ -1606,7 +1625,9 @@ void APIConnection::on_serial_proxy_request(const SerialProxyRequest &msg) {
          resp.status = enums::SERIAL_PROXY_STATUS_ERROR;
          break;
      }
-      this->send_message(resp);
+      if (!this->send_message(resp)) {
+        API_LOG_MSG_DROPPED(TAG, "Serial proxy response");
+      }
      break;
    }
    default:
@@ -1615,7 +1636,11 @@ void APIConnection::on_serial_proxy_request(const SerialProxyRequest &msg) {
  }
 }

-void APIConnection::send_serial_proxy_data(const SerialProxyDataReceived &msg) { this->send_message(msg); }
+void APIConnection::send_serial_proxy_data(const SerialProxyDataReceived &msg) {
+  if (!this->send_message(msg)) {
+    ESP_LOGV(TAG, "Serial proxy data dropped, TCP buffer full");
+  }
+}
 #endif

 #ifdef USE_INFRARED
@@ -1735,7 +1760,7 @@ bool APIConnection::send_hello_response_(const HelloRequest &msg) {

  HelloResponse resp;
  resp.api_version_major = 1;
-  resp.api_version_minor = 14;
+  resp.api_version_minor = 15;
  // Send only the version string - the client only logs this for debugging and doesn't use it otherwise
  resp.server_info = ESPHOME_VERSION_REF;
  resp.name = StringRef(App.get_name());
@@ -1746,7 +1771,9 @@ bool APIConnection::send_hello_response_(const HelloRequest &msg) {
    // Acknowledge the hello so the client can read the server name, then request
    // disconnect with the reason. Authentication is intentionally not completed.
    this->log_client_(ESPHOME_LOG_LEVEL_WARN, LOG_STR("Provisioning closed; rejecting connection"));
-    this->send_message(resp);
+    if (!this->send_message(resp)) {
+      API_LOG_MSG_DROPPED(TAG, "Hello response");
+    }
    DisconnectRequest req;
    req.reason = enums::DISCONNECT_REASON_PROVISIONING_CLOSED;
    return this->send_message(req);
@@ -1771,9 +1798,8 @@ bool APIConnection::send_device_info_response_() {
 #ifdef USE_AREAS
  resp.suggested_area = StringRef(App.get_area());
 #endif
-  // Stack buffer for MAC address (XX:XX:XX:XX:XX:XX\0 = 18 bytes)
-  char mac_address[18];
-  uint8_t mac[6];
+  char mac_address[MAC_ADDRESS_PRETTY_BUFFER_SIZE];
+  uint8_t mac[MAC_ADDRESS_SIZE];
  get_mac_address_raw(mac);
  format_mac_addr_upper(mac, mac_address);
  resp.mac_address = StringRef(mac_address);
@@ -1849,8 +1875,7 @@ bool APIConnection::send_device_info_response_() {
 #endif
 #ifdef USE_BLUETOOTH_PROXY
  resp.bluetooth_proxy_feature_flags = bluetooth_proxy::global_bluetooth_proxy->get_feature_flags();
-  // Stack buffer for Bluetooth MAC address (XX:XX:XX:XX:XX:XX\0 = 18 bytes)
-  char bluetooth_mac[18];
+  char bluetooth_mac[MAC_ADDRESS_PRETTY_BUFFER_SIZE];
  bluetooth_proxy::global_bluetooth_proxy->get_bluetooth_mac_address_pretty(bluetooth_mac);
  resp.bluetooth_mac_address = StringRef(bluetooth_mac);
 #endif
@@ -1904,6 +1929,35 @@ bool APIConnection::send_device_info_response_() {

  return this->send_message(resp);
 }
+bool APIConnection::send_device_capabilities_response_() {
+  // These are the same values DeviceInfoResponse still reports for older clients. Keep the blocks
+  // below in sync with send_device_info_response_() until those copies are removed.
+  DeviceCapabilitiesResponse resp;
+#ifdef USE_BLUETOOTH_PROXY
+  resp.bluetooth_proxy.feature_flags = bluetooth_proxy::global_bluetooth_proxy->get_feature_flags();
+  char bluetooth_mac[MAC_ADDRESS_PRETTY_BUFFER_SIZE];
+  bluetooth_proxy::global_bluetooth_proxy->get_bluetooth_mac_address_pretty(bluetooth_mac);
+  resp.bluetooth_proxy.mac_address = StringRef(bluetooth_mac);
+#endif
+#ifdef USE_VOICE_ASSISTANT
+  resp.voice_assistant.feature_flags = voice_assistant::global_voice_assistant->get_feature_flags();
+#endif
+#ifdef USE_ZWAVE_PROXY
+  resp.zwave_proxy.feature_flags = zwave_proxy::global_zwave_proxy->get_feature_flags();
+  resp.zwave_proxy.home_id = zwave_proxy::global_zwave_proxy->get_home_id();
+#endif
+#ifdef USE_SERIAL_PROXY
+  size_t serial_proxy_index = 0;
+  for (auto const &proxy : App.get_serial_proxies()) {
+    if (serial_proxy_index >= SERIAL_PROXY_COUNT)
+      break;
+    auto &info = resp.serial_proxies[serial_proxy_index++];
+    info.name = StringRef(proxy->get_name());
+    info.port_type = proxy->get_port_type();
+  }
+#endif
+  return this->send_message(resp);
+}
 void APIConnection::on_hello_request(const HelloRequest &msg) {
  if (!this->send_hello_response_(msg)) {
    this->on_fatal_error();
@@ -1925,6 +1979,11 @@ void APIConnection::on_device_info_request() {
    this->on_fatal_error();
  }
 }
+void APIConnection::on_device_capabilities_request() {
+  if (!this->send_device_capabilities_response_()) {
+    this->on_fatal_error();
+  }
+}

 #ifdef USE_API_HOMEASSISTANT_STATES
 void APIConnection::on_home_assistant_state_response(const HomeAssistantStateResponse &msg) {
@@ -2003,7 +2062,9 @@ void APIConnection::send_execute_service_response(uint32_t call_id, bool success
  resp.call_id = call_id;
  resp.success = success;
  resp.error_message = error_message;
-  this->send_message(resp);
+  if (!this->send_message(resp)) {
+    API_LOG_MSG_DROPPED(TAG, "Action response");
+  }
 }
 #ifdef USE_API_USER_DEFINED_ACTION_RESPONSES_JSON
 void APIConnection::send_execute_service_response(uint32_t call_id, bool success, StringRef error_message,
@@ -2014,12 +2075,34 @@ void APIConnection::send_execute_service_response(uint32_t call_id, bool success
  resp.error_message = error_message;
  resp.response_data = response_data;
  resp.response_data_len = response_data_len;
-  this->send_message(resp);
+  if (!this->send_message(resp)) {
+    API_LOG_MSG_DROPPED(TAG, "Action response");
+  }
 }
 #endif  // USE_API_USER_DEFINED_ACTION_RESPONSES_JSON
 #endif  // USE_API_USER_DEFINED_ACTION_RESPONSES
 #endif

+#ifdef USE_API_HOMEASSISTANT_SERVICES
+bool APIConnection::send_homeassistant_action(const HomeassistantActionRequest &call) {
+  if (!this->flags_.service_call_subscription)
+    return false;
+  if (!this->send_message(call)) {
+    API_LOG_MSG_DROPPED(TAG, "Action request");
+  }
+  return true;
+}
+#endif  // USE_API_HOMEASSISTANT_SERVICES
+
+#ifdef USE_HOMEASSISTANT_TIME
+void APIConnection::send_time_request() {
+  GetTimeRequest req;
+  if (!this->send_message(req)) {
+    API_LOG_MSG_DROPPED(TAG, "Time request");
+  }
+}
+#endif  // USE_HOMEASSISTANT_TIME
+
 #ifdef USE_API_HOMEASSISTANT_ACTION_RESPONSES
 void APIConnection::on_homeassistant_action_response(const HomeassistantActionResponse &msg) {
 #ifdef USE_API_HOMEASSISTANT_ACTION_RESPONSES_JSON
@@ -2094,7 +2177,10 @@ bool APIConnection::try_to_clear_buffer_slow_(bool log_out_of_space) {
  if (this->helper_->can_write_without_blocking())
    return true;
  if (log_out_of_space) {
-    ESP_LOGV(TAG, "Cannot send message because of TCP buffer space");
+    // VV: refusals are either reported by the sending call site (naming what
+    // was lost) or retried without loss (the deferred batch), so this generic
+    // line only duplicates them.
+    ESP_LOGVV(TAG, "Cannot send message because of TCP buffer space");
  }
  return false;
 }
@@ -25,6 +25,7 @@
 #include "esphome/components/esp8266/crash_handler.h"
 #endif
 #include "esphome/core/entity_base.h"
+#include "esphome/core/log.h"
 #include "esphome/core/string_ref.h"

 #include <functional>
@@ -40,6 +41,16 @@ namespace esphome::api {
 // Forward-declared to break the api_server.h cycle; full-type inlines are in api_connection_buffer.h.
 class APIServer;

+// One shared flash string for every refused-frame warning: send_message()
+// fails as soon as the TCP buffer is full, and each caller only pays for its
+// short name. The guard drops the helper and its arguments below WARN.
+#if ESPHOME_LOG_LEVEL >= ESPHOME_LOG_LEVEL_WARN
+void log_dropped_message(const char *tag, int line, const LogString *what);
+#define API_LOG_MSG_DROPPED(tag, what) esphome::api::log_dropped_message(tag, __LINE__, LOG_STR(what))
+#else
+#define API_LOG_MSG_DROPPED(tag, what)
+#endif
+
 // Keepalive timeout in milliseconds
 static constexpr uint32_t KEEPALIVE_TIMEOUT_MS = 60000;
 // Maximum number of entities to process in a single batch during initial state/info sending
@@ -169,12 +180,7 @@ class APIConnection final : public APIServerConnectionBase {
  // Returns whether this client has subscribed to Home Assistant actions; the message
  // is only handed to the send path when subscribed. A true return does not guarantee
  // delivery - it lets the caller warn when no connected client has the subscription.
-  bool send_homeassistant_action(const HomeassistantActionRequest &call) {
-    if (!this->flags_.service_call_subscription)
-      return false;
-    this->send_message(call);
-    return true;
-  }
+  bool send_homeassistant_action(const HomeassistantActionRequest &call);
 #ifdef USE_API_HOMEASSISTANT_ACTION_RESPONSES
  void on_homeassistant_action_response(const HomeassistantActionResponse &msg);
 #endif  // USE_API_HOMEASSISTANT_ACTION_RESPONSES
@@ -183,6 +189,7 @@ class APIConnection final : public APIServerConnectionBase {
  void on_subscribe_bluetooth_le_advertisements_request(const SubscribeBluetoothLEAdvertisementsRequest &msg);
  void on_unsubscribe_bluetooth_le_advertisements_request();

+#ifdef USE_BLUETOOTH_PROXY_CONNECTIONS
  void on_bluetooth_device_request(const BluetoothDeviceRequest &msg);
  void on_bluetooth_gatt_read_request(const BluetoothGATTReadRequest &msg);
  void on_bluetooth_gatt_write_request(const BluetoothGATTWriteRequest &msg);
@@ -191,15 +198,13 @@ class APIConnection final : public APIServerConnectionBase {
  void on_bluetooth_gatt_get_services_request(const BluetoothGATTGetServicesRequest &msg);
  void on_bluetooth_gatt_notify_request(const BluetoothGATTNotifyRequest &msg);
  void on_subscribe_bluetooth_connections_free_request();
-  void on_bluetooth_scanner_set_mode_request(const BluetoothScannerSetModeRequest &msg);
  void on_bluetooth_set_connection_params_request(const BluetoothSetConnectionParamsRequest &msg);
+#endif
+  void on_bluetooth_scanner_set_mode_request(const BluetoothScannerSetModeRequest &msg);

 #endif
 #ifdef USE_HOMEASSISTANT_TIME
-  void send_time_request() {
-    GetTimeRequest req;
-    this->send_message(req);
-  }
+  void send_time_request();
 #endif

 #ifdef USE_VOICE_ASSISTANT
@@ -266,6 +271,7 @@ class APIConnection final : public APIServerConnectionBase {
  void on_disconnect_request(const DisconnectRequest &msg);
  void on_ping_request();
  void on_device_info_request();
+  void on_device_capabilities_request();
  void on_list_entities_request() { this->begin_iterator_(ActiveIterator::LIST_ENTITIES); }
  void on_subscribe_states_request() {
    this->flags_.state_subscription = true;
@@ -334,7 +340,9 @@ class APIConnection final : public APIServerConnectionBase {
  // Function pointer type for type-erased size calculation
  using CalculateSizeFn = uint32_t (*)(const void *);

-  template<typename T> bool send_message(const T &msg) {
+  /// Returns false as soon as the TCP buffer is full. Marked nodiscard so we
+  /// have no silent failures: every caller must handle (or log) a refusal.
+  template<typename T> [[nodiscard]] bool send_message(const T &msg) {
    if constexpr (T::ESTIMATED_SIZE == 0) {
      return this->send_message_(0, T::MESSAGE_TYPE, &encode_msg_noop, &msg);
    } else {
@@ -385,10 +393,11 @@ class APIConnection final : public APIServerConnectionBase {
  bool send_disconnect_response_();
  bool send_ping_response_();
  bool send_device_info_response_();
+  bool send_device_capabilities_response_();
 #ifdef USE_API_NOISE
  bool send_noise_encryption_set_key_response_(const NoiseEncryptionSetKeyRequest &msg);
 #endif
-#ifdef USE_BLUETOOTH_PROXY
+#ifdef USE_BLUETOOTH_PROXY_CONNECTIONS
  bool send_subscribe_bluetooth_connections_free_response_();
 #endif
 #ifdef USE_VOICE_ASSISTANT
@@ -149,7 +149,7 @@ class APIFrameHelper {
  // holding data too long waiting for Nagle's timer causes buffer exhaustion
  // and dropped messages.
  //
-  // ESP32 (TCP_SND_BUF=4×MSS+) / RP2040 (8×MSS) / LibreTiny (4×MSS): 4 logs per cycle
+  // ESP32 (TCP_SND_BUF=4×MSS+) / RP2040 (4×MSS) / LibreTiny (4×MSS): 4 logs per cycle
  // ESP8266 (2×MSS): 3 logs per cycle (tightest buffers)
  //
  // Flow (ESP32/RP2040/LT): Log 1 (Nagle on) -> Log 2 -> Log 3 -> Log 4 (NODELAY, flush)
@@ -312,7 +312,7 @@ class APIFrameHelper {
  // Values 1..LOG_NAGLE_COUNT count log messages in the current Nagle batch.
  // After LOG_NAGLE_COUNT logs, we flush by re-enabling NODELAY and resetting to 0.
  // ESP8266 has the tightest TCP send buffer (2×MSS) and needs conservative batching.
-  // ESP32 (4×MSS+), RP2040 (8×MSS), and LibreTiny (4×MSS) can coalesce more.
+  // ESP32 (4×MSS+), RP2040 (4×MSS), and LibreTiny (4×MSS) can coalesce more.
 #ifdef USE_ESP8266
  static constexpr uint8_t LOG_NAGLE_COUNT = 2;
 #else
@@ -591,18 +591,21 @@ APIError APINoiseFrameHelper::write_frame_(const uint8_t *data, uint16_t len) {
 */
 APIError APINoiseFrameHelper::init_handshake_() {
  int err;
-  memset(&nid_, 0, sizeof(nid_));
-  // const char *proto = "Noise_NNpsk0_25519_ChaChaPoly_SHA256";
-  // err = noise_protocol_name_to_id(&nid_, proto, strlen(proto));
-  nid_.pattern_id = NOISE_PATTERN_NN;
-  nid_.cipher_id = NOISE_CIPHER_CHACHAPOLY;
-  nid_.dh_id = NOISE_DH_CURVE25519;
-  nid_.prefix_id = NOISE_PREFIX_STANDARD;
-  nid_.hybrid_id = NOISE_DH_NONE;
-  nid_.hash_id = NOISE_HASH_SHA256;
-  nid_.modifier_ids[0] = NOISE_MODIFIER_PSK0;
+  // Noise_NNpsk0_25519_ChaChaPoly_SHA256, built on the stack:
+  // noise_handshakestate_new_by_id copies it, so a member would waste
+  // 104 bytes per connection, and a static const would sit in RAM on
+  // ESP8266 (.rodata is DRAM there).
+  const NoiseProtocolId nid = {
+      .prefix_id = NOISE_PREFIX_STANDARD,
+      .pattern_id = NOISE_PATTERN_NN,
+      .modifier_ids = {NOISE_MODIFIER_PSK0},
+      .dh_id = NOISE_DH_CURVE25519,
+      .cipher_id = NOISE_CIPHER_CHACHAPOLY,
+      .hash_id = NOISE_HASH_SHA256,
+      .hybrid_id = NOISE_DH_NONE,
+  };

-  err = noise_handshakestate_new_by_id(&handshake_, &nid_, NOISE_ROLE_RESPONDER);
+  err = noise_handshakestate_new_by_id(&handshake_, &nid, NOISE_ROLE_RESPONDER);
  APIError aerr =
      handle_noise_error_(err, LOG_STR("noise_handshakestate_new_by_id"), APIError::HANDSHAKESTATE_SETUP_FAILED);
  if (aerr != APIError::OK)
@@ -63,9 +63,6 @@ class APINoiseFrameHelper final : public APIFrameHelper {
  // Buffer for noise handshake prologue (released after handshake)
  APIBuffer prologue_;

-  // NoiseProtocolId (size depends on implementation)
-  NoiseProtocolId nid_;
-
  // Group small types together
  // Fixed-size header buffer for noise protocol:
  // 1 byte for indicator + 2 bytes for message size (16-bit value, not varint)
@@ -241,6 +241,82 @@ uint32_t DeviceInfoResponse::calculate_size() const {
 #endif
  return size;
 }
+#ifdef USE_BLUETOOTH_PROXY
+uint8_t *BluetoothProxyCapabilities::encode(ProtoWriteBuffer &buffer PROTO_ENCODE_DEBUG_PARAM) const {
+  uint8_t *__restrict__ pos = buffer.get_pos();
+  ProtoEncode::encode_uint32(pos PROTO_ENCODE_DEBUG_ARG, 1, this->feature_flags);
+  ProtoEncode::encode_short_string_force(pos PROTO_ENCODE_DEBUG_ARG, 18, this->mac_address);
+  return pos;
+}
+uint32_t BluetoothProxyCapabilities::calculate_size() const {
+  uint32_t size = 0;
+  size += ProtoSize::calc_uint32(1, this->feature_flags);
+  size += 2 + this->mac_address.size();
+  return size;
+}
+#endif
+#ifdef USE_VOICE_ASSISTANT
+uint8_t *VoiceAssistantCapabilities::encode(ProtoWriteBuffer &buffer PROTO_ENCODE_DEBUG_PARAM) const {
+  uint8_t *__restrict__ pos = buffer.get_pos();
+  ProtoEncode::encode_uint32(pos PROTO_ENCODE_DEBUG_ARG, 1, this->feature_flags);
+  return pos;
+}
+uint32_t VoiceAssistantCapabilities::calculate_size() const {
+  uint32_t size = 0;
+  size += ProtoSize::calc_uint32(1, this->feature_flags);
+  return size;
+}
+#endif
+#ifdef USE_ZWAVE_PROXY
+uint8_t *ZWaveProxyCapabilities::encode(ProtoWriteBuffer &buffer PROTO_ENCODE_DEBUG_PARAM) const {
+  uint8_t *__restrict__ pos = buffer.get_pos();
+  ProtoEncode::encode_uint32(pos PROTO_ENCODE_DEBUG_ARG, 1, this->feature_flags);
+  ProtoEncode::encode_uint32(pos PROTO_ENCODE_DEBUG_ARG, 2, this->home_id);
+  return pos;
+}
+uint32_t ZWaveProxyCapabilities::calculate_size() const {
+  uint32_t size = 0;
+  size += ProtoSize::calc_uint32(1, this->feature_flags);
+  size += ProtoSize::calc_uint32(1, this->home_id);
+  return size;
+}
+#endif
+uint8_t *DeviceCapabilitiesResponse::encode(ProtoWriteBuffer &buffer PROTO_ENCODE_DEBUG_PARAM) const {
+  uint8_t *__restrict__ pos = buffer.get_pos();
+#ifdef USE_BLUETOOTH_PROXY
+  ProtoEncode::encode_optional_sub_message(pos PROTO_ENCODE_DEBUG_ARG, buffer, 1, this->bluetooth_proxy);
+#endif
+#ifdef USE_VOICE_ASSISTANT
+  ProtoEncode::encode_optional_sub_message(pos PROTO_ENCODE_DEBUG_ARG, buffer, 2, this->voice_assistant);
+#endif
+#ifdef USE_ZWAVE_PROXY
+  ProtoEncode::encode_optional_sub_message(pos PROTO_ENCODE_DEBUG_ARG, buffer, 3, this->zwave_proxy);
+#endif
+#ifdef USE_SERIAL_PROXY
+  for (const auto &it : this->serial_proxies) {
+    ProtoEncode::encode_sub_message(pos PROTO_ENCODE_DEBUG_ARG, buffer, 4, it);
+  }
+#endif
+  return pos;
+}
+uint32_t DeviceCapabilitiesResponse::calculate_size() const {
+  uint32_t size = 0;
+#ifdef USE_BLUETOOTH_PROXY
+  size += ProtoSize::calc_message(1, this->bluetooth_proxy.calculate_size());
+#endif
+#ifdef USE_VOICE_ASSISTANT
+  size += ProtoSize::calc_message(1, this->voice_assistant.calculate_size());
+#endif
+#ifdef USE_ZWAVE_PROXY
+  size += ProtoSize::calc_message(1, this->zwave_proxy.calculate_size());
+#endif
+#ifdef USE_SERIAL_PROXY
+  for (const auto &it : this->serial_proxies) {
+    size += ProtoSize::calc_message_force(1, it.calculate_size());
+  }
+#endif
+  return size;
+}
 #ifdef USE_BINARY_SENSOR
 uint8_t *ListEntitiesBinarySensorResponse::encode(ProtoWriteBuffer &buffer PROTO_ENCODE_DEBUG_PARAM) const {
  uint8_t *__restrict__ pos = buffer.get_pos();
@@ -2406,6 +2482,8 @@ BluetoothLERawAdvertisementsResponse::calculate_size() const {
  }
  return size;
 }
+#endif
+#ifdef USE_BLUETOOTH_PROXY_CONNECTIONS
 bool BluetoothDeviceRequest::decode_varint(uint32_t field_id, proto_varint_value_t value) {
  switch (field_id) {
    case 1:
@@ -2782,6 +2860,8 @@ uint32_t BluetoothDeviceClearCacheResponse::calculate_size() const {
  size += ProtoSize::calc_int32(1, this->error);
  return size;
 }
+#endif
+#ifdef USE_BLUETOOTH_PROXY
 uint8_t *BluetoothScannerStateResponse::encode(ProtoWriteBuffer &buffer PROTO_ENCODE_DEBUG_PARAM) const {
  uint8_t *__restrict__ pos = buffer.get_pos();
  ProtoEncode::encode_uint32(pos PROTO_ENCODE_DEBUG_ARG, 1, static_cast<uint32_t>(this->state));
@@ -4145,7 +4225,7 @@ uint32_t SerialProxyRequestResponse::calculate_size() const {
  return size;
 }
 #endif
-#ifdef USE_BLUETOOTH_PROXY
+#ifdef USE_BLUETOOTH_PROXY_CONNECTIONS
 bool BluetoothSetConnectionParamsRequest::decode_varint(uint32_t field_id, proto_varint_value_t value) {
  switch (field_id) {
    case 1:
@@ -225,7 +225,7 @@ enum MediaPlayerFormatPurpose : uint32_t {
  MEDIA_PLAYER_FORMAT_PURPOSE_ANNOUNCEMENT = 1,
 };
 #endif
-#ifdef USE_BLUETOOTH_PROXY
+#ifdef USE_BLUETOOTH_PROXY_CONNECTIONS
 enum BluetoothDeviceRequestType : uint32_t {
  BLUETOOTH_DEVICE_REQUEST_TYPE_CONNECT = 0,
  BLUETOOTH_DEVICE_REQUEST_TYPE_DISCONNECT = 1,
@@ -235,6 +235,8 @@ enum BluetoothDeviceRequestType : uint32_t {
  BLUETOOTH_DEVICE_REQUEST_TYPE_CONNECT_V3_WITHOUT_CACHE = 5,
  BLUETOOTH_DEVICE_REQUEST_TYPE_CLEAR_CACHE = 6,
 };
+#endif
+#ifdef USE_BLUETOOTH_PROXY
 enum BluetoothScannerState : uint32_t {
  BLUETOOTH_SCANNER_STATE_IDLE = 0,
  BLUETOOTH_SCANNER_STATE_STARTING = 1,
@@ -600,6 +602,74 @@ class DeviceInfoResponse final : public ProtoMessage {

 protected:
 };
+#ifdef USE_BLUETOOTH_PROXY
+class BluetoothProxyCapabilities final : public ProtoMessage {
+ public:
+  uint32_t feature_flags{0};
+  StringRef mac_address{};
+  uint8_t *encode(ProtoWriteBuffer &buffer PROTO_ENCODE_DEBUG_PARAM) const;
+  uint32_t calculate_size() const;
+#ifdef HAS_PROTO_MESSAGE_DUMP
+  const char *dump_to(DumpBuffer &out) const override;
+#endif
+
+ protected:
+};
+#endif
+#ifdef USE_VOICE_ASSISTANT
+class VoiceAssistantCapabilities final : public ProtoMessage {
+ public:
+  uint32_t feature_flags{0};
+  uint8_t *encode(ProtoWriteBuffer &buffer PROTO_ENCODE_DEBUG_PARAM) const;
+  uint32_t calculate_size() const;
+#ifdef HAS_PROTO_MESSAGE_DUMP
+  const char *dump_to(DumpBuffer &out) const override;
+#endif
+
+ protected:
+};
+#endif
+#ifdef USE_ZWAVE_PROXY
+class ZWaveProxyCapabilities final : public ProtoMessage {
+ public:
+  uint32_t feature_flags{0};
+  uint32_t home_id{0};
+  uint8_t *encode(ProtoWriteBuffer &buffer PROTO_ENCODE_DEBUG_PARAM) const;
+  uint32_t calculate_size() const;
+#ifdef HAS_PROTO_MESSAGE_DUMP
+  const char *dump_to(DumpBuffer &out) const override;
+#endif
+
+ protected:
+};
+#endif
+class DeviceCapabilitiesResponse final : public ProtoMessage {
+ public:
+  static constexpr uint8_t MESSAGE_TYPE = 150;
+  static constexpr uint8_t ESTIMATED_SIZE = 102;
+#ifdef HAS_PROTO_MESSAGE_DUMP
+  const LogString *message_name() const override { return LOG_STR("device_capabilities_response"); }
+#endif
+#ifdef USE_BLUETOOTH_PROXY
+  BluetoothProxyCapabilities bluetooth_proxy{};
+#endif
+#ifdef USE_VOICE_ASSISTANT
+  VoiceAssistantCapabilities voice_assistant{};
+#endif
+#ifdef USE_ZWAVE_PROXY
+  ZWaveProxyCapabilities zwave_proxy{};
+#endif
+#ifdef USE_SERIAL_PROXY
+  std::array<SerialProxyInfo, SERIAL_PROXY_COUNT> serial_proxies{};
+#endif
+  uint8_t *encode(ProtoWriteBuffer &buffer PROTO_ENCODE_DEBUG_PARAM) const;
+  uint32_t calculate_size() const;
+#ifdef HAS_PROTO_MESSAGE_DUMP
+  const char *dump_to(DumpBuffer &out) const override;
+#endif
+
+ protected:
+};
 class ListEntitiesDoneResponse final : public ProtoMessage {
 public:
  static constexpr uint8_t MESSAGE_TYPE = 19;
@@ -1931,6 +2001,8 @@ class BluetoothLERawAdvertisementsResponse final : public ProtoMessage {

 protected:
 };
+#endif
+#ifdef USE_BLUETOOTH_PROXY_CONNECTIONS
 class BluetoothDeviceRequest final : public ProtoDecodableMessage {
 public:
  static constexpr uint8_t MESSAGE_TYPE = 68;
@@ -2316,6 +2388,8 @@ class BluetoothDeviceClearCacheResponse final : public ProtoMessage {

 protected:
 };
+#endif
+#ifdef USE_BLUETOOTH_PROXY
 class BluetoothScannerStateResponse final : public ProtoMessage {
 public:
  static constexpr uint8_t MESSAGE_TYPE = 126;
@@ -3290,7 +3364,7 @@ class SerialProxyRequestResponse final : public ProtoMessage {
 protected:
 };
 #endif
-#ifdef USE_BLUETOOTH_PROXY
+#ifdef USE_BLUETOOTH_PROXY_CONNECTIONS
 class BluetoothSetConnectionParamsRequest final : public ProtoDecodableMessage {
 public:
  static constexpr uint8_t MESSAGE_TYPE = 145;
@@ -3,7 +3,7 @@
 #pragma once

 #include "esphome/core/defines.h"
-#ifdef USE_BLUETOOTH_PROXY
+#if defined(USE_BLUETOOTH_PROXY) || defined(USE_BLUETOOTH_PROXY_CONNECTIONS)
 #ifndef USE_API_VARINT64
 #define USE_API_VARINT64
 #endif
@@ -584,7 +584,7 @@ template<> const char *proto_enum_to_string<enums::MediaPlayerFormatPurpose>(enu
  }
 }
 #endif
-#ifdef USE_BLUETOOTH_PROXY
+#ifdef USE_BLUETOOTH_PROXY_CONNECTIONS
 template<>
 const char *proto_enum_to_string<enums::BluetoothDeviceRequestType>(enums::BluetoothDeviceRequestType value) {
  switch (value) {
@@ -606,6 +606,8 @@ const char *proto_enum_to_string<enums::BluetoothDeviceRequestType>(enums::Bluet
      return ESPHOME_PSTR("UNKNOWN");
  }
 }
+#endif
+#ifdef USE_BLUETOOTH_PROXY
 template<> const char *proto_enum_to_string<enums::BluetoothScannerState>(enums::BluetoothScannerState value) {
  switch (value) {
    case enums::BLUETOOTH_SCANNER_STATE_IDLE:
@@ -988,6 +990,55 @@ const char *DeviceInfoResponse::dump_to(DumpBuffer &out) const {
 #endif
  return out.c_str();
 }
+#ifdef USE_BLUETOOTH_PROXY
+const char *BluetoothProxyCapabilities::dump_to(DumpBuffer &out) const {
+  MessageDumpHelper helper(out, ESPHOME_PSTR("BluetoothProxyCapabilities"));
+  dump_field(out, ESPHOME_PSTR("feature_flags"), this->feature_flags);
+  dump_field(out, ESPHOME_PSTR("mac_address"), this->mac_address);
+  return out.c_str();
+}
+#endif
+#ifdef USE_VOICE_ASSISTANT
+const char *VoiceAssistantCapabilities::dump_to(DumpBuffer &out) const {
+  MessageDumpHelper helper(out, ESPHOME_PSTR("VoiceAssistantCapabilities"));
+  dump_field(out, ESPHOME_PSTR("feature_flags"), this->feature_flags);
+  return out.c_str();
+}
+#endif
+#ifdef USE_ZWAVE_PROXY
+const char *ZWaveProxyCapabilities::dump_to(DumpBuffer &out) const {
+  MessageDumpHelper helper(out, ESPHOME_PSTR("ZWaveProxyCapabilities"));
+  dump_field(out, ESPHOME_PSTR("feature_flags"), this->feature_flags);
+  dump_field(out, ESPHOME_PSTR("home_id"), this->home_id);
+  return out.c_str();
+}
+#endif
+const char *DeviceCapabilitiesResponse::dump_to(DumpBuffer &out) const {
+  MessageDumpHelper helper(out, ESPHOME_PSTR("DeviceCapabilitiesResponse"));
+#ifdef USE_BLUETOOTH_PROXY
+  out.append(2, ' ').append_p(ESPHOME_PSTR("bluetooth_proxy")).append(": ");
+  this->bluetooth_proxy.dump_to(out);
+  out.append("\n");
+#endif
+#ifdef USE_VOICE_ASSISTANT
+  out.append(2, ' ').append_p(ESPHOME_PSTR("voice_assistant")).append(": ");
+  this->voice_assistant.dump_to(out);
+  out.append("\n");
+#endif
+#ifdef USE_ZWAVE_PROXY
+  out.append(2, ' ').append_p(ESPHOME_PSTR("zwave_proxy")).append(": ");
+  this->zwave_proxy.dump_to(out);
+  out.append("\n");
+#endif
+#ifdef USE_SERIAL_PROXY
+  for (const auto &it : this->serial_proxies) {
+    out.append(4, ' ').append_p(ESPHOME_PSTR("serial_proxies")).append(": ");
+    it.dump_to(out);
+    out.append("\n");
+  }
+#endif
+  return out.c_str();
+}
 const char *ListEntitiesDoneResponse::dump_to(DumpBuffer &out) const {
  out.append_p(ESPHOME_PSTR("ListEntitiesDoneResponse {}"));
  return out.c_str();
@@ -1953,6 +2004,8 @@ const char *BluetoothLERawAdvertisementsResponse::dump_to(DumpBuffer &out) const
  }
  return out.c_str();
 }
+#endif
+#ifdef USE_BLUETOOTH_PROXY_CONNECTIONS
 const char *BluetoothDeviceRequest::dump_to(DumpBuffer &out) const {
  MessageDumpHelper helper(out, ESPHOME_PSTR("BluetoothDeviceRequest"));
  dump_field(out, ESPHOME_PSTR("address"), this->address);
@@ -2124,6 +2177,8 @@ const char *BluetoothDeviceClearCacheResponse::dump_to(DumpBuffer &out) const {
  dump_field(out, ESPHOME_PSTR("error"), this->error);
  return out.c_str();
 }
+#endif
+#ifdef USE_BLUETOOTH_PROXY
 const char *BluetoothScannerStateResponse::dump_to(DumpBuffer &out) const {
  MessageDumpHelper helper(out, ESPHOME_PSTR("BluetoothScannerStateResponse"));
  dump_field(out, ESPHOME_PSTR("state"), static_cast<enums::BluetoothScannerState>(this->state));
@@ -2715,7 +2770,7 @@ const char *SerialProxyRequestResponse::dump_to(DumpBuffer &out) const {
  return out.c_str();
 }
 #endif
-#ifdef USE_BLUETOOTH_PROXY
+#ifdef USE_BLUETOOTH_PROXY_CONNECTIONS
 const char *BluetoothSetConnectionParamsRequest::dump_to(DumpBuffer &out) const {
  MessageDumpHelper helper(out, ESPHOME_PSTR("BluetoothSetConnectionParamsRequest"));
  dump_field(out, ESPHOME_PSTR("address"), this->address);
@@ -302,7 +302,7 @@ void APIConnection::read_message_(uint32_t msg_size, uint32_t msg_type, const ui
      break;
    }
 #endif
-#ifdef USE_BLUETOOTH_PROXY
+#ifdef USE_BLUETOOTH_PROXY_CONNECTIONS
    case BluetoothDeviceRequest::MESSAGE_TYPE: {
      BluetoothDeviceRequest msg;
      msg.decode(msg_data, msg_size);
@@ -313,7 +313,7 @@ void APIConnection::read_message_(uint32_t msg_size, uint32_t msg_type, const ui
      break;
    }
 #endif
-#ifdef USE_BLUETOOTH_PROXY
+#ifdef USE_BLUETOOTH_PROXY_CONNECTIONS
    case BluetoothGATTGetServicesRequest::MESSAGE_TYPE: {
      BluetoothGATTGetServicesRequest msg;
      msg.decode(msg_data, msg_size);
@@ -324,7 +324,7 @@ void APIConnection::read_message_(uint32_t msg_size, uint32_t msg_type, const ui
      break;
    }
 #endif
-#ifdef USE_BLUETOOTH_PROXY
+#ifdef USE_BLUETOOTH_PROXY_CONNECTIONS
    case BluetoothGATTReadRequest::MESSAGE_TYPE: {
      BluetoothGATTReadRequest msg;
      msg.decode(msg_data, msg_size);
@@ -335,7 +335,7 @@ void APIConnection::read_message_(uint32_t msg_size, uint32_t msg_type, const ui
      break;
    }
 #endif
-#ifdef USE_BLUETOOTH_PROXY
+#ifdef USE_BLUETOOTH_PROXY_CONNECTIONS
    case BluetoothGATTWriteRequest::MESSAGE_TYPE: {
      BluetoothGATTWriteRequest msg;
      msg.decode(msg_data, msg_size);
@@ -346,7 +346,7 @@ void APIConnection::read_message_(uint32_t msg_size, uint32_t msg_type, const ui
      break;
    }
 #endif
-#ifdef USE_BLUETOOTH_PROXY
+#ifdef USE_BLUETOOTH_PROXY_CONNECTIONS
    case BluetoothGATTReadDescriptorRequest::MESSAGE_TYPE: {
      BluetoothGATTReadDescriptorRequest msg;
      msg.decode(msg_data, msg_size);
@@ -357,7 +357,7 @@ void APIConnection::read_message_(uint32_t msg_size, uint32_t msg_type, const ui
      break;
    }
 #endif
-#ifdef USE_BLUETOOTH_PROXY
+#ifdef USE_BLUETOOTH_PROXY_CONNECTIONS
    case BluetoothGATTWriteDescriptorRequest::MESSAGE_TYPE: {
      BluetoothGATTWriteDescriptorRequest msg;
      msg.decode(msg_data, msg_size);
@@ -368,7 +368,7 @@ void APIConnection::read_message_(uint32_t msg_size, uint32_t msg_type, const ui
      break;
    }
 #endif
-#ifdef USE_BLUETOOTH_PROXY
+#ifdef USE_BLUETOOTH_PROXY_CONNECTIONS
    case BluetoothGATTNotifyRequest::MESSAGE_TYPE: {
      BluetoothGATTNotifyRequest msg;
      msg.decode(msg_data, msg_size);
@@ -379,7 +379,7 @@ void APIConnection::read_message_(uint32_t msg_size, uint32_t msg_type, const ui
      break;
    }
 #endif
-#ifdef USE_BLUETOOTH_PROXY
+#ifdef USE_BLUETOOTH_PROXY_CONNECTIONS
    case 80 /* SubscribeBluetoothConnectionsFreeRequest is empty */: {
 #ifdef HAS_PROTO_MESSAGE_DUMP
      this->log_receive_message_(LOG_STR("on_subscribe_bluetooth_connections_free_request"));
@@ -694,7 +694,7 @@ void APIConnection::read_message_(uint32_t msg_size, uint32_t msg_type, const ui
      break;
    }
 #endif
-#ifdef USE_BLUETOOTH_PROXY
+#ifdef USE_BLUETOOTH_PROXY_CONNECTIONS
    case BluetoothSetConnectionParamsRequest::MESSAGE_TYPE: {
      BluetoothSetConnectionParamsRequest msg;
      msg.decode(msg_data, msg_size);
@@ -705,6 +705,13 @@ void APIConnection::read_message_(uint32_t msg_size, uint32_t msg_type, const ui
      break;
    }
 #endif
+    case 149 /* DeviceCapabilitiesRequest is empty */: {
+#ifdef HAS_PROTO_MESSAGE_DUMP
+      this->log_receive_message_(LOG_STR("on_device_capabilities_request"));
+#endif
+      this->on_device_capabilities_request();
+      break;
+    }
    default:
      break;
  }
@@ -27,6 +27,8 @@ class APIServerConnectionBase {
  void on_ping_response(){};
  void on_device_info_request(){};

+  void on_device_capabilities_request(){};
+
  void on_list_entities_request(){};

  void on_subscribe_states_request(){};
@@ -113,32 +115,32 @@ class APIServerConnectionBase {
  void on_subscribe_bluetooth_le_advertisements_request(const SubscribeBluetoothLEAdvertisementsRequest &value){};
 #endif

-#ifdef USE_BLUETOOTH_PROXY
+#ifdef USE_BLUETOOTH_PROXY_CONNECTIONS
  void on_bluetooth_device_request(const BluetoothDeviceRequest &value){};
 #endif

-#ifdef USE_BLUETOOTH_PROXY
+#ifdef USE_BLUETOOTH_PROXY_CONNECTIONS
  void on_bluetooth_gatt_get_services_request(const BluetoothGATTGetServicesRequest &value){};
 #endif

-#ifdef USE_BLUETOOTH_PROXY
+#ifdef USE_BLUETOOTH_PROXY_CONNECTIONS
  void on_bluetooth_gatt_read_request(const BluetoothGATTReadRequest &value){};
 #endif

-#ifdef USE_BLUETOOTH_PROXY
+#ifdef USE_BLUETOOTH_PROXY_CONNECTIONS
  void on_bluetooth_gatt_write_request(const BluetoothGATTWriteRequest &value){};
 #endif
-#ifdef USE_BLUETOOTH_PROXY
+#ifdef USE_BLUETOOTH_PROXY_CONNECTIONS
  void on_bluetooth_gatt_read_descriptor_request(const BluetoothGATTReadDescriptorRequest &value){};
 #endif
-#ifdef USE_BLUETOOTH_PROXY
+#ifdef USE_BLUETOOTH_PROXY_CONNECTIONS
  void on_bluetooth_gatt_write_descriptor_request(const BluetoothGATTWriteDescriptorRequest &value){};
 #endif
-#ifdef USE_BLUETOOTH_PROXY
+#ifdef USE_BLUETOOTH_PROXY_CONNECTIONS
  void on_bluetooth_gatt_notify_request(const BluetoothGATTNotifyRequest &value){};
 #endif

-#ifdef USE_BLUETOOTH_PROXY
+#ifdef USE_BLUETOOTH_PROXY_CONNECTIONS
  void on_subscribe_bluetooth_connections_free_request(){};
 #endif

@@ -233,7 +235,7 @@ class APIServerConnectionBase {
  void on_serial_proxy_request(const SerialProxyRequest &value){};
 #endif

-#ifdef USE_BLUETOOTH_PROXY
+#ifdef USE_BLUETOOTH_PROXY_CONNECTIONS
  void on_bluetooth_set_connection_params_request(const BluetoothSetConnectionParamsRequest &value){};
 #endif
 };
@@ -123,7 +123,9 @@ void APIServer::setup() {
        // Best-effort: if the send buffer is full the reason is dropped, but the
        // client still learns the window is closed when it reconnects (rejected at
        // hello) or via the socket close.
-        c->send_message(req);
+        if (!c->send_message(req)) {
+          API_LOG_MSG_DROPPED(TAG, "Disconnect request");
+        }
      }
    });
  }
@@ -394,8 +396,11 @@ void APIServer::on_update(update::UpdateEntity *obj) {
 void APIServer::on_zwave_proxy_request(const ZWaveProxyRequest &msg) {
  // We could add code to manage a second subscription type, but, since this message type is
  //  very infrequent and small, we simply send it to all clients
-  for (auto &c : this->active_clients())
-    c->send_message(msg);
+  for (auto &c : this->active_clients()) {
+    if (!c->send_message(msg)) {
+      API_LOG_MSG_DROPPED(TAG, "Home ID notification");
+    }
+  }
 }
 #endif

@@ -418,12 +423,6 @@ void APIServer::send_infrared_rf_receive_event([[maybe_unused]] uint32_t device_
 API_DISPATCH_UPDATE(alarm_control_panel::AlarmControlPanel, alarm_control_panel)
 #endif

-float APIServer::get_setup_priority() const { return setup_priority::AFTER_WIFI; }
-
-void APIServer::set_port(uint16_t port) { this->port_ = port; }
-
-void APIServer::set_batch_delay(uint16_t batch_delay) { this->batch_delay_ = batch_delay; }
-
 #ifdef USE_API_HOMEASSISTANT_SERVICES
 void APIServer::send_homeassistant_action(const HomeassistantActionRequest &call) {
  bool has_subscriber = false;
@@ -548,10 +547,6 @@ const std::vector<APIServer::HomeAssistantStateSubscription> &APIServer::get_sta
 }
 #endif

-uint16_t APIServer::get_port() const { return this->port_; }
-
-void APIServer::set_reboot_timeout(uint32_t reboot_timeout) { this->reboot_timeout_ = reboot_timeout; }
-
 #ifdef USE_API_NOISE
 bool APIServer::update_noise_psk_(const SavedNoisePsk &new_psk, const LogString *save_log_msg,
                                  const LogString *fail_log_msg, bool make_active) {
@@ -576,7 +571,9 @@ bool APIServer::update_noise_psk_(const SavedNoisePsk &new_psk, const LogString
      ESP_LOGW(TAG, "Disconnecting all clients to reset PSK");
      for (auto &c : this->active_clients()) {
        DisconnectRequest req;
-        c->send_message(req);
+        if (!c->send_message(req)) {
+          API_LOG_MSG_DROPPED(TAG, "Disconnect request");
+        }
      }
    });
  }
@@ -51,8 +51,8 @@ class APIServer final : public Component,
 public:
  APIServer();
  void setup() override;
-  uint16_t get_port() const;
-  float get_setup_priority() const override;
+  uint16_t get_port() const { return this->port_; }
+  float get_setup_priority() const override { return setup_priority::AFTER_WIFI; }
  void loop() override;
  void dump_config() override;
  void on_shutdown() override;
@@ -63,9 +63,9 @@ class APIServer final : public Component,
 #ifdef USE_CAMERA
  void on_camera_image(const std::shared_ptr<camera::CameraImage> &image) override;
 #endif
-  void set_port(uint16_t port);
-  void set_reboot_timeout(uint32_t reboot_timeout);
-  void set_batch_delay(uint16_t batch_delay);
+  void set_port(uint16_t port) { this->port_ = port; }
+  void set_reboot_timeout(uint32_t reboot_timeout) { this->reboot_timeout_ = reboot_timeout; }
+  void set_batch_delay(uint16_t batch_delay) { this->batch_delay_ = batch_delay; }
  uint16_t get_batch_delay() const { return batch_delay_; }
  void set_listen_backlog(uint8_t listen_backlog) { this->listen_backlog_ = listen_backlog; }

@@ -1,177 +1,10 @@
-from __future__ import annotations
+"""Backward-compatibility shim; the log client lives in esphome.api_client.

-import asyncio
-from datetime import datetime
-import importlib
-import logging
-from typing import TYPE_CHECKING, Any
-import warnings
+Importing this module executes the whole api component package, which pulls
+in the validation stack. CLI code paths should import esphome.api_client
+directly so the logs fast path stays light.
+"""

-# Suppress protobuf version warnings
-with warnings.catch_warnings():
-    warnings.filterwarnings(
-        "ignore", category=UserWarning, message=".*Protobuf gencode version.*"
-    )
-    from aioesphomeapi import APIClient, parse_log_message
-    from aioesphomeapi.log_runner import async_run
+from esphome.api_client import async_run_logs, run_logs

-import contextlib
-
-from esphome.const import CONF_KEY, CONF_PORT, __version__
-from esphome.core import CORE
-from esphome.util import safe_print
-
-from . import CONF_ENCRYPTION
-
-if TYPE_CHECKING:
-    from aioesphomeapi.api_pb2 import (
-        SubscribeLogsResponse,  # pylint: disable=no-name-in-module
-    )
-
-
-_LOGGER = logging.getLogger(__name__)
-
-
-class _LogLineProcessor:
-    """Feeds incoming log lines to the stack-trace decoder.
-
-    Two responsibilities beyond just calling the decoder:
-    1. Catch everything the decoder can raise. aioesphomeapi isolates
-       exceptions raised by log handlers, so an escaping one no longer
-       kills the session, but it does log a full traceback per line. A
-       crash dump carries a PC line plus one per backtrace frame, so the
-       tracebacks bury the dump the user is trying to read. Decoding is a
-       diagnostic nicety; nothing it raises is worth that noise.
-    2. Disable decoding after the first failure. _decode_pc shells out to
-       the toolchain to resolve addr2line, which is expensive; a single
-       crash dump can contain many PC/BT lines and we don't want to retry
-       the failing subprocess for each one. This only works if every
-       failure is caught, which is why 1 is not narrowed to EsphomeError.
-    """
-
-    def __init__(self, config: dict[str, Any], platform_handler: Any | None) -> None:
-        self._config = config
-        self._platform_handler = platform_handler
-        self._decode_enabled = True
-        self.backtrace_state = False
-
-    def process_line(self, raw_line: str) -> None:
-        if not self._decode_enabled:
-            return
-        try:
-            if self._platform_handler is not None:
-                self.backtrace_state = self._platform_handler(
-                    self._config, raw_line, self.backtrace_state
-                )
-        except Exception as exc:  # noqa: BLE001  # pylint: disable=broad-except
-            self._decode_enabled = False
-            self.backtrace_state = False
-            # _run_idedata raises EsphomeError with no message; fall back
-            # to a generic explanation when str(exc) is empty.
-            detail = str(exc) or "build artifacts not found locally"
-            _LOGGER.debug("Stack-trace decoding failed", exc_info=True)
-            _LOGGER.warning(
-                "Crash trace decoding unavailable: %s. "
-                "Run 'esphome compile' for this device to enable PC decoding.",
-                detail,
-            )
-
-
-async def async_run_logs(
-    config: dict[str, Any],
-    addresses: list[str],
-    subscribe_states: bool = True,
-) -> None:
-    """Run the logs command in the event loop."""
-    conf = config["api"]
-    name = config["esphome"]["name"]
-    port: int = int(conf[CONF_PORT])
-    noise_psk: str | None = None
-    if (encryption := conf.get(CONF_ENCRYPTION)) and (key := encryption.get(CONF_KEY)):
-        noise_psk = key
-
-    if len(addresses) == 1:
-        _LOGGER.info("Starting log output from %s using esphome API", addresses[0])
-    else:
-        _LOGGER.info(
-            "Starting log output from %s using esphome API", " or ".join(addresses)
-        )
-
-    cli = APIClient(
-        addresses[0],  # Primary address for compatibility
-        port,
-        "",  # Password auth removed in 2026.1.0
-        client_info=f"ESPHome Logs {__version__}",
-        noise_psk=noise_psk,
-        addresses=addresses,  # Pass all addresses for automatic retry
-        provide_time=False,
-    )
-
-    # Try platform-specific stacktrace handler first, fall back to generic
-    platform_process_stacktrace = None
-    try:
-        module = importlib.import_module("esphome.components." + CORE.target_platform)
-        platform_process_stacktrace = module.process_stacktrace
-    except (AttributeError, ImportError):
-        _LOGGER.info(
-            'Stacktrace analysis is unavailable: no compatible analyzer found for target platform "%s".',
-            CORE.target_platform,
-        )
-
-    processor = _LogLineProcessor(config, platform_process_stacktrace)
-
-    def on_log(msg: SubscribeLogsResponse) -> None:
-        """Handle a new log message."""
-        time_ = datetime.now().astimezone()
-        message: bytes = msg.message
-        text = message.decode("utf8", "backslashreplace")
-        nanoseconds = time_.microsecond // 1000
-        timestamp = (
-            f"[{time_.hour:02}:{time_.minute:02}:{time_.second:02}.{nanoseconds:03}]"
-        )
-        for parsed_msg in parse_log_message(text, timestamp):
-            # safe_print handles the dashboard \033 escaping and falls back
-            # to backslashreplace encoding on stdouts that can't represent
-            # the wifi signal-bar block characters (Windows redirected
-            # cp1252 pipe).
-            safe_print(parsed_msg)
-        for raw_line in text.splitlines():
-            processor.process_line(raw_line)
-
-    # Safe to fall back to plaintext here only for this diagnostics use
-    # case: the stream is one-way from device to client, and this code
-    # never accepts commands or acts on any message the device sends.
-    # An on-path attacker could still both inject fabricated log lines
-    # and passively read the device's log output (and any state data
-    # delivered when subscribe_states is enabled), so this does lose
-    # confidentiality as well as authentication/integrity. That tradeoff
-    # is acceptable for operator-visible logs, which aioesphomeapi also
-    # warns may come from an unverified device. Never mirror this opt-in
-    # for any connection that sends data to the device or uses Home
-    # Assistant actions.
-    stop = await async_run(
-        cli,
-        on_log,
-        name=name,
-        subscribe_states=subscribe_states,
-        allow_plaintext_fallback=True,
-        # A top-level ``deep_sleep:`` block means the device is only awake
-        # briefly; cap the reconnect backoff so a wake window is not missed.
-        deep_sleep="deep_sleep" in config,
-    )
-    try:
-        await asyncio.Event().wait()
-    finally:
-        await stop()
-
-
-def run_logs(
-    config: dict[str, Any],
-    addresses: list[str],
-    subscribe_states: bool = True,
-) -> None:
-    """Run the logs command."""
-    with contextlib.suppress(KeyboardInterrupt):
-        asyncio.run(
-            async_run_logs(config, addresses, subscribe_states=subscribe_states)
-        )
+__all__ = ["async_run_logs", "run_logs"]
@@ -14,6 +14,8 @@ from esphome.const import (
    CONF_TUNE_ANTENNA,
    CONF_WATCHDOG_THRESHOLD,
 )
+from esphome.cpp_generator import MockObj
+from esphome.types import ConfigType

 MULTI_CONF = True

@@ -42,7 +44,7 @@ AS3935_SCHEMA = cv.Schema(
 )


-async def setup_as3935(var, config):
+async def setup_as3935(var: MockObj, config: ConfigType) -> None:
    await cg.register_component(var, config)

    irq_pin = await cg.gpio_pin_expression(config[CONF_IRQ_PIN])
@@ -1,6 +1,7 @@
 import esphome.codegen as cg
 from esphome.components import binary_sensor
 import esphome.config_validation as cv
+from esphome.types import ConfigType

 from . import AS3935, CONF_AS3935_ID

@@ -13,7 +14,7 @@ CONFIG_SCHEMA = binary_sensor.binary_sensor_schema().extend(
 )


-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
    hub = await cg.get_variable(config[CONF_AS3935_ID])
    var = await binary_sensor.new_binary_sensor(config)
    cg.add(hub.set_thunder_alert_binary_sensor(var))
@@ -9,6 +9,7 @@ from esphome.const import (
    STATE_CLASS_MEASUREMENT,
    UNIT_KILOMETER,
 )
+from esphome.types import ConfigType

 from . import AS3935, CONF_AS3935_ID

@@ -31,7 +32,7 @@ CONFIG_SCHEMA = cv.Schema(
 )


-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
    hub = await cg.get_variable(config[CONF_AS3935_ID])

    if distance_config := config.get(CONF_DISTANCE):
@@ -1,3 +1,6 @@
+from collections.abc import Callable
+from typing import Any
+
 from esphome import pins
 import esphome.codegen as cg
 from esphome.components import i2c
@@ -11,6 +14,7 @@ from esphome.const import (
    CONF_RANGE,
    CONF_WATCHDOG,
 )
+from esphome.types import ConfigType

 CODEOWNERS = ["@ammmze"]
 DEPENDENCIES = ["i2c"]
@@ -72,13 +76,13 @@ POSITION_TO_ANGLE = 360 / RESOLUTION
 MIN_RANGE = round(18 * ANGLE_TO_POSITION)


-def angle(min=-360, max=360):
+def angle(min: float = -360, max: float = 360) -> Callable[[Any], Any]:
    return cv.All(
        cv.float_with_unit("angle", "(°|deg)"), cv.float_range(min=min, max=max)
    )


-def angle_to_position(value, min=-360, max=360):
+def angle_to_position(value: Any, min: float = -360, max: float = 360) -> int:
    try:
        value = angle(min=min, max=max)(value)
        return (RESOLUTION + round(value * ANGLE_TO_POSITION)) % RESOLUTION
@@ -86,17 +90,17 @@ def angle_to_position(value, min=-360, max=360):
        raise cv.Invalid(f"When using angle, {e.error_message}") from e


-def percent_to_position(value):
+def percent_to_position(value: Any) -> int:
    value = cv.possibly_negative_percentage(value)
    return (RESOLUTION + round(value * RESOLUTION)) % RESOLUTION


-def position(min=-MAX_POSITION, max=MAX_POSITION):
+def position(min: int = -MAX_POSITION, max: int = MAX_POSITION) -> Callable[[Any], Any]:
    """Validate that the config option is a position.
    Accepts integers, degrees, or percentage (of 360 degrees).
    """

-    def validator(value):
+    def validator(value: Any) -> int:
        if isinstance(value, str) and value.endswith("%"):
            value = percent_to_position(value)

@@ -112,7 +116,7 @@ def position(min=-MAX_POSITION, max=MAX_POSITION):
    return validator


-def position_range():
+def position_range() -> Callable[[Any], Any]:
    """Validate that value given is a valid range for the device.
    A valid range is one of the following:
    - a value of 0 (meaning full range)
@@ -129,7 +133,7 @@ def position_range():
        zero_validator,
    )

-    def validator(value):
+    def validator(value: Any) -> Any:
        is_negative_str = isinstance(value, str) and value.startswith("-")
        is_negative_num = isinstance(value, (float, int)) and value < 0
        if is_negative_str or is_negative_num:
@@ -139,13 +143,13 @@ def position_range():
    return validator


-def has_valid_range_config():
+def has_valid_range_config() -> Callable[[ConfigType], ConfigType]:
    """Validate that that the config start + end position results in a valid
    positional range, which must be >= 18degrees
    """
    range_validator = position_range()

-    def validator(config):
+    def validator(config: ConfigType) -> ConfigType:
        # if we don't have an end position, then there is nothing to do
        if CONF_END_POSITION not in config:
            return config
@@ -203,7 +207,7 @@ CONFIG_SCHEMA = cv.All(
 )


-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
    var = cg.new_Pvariable(config[CONF_ID])
    await cg.register_component(var, config)
    await i2c.register_i2c_device(var, config)
@@ -11,6 +11,7 @@ from esphome.const import (
    ICON_ROTATE_RIGHT,
    STATE_CLASS_MEASUREMENT,
 )
+from esphome.types import ConfigType

 from .. import AS5600Component, as5600_ns

@@ -77,7 +78,7 @@ CONFIG_SCHEMA = (
 )


-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
    var = cg.new_Pvariable(config[CONF_ID])
    await cg.register_parented(var, config[CONF_AS5600_ID])
    await cg.register_component(var, config)
@@ -4,6 +4,9 @@ import esphome.codegen as cg
 from esphome.components import i2c
 import esphome.config_validation as cv
 from esphome.const import CONF_FREQUENCY, CONF_ID
+from esphome.core import ID
+from esphome.cpp_generator import MockObj, TemplateArgsType
+from esphome.types import ConfigType

 CODEOWNERS = ["@X-Ryl669"]
 DEPENDENCIES = ["i2c"]
@@ -70,7 +73,7 @@ CONFIG_SCHEMA = cv.All(
 )


-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
    var = cg.new_Pvariable(config[CONF_ID])
    await cg.register_component(var, config)
    await i2c.register_i2c_device(var, config)
@@ -91,7 +94,12 @@ AT581XSettingsAction = at581x_ns.class_("AT581XSettingsAction", automation.Actio
    ),
    synchronous=True,
 )
-async def at581x_reset_to_code(config, action_id, template_arg, args):
+async def at581x_reset_to_code(
+    config: ConfigType,
+    action_id: ID,
+    template_arg: cg.TemplateArguments,
+    args: TemplateArgsType,
+) -> MockObj:
    var = cg.new_Pvariable(action_id, template_arg)
    await cg.register_parented(var, config[CONF_ID])

@@ -163,7 +171,12 @@ RADAR_SETTINGS_SCHEMA = cv.Schema(
    RADAR_SETTINGS_SCHEMA,
    synchronous=True,
 )
-async def at581x_settings_to_code(config, action_id, template_arg, args):
+async def at581x_settings_to_code(
+    config: ConfigType,
+    action_id: ID,
+    template_arg: cg.TemplateArguments,
+    args: TemplateArgsType,
+) -> MockObj:
    var = cg.new_Pvariable(action_id, template_arg)
    await cg.register_parented(var, config[CONF_ID])

@@ -2,6 +2,7 @@ import esphome.codegen as cg
 from esphome.components import switch
 import esphome.config_validation as cv
 from esphome.const import DEVICE_CLASS_SWITCH, ICON_WIFI
+from esphome.types import ConfigType

 from .. import CONF_AT581X_ID, AT581XComponent, at581x_ns

@@ -22,7 +23,7 @@ CONFIG_SCHEMA = switch.switch_schema(
 )


-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
    at581x_component = await cg.get_variable(config[CONF_AT581X_ID])
    s = await switch.new_switch(config)
    await cg.register_parented(s, config[CONF_AT581X_ID])
@@ -1,8 +1,6 @@
 #include "atc_mithermometer.h"
 #include "esphome/core/log.h"

-#ifdef USE_ESP32
-
 namespace esphome::atc_mithermometer {

 static const char *const TAG = "atc_mithermometer";
@@ -15,7 +13,7 @@ void ATCMiThermometer::dump_config() {
  LOG_SENSOR("  ", "Battery Voltage", this->battery_voltage_);
 }

-bool ATCMiThermometer::parse_device(const esp32_ble_tracker::ESPBTDevice &device) {
+bool ATCMiThermometer::parse_device(const ble_device_base::ESPBTDevice &device) {
  if (device.address_uint64() != this->address_) {
    ESP_LOGVV(TAG, "parse_device(): unknown MAC address.");
    return false;
@@ -52,7 +50,7 @@ bool ATCMiThermometer::parse_device(const esp32_ble_tracker::ESPBTDevice &device
  return success;
 }

-optional<ParseResult> ATCMiThermometer::parse_header_(const esp32_ble_tracker::ServiceData &service_data) {
+optional<ParseResult> ATCMiThermometer::parse_header_(const ble_device_base::ServiceData &service_data) {
  ParseResult result;
  if (!service_data.uuid.contains(0x1A, 0x18)) {
    ESP_LOGVV(TAG, "parse_header(): no service data UUID magic bytes.");
@@ -132,5 +130,3 @@ bool ATCMiThermometer::report_results_(const optional<ParseResult> &result, cons
 }

 }  // namespace esphome::atc_mithermometer
-
-#endif
@@ -2,12 +2,10 @@

 #include "esphome/core/component.h"
 #include "esphome/components/sensor/sensor.h"
-#include "esphome/components/esp32_ble_tracker/esp32_ble_tracker.h"
+#include "esphome/components/ble_device_base/ble_device.h"

 #include <vector>

-#ifdef USE_ESP32
-
 namespace esphome::atc_mithermometer {

 struct ParseResult {
@@ -18,11 +16,11 @@ struct ParseResult {
  int raw_offset;
 };

-class ATCMiThermometer final : public Component, public esp32_ble_tracker::ESPBTDeviceListener {
+class ATCMiThermometer final : public Component, public ble_device_base::ESPBTDeviceListener {
 public:
  void set_address(uint64_t address) { address_ = address; };

-  bool parse_device(const esp32_ble_tracker::ESPBTDevice &device) override;
+  bool parse_device(const ble_device_base::ESPBTDevice &device) override;
  void dump_config() override;
  void set_temperature(sensor::Sensor *temperature) { temperature_ = temperature; }
  void set_humidity(sensor::Sensor *humidity) { humidity_ = humidity; }
@@ -40,11 +38,9 @@ class ATCMiThermometer final : public Component, public esp32_ble_tracker::ESPBT

  uint8_t last_frame_count_{0};

-  optional<ParseResult> parse_header_(const esp32_ble_tracker::ServiceData &service_data);
+  optional<ParseResult> parse_header_(const ble_device_base::ServiceData &service_data);
  bool parse_message_(const std::vector<uint8_t> &message, ParseResult &result);
  bool report_results_(const optional<ParseResult> &result, const char *address);
 };

 }  // namespace esphome::atc_mithermometer
-
-#endif
@@ -1,5 +1,5 @@
 import esphome.codegen as cg
-from esphome.components import esp32_ble_tracker, sensor
+from esphome.components import ble_device_base, sensor
 import esphome.config_validation as cv
 from esphome.const import (
    CONF_BATTERY_LEVEL,
@@ -24,14 +24,15 @@ from esphome.const import (

 CODEOWNERS = ["@ahpohl"]

-DEPENDENCIES = ["esp32_ble_tracker"]
+AUTO_LOAD = ["ble_device_base"]

 atc_mithermometer_ns = cg.esphome_ns.namespace("atc_mithermometer")
 ATCMiThermometer = atc_mithermometer_ns.class_(
-    "ATCMiThermometer", esp32_ble_tracker.ESPBTDeviceListener, cg.Component
+    "ATCMiThermometer", ble_device_base.ESPBTDeviceListener, cg.Component
 )

-CONFIG_SCHEMA = (
+CONFIG_SCHEMA = cv.All(
+    ble_device_base.rename_legacy_hub_id("atc_mithermometer"),
    cv.Schema(
        {
            cv.GenerateID(): cv.declare_id(ATCMiThermometer),
@@ -71,15 +72,15 @@ CONFIG_SCHEMA = (
            ),
        }
    )
-    .extend(esp32_ble_tracker.ESP_BLE_DEVICE_SCHEMA)
    .extend(cv.COMPONENT_SCHEMA)
+    .extend(ble_device_base.BLE_DEVICE_SCHEMA),
 )


 async def to_code(config):
    var = cg.new_Pvariable(config[CONF_ID])
    await cg.register_component(var, config)
-    await esp32_ble_tracker.register_ble_device(var, config)
+    await ble_device_base.register_ble_device(var, config)

    cg.add(var.set_address(config[CONF_MAC_ADDRESS].as_hex))

@@ -2,6 +2,7 @@ import esphome.codegen as cg
 from esphome.components import button
 import esphome.config_validation as cv
 from esphome.const import CONF_ID, ENTITY_CATEGORY_CONFIG, ICON_SCALE
+from esphome.types import ConfigType

 from .. import atm90e32_ns
 from ..sensor import ATM90E32Component
@@ -67,7 +68,7 @@ CONFIG_SCHEMA = {
 }


-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
    parent = await cg.get_variable(config[CONF_ID])

    if run_gain := config.get(CONF_RUN_GAIN_CALIBRATION):
@@ -15,6 +15,7 @@ from esphome.const import (
    UNIT_AMPERE,
    UNIT_VOLT,
 )
+from esphome.types import ConfigType

 from .. import atm90e32_ns
 from ..sensor import ATM90E32Component
@@ -90,7 +91,7 @@ CONFIG_SCHEMA = cv.Schema(
 )


-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
    parent = await cg.get_variable(config[CONF_ID])

    if voltage_cfg := config.get(CONF_REFERENCE_VOLTAGE):
@@ -41,6 +41,7 @@ from esphome.const import (
    UNIT_WATT,
    UNIT_WATT_HOURS,
 )
+from esphome.types import ConfigType

 from . import atm90e32_ns

@@ -191,7 +192,7 @@ CONFIG_SCHEMA = (
 )


-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
    var = cg.new_Pvariable(config[CONF_ID])
    cg.add(var.set_instance_id(str(config[CONF_ID])))
    await cg.register_component(var, config)
@@ -2,6 +2,7 @@ import esphome.codegen as cg
 from esphome.components import text_sensor
 import esphome.config_validation as cv
 from esphome.const import CONF_ID, CONF_PHASE_A, CONF_PHASE_B, CONF_PHASE_C
+from esphome.types import ConfigType

 from ..sensor import ATM90E32Component

@@ -34,7 +35,7 @@ CONFIG_SCHEMA = cv.Schema(
 )


-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
    parent = await cg.get_variable(config[CONF_ID])

    if phase_cfg := config.get(CONF_PHASE_STATUS):
@@ -1,4 +1,6 @@
+from collections.abc import Callable
 from dataclasses import dataclass, field
+from typing import Any

 import esphome.codegen as cg
 from esphome.components.esp32 import (
@@ -15,6 +17,7 @@ from esphome.const import (
 )
 from esphome.core import CORE
 import esphome.final_validate as fv
+from esphome.types import ConfigType

 AUTO_LOAD = ["ring_buffer"]
 CODEOWNERS = ["@kahrendt"]
@@ -125,10 +128,10 @@ CONF_THREADSAFE = "threadsafe"
 _MEMORY_LOCATION_VALIDATOR = cv.one_of(*MEMORY_LOCATIONS, lower=True)


-def _maybe_empty_codec(schema):
+def _maybe_empty_codec(schema: cv.Schema) -> Callable[[Any], Any]:
    """Wrap a codec dict schema so that a bare key (None value) is treated as an empty dict."""

-    def validator(value):
+    def validator(value: Any) -> Any:
        if value is None:
            value = {}
        return schema(value)
@@ -200,14 +203,14 @@ def set_stream_limits(
    max_channels: int = cv.UNDEFINED,
    min_sample_rate: int = cv.UNDEFINED,
    max_sample_rate: int = cv.UNDEFINED,
-):
+) -> Callable[[ConfigType], None]:
    """Sets the limits for the audio stream that audio component can handle

    When the component sinks audio (e.g., a speaker), these indicate the limits to the audio it can receive.
    When the component sources audio (e.g., a microphone), these indicate the limits to the audio it can send.
    """

-    def set_limits_in_config(config):
+    def set_limits_in_config(config: ConfigType) -> None:
        if min_bits_per_sample is not cv.UNDEFINED:
            config[CONF_MIN_BITS_PER_SAMPLE] = min_bits_per_sample
        if max_bits_per_sample is not cv.UNDEFINED:
@@ -233,7 +236,7 @@ def final_validate_audio_schema(
    sample_rate: int = cv.UNDEFINED,
    enabled_channels: list[int] = cv.UNDEFINED,
    audio_device_issue: bool = False,
-):
+) -> cv.Schema:
    """Validates audio compatibility when passed between different components.

    The component derived from ``AUDIO_COMPONENT_SCHEMA`` should call ``set_stream_limits`` in a validator to specify its compatible settings
@@ -251,7 +254,7 @@ def final_validate_audio_schema(
        audio_device_issue (bool, optional): Format the error message to indicate the problem is in the configuration for the ``audio_device`` component. Defaults to False.
    """

-    def validate_audio_compatiblity(audio_config):
+    def validate_audio_compatiblity(audio_config: ConfigType) -> ConfigType:
        audio_schema = {}

        if bits_per_sample is not cv.UNDEFINED:
@@ -329,7 +332,7 @@ def _emit_memory_pair(value: str | None, psram_key: str, internal_key: str) -> N
        add_idf_sdkconfig_option(internal_key, True)


-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
    # Re-enable ESP-IDF's HTTP client (excluded by default to save compile time)
    include_builtin_idf_component("esp_http_client")

@@ -371,7 +374,7 @@ async def to_code(config):
            data.wav_support = True

    if data.micro_decoder_support:
-        add_idf_component(name="esphome/micro-decoder", ref="0.2.0")
+        add_idf_component(name="esphome/micro-decoder", ref="0.4.0")

        # All codecs are enabled by default in micro-decoder, so disable the ones that aren't requested to save flash
        if not data.flac_support:
@@ -380,6 +383,8 @@ async def to_code(config):
            add_idf_sdkconfig_option("CONFIG_MICRO_DECODER_CODEC_MP3", False)
        if not data.opus_support:
            add_idf_sdkconfig_option("CONFIG_MICRO_DECODER_CODEC_OPUS", False)
+        # Vorbis is unsupported in ESPHome, so always disable it
+        add_idf_sdkconfig_option("CONFIG_MICRO_DECODER_CODEC_VORBIS", False)
        if not data.wav_support:
            add_idf_sdkconfig_option("CONFIG_MICRO_DECODER_CODEC_WAV", False)

@@ -86,7 +86,7 @@ AudioFileType detect_audio_file_type(const char *content_type, const char *url)
    // Match "audio/ogg" with a codecs parameter containing "opus"
    // Valid forms: audio/ogg;codecs=opus, audio/ogg; codecs="opus", etc.
    // Plain "audio/ogg" without opus is not matched (almost always Ogg Vorbis)
-    if (strncasecmp(content_type, "audio/ogg", 9) == 0 && strcasestr(content_type + 9, "opus") != nullptr) {
+    if (strncasecmp(content_type, "audio/ogg", 9) == 0 && str_contains_ignore_case(content_type + 9, "opus")) {
      return AudioFileType::OPUS;
    }
 #endif
@@ -2,7 +2,9 @@ from esphome import automation
 import esphome.codegen as cg
 import esphome.config_validation as cv
 from esphome.const import CONF_ID, CONF_MIC_GAIN
-from esphome.core import CoroPriority, coroutine_with_priority
+from esphome.core import ID, CoroPriority, coroutine_with_priority
+from esphome.cpp_generator import MockObj, TemplateArgsType
+from esphome.types import ConfigType

 CODEOWNERS = ["@kbx81"]
 IS_PLATFORM_COMPONENT = True
@@ -28,7 +30,12 @@ SET_MIC_GAIN_ACTION_SCHEMA = cv.maybe_simple_value(
    SET_MIC_GAIN_ACTION_SCHEMA,
    synchronous=True,
 )
-async def audio_adc_set_mic_gain_to_code(config, action_id, template_arg, args):
+async def audio_adc_set_mic_gain_to_code(
+    config: ConfigType,
+    action_id: ID,
+    template_arg: cg.TemplateArguments,
+    args: TemplateArgsType,
+) -> MockObj:
    paren = await cg.get_variable(config[CONF_ID])
    var = cg.new_Pvariable(action_id, template_arg, paren)

@@ -39,6 +46,6 @@ async def audio_adc_set_mic_gain_to_code(config, action_id, template_arg, args):


@coroutine_with_priority(CoroPriority.CORE)
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
    cg.add_define("USE_AUDIO_ADC")
    cg.add_global(audio_adc_ns.using)
@@ -3,7 +3,9 @@ from esphome.automation import maybe_simple_id
 import esphome.codegen as cg
 import esphome.config_validation as cv
 from esphome.const import CONF_ID, CONF_VOLUME
-from esphome.core import CoroPriority, coroutine_with_priority
+from esphome.core import ID, CoroPriority, coroutine_with_priority
+from esphome.cpp_generator import MockObj, TemplateArgsType
+from esphome.types import ConfigType

 CODEOWNERS = ["@kbx81"]
 IS_PLATFORM_COMPONENT = True
@@ -37,7 +39,12 @@ SET_VOLUME_ACTION_SCHEMA = cv.maybe_simple_value(
@automation.register_action(
    "audio_dac.mute_on", MuteOnAction, MUTE_ACTION_SCHEMA, synchronous=True
 )
-async def audio_dac_mute_action_to_code(config, action_id, template_arg, args):
+async def audio_dac_mute_action_to_code(
+    config: ConfigType,
+    action_id: ID,
+    template_arg: cg.TemplateArguments,
+    args: TemplateArgsType,
+) -> MockObj:
    paren = await cg.get_variable(config[CONF_ID])
    return cg.new_Pvariable(action_id, template_arg, paren)

@@ -48,7 +55,12 @@ async def audio_dac_mute_action_to_code(config, action_id, template_arg, args):
    SET_VOLUME_ACTION_SCHEMA,
    synchronous=True,
 )
-async def audio_dac_set_volume_to_code(config, action_id, template_arg, args):
+async def audio_dac_set_volume_to_code(
+    config: ConfigType,
+    action_id: ID,
+    template_arg: cg.TemplateArguments,
+    args: TemplateArgsType,
+) -> MockObj:
    paren = await cg.get_variable(config[CONF_ID])
    var = cg.new_Pvariable(action_id, template_arg, paren)

@@ -59,6 +71,6 @@ async def audio_dac_set_volume_to_code(config, action_id, template_arg, args):


@coroutine_with_priority(CoroPriority.CORE)
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
    cg.add_define("USE_AUDIO_DAC")
    cg.add_global(audio_dac_ns.using)
@@ -1,8 +1,6 @@
 #include "b_parasite.h"
 #include "esphome/core/log.h"

-#ifdef USE_ESP32
-
 namespace esphome::b_parasite {

 static const char *const TAG = "b_parasite";
@@ -16,7 +14,7 @@ void BParasite::dump_config() {
  LOG_SENSOR("  ", "Illuminance", this->illuminance_);
 }

-bool BParasite::parse_device(const esp32_ble_tracker::ESPBTDevice &device) {
+bool BParasite::parse_device(const ble_device_base::ESPBTDevice &device) {
  if (device.address_uint64() != address_) {
    ESP_LOGVV(TAG, "parse_device(): unknown MAC address.");
    return false;
@@ -113,5 +111,3 @@ bool BParasite::parse_device(const esp32_ble_tracker::ESPBTDevice &device) {
 }

 }  // namespace esphome::b_parasite
-
-#endif  // USE_ESP32
@@ -2,18 +2,16 @@

 #include "esphome/core/component.h"
 #include "esphome/components/sensor/sensor.h"
-#include "esphome/components/esp32_ble_tracker/esp32_ble_tracker.h"
-
-#ifdef USE_ESP32
+#include "esphome/components/ble_device_base/ble_device.h"

 namespace esphome::b_parasite {

-class BParasite final : public Component, public esp32_ble_tracker::ESPBTDeviceListener {
+class BParasite final : public Component, public ble_device_base::ESPBTDeviceListener {
 public:
  void set_address(uint64_t address) { address_ = address; };
  void set_bindkey(const std::string &bindkey);

-  bool parse_device(const esp32_ble_tracker::ESPBTDevice &device) override;
+  bool parse_device(const ble_device_base::ESPBTDevice &device) override;
  void dump_config() override;

  void set_battery_voltage(sensor::Sensor *battery_voltage) { battery_voltage_ = battery_voltage; }
@@ -35,5 +33,3 @@ class BParasite final : public Component, public esp32_ble_tracker::ESPBTDeviceL
 };

 }  // namespace esphome::b_parasite
-
-#endif  // USE_ESP32
@@ -1,5 +1,5 @@
 import esphome.codegen as cg
-from esphome.components import esp32_ble_tracker, sensor
+from esphome.components import ble_device_base, sensor
 import esphome.config_validation as cv
 from esphome.const import (
    CONF_BATTERY_VOLTAGE,
@@ -23,14 +23,15 @@ from esphome.const import (

 CODEOWNERS = ["@rbaron"]

-DEPENDENCIES = ["esp32_ble_tracker"]
+AUTO_LOAD = ["ble_device_base"]

 b_parasite_ns = cg.esphome_ns.namespace("b_parasite")
 BParasite = b_parasite_ns.class_(
-    "BParasite", esp32_ble_tracker.ESPBTDeviceListener, cg.Component
+    "BParasite", ble_device_base.ESPBTDeviceListener, cg.Component
 )

-CONFIG_SCHEMA = (
+CONFIG_SCHEMA = cv.All(
+    ble_device_base.rename_legacy_hub_id("b_parasite"),
    cv.Schema(
        {
            cv.GenerateID(): cv.declare_id(BParasite),
@@ -68,15 +69,15 @@ CONFIG_SCHEMA = (
            ),
        }
    )
-    .extend(esp32_ble_tracker.ESP_BLE_DEVICE_SCHEMA)
    .extend(cv.COMPONENT_SCHEMA)
+    .extend(ble_device_base.BLE_DEVICE_SCHEMA),
 )


 async def to_code(config):
    var = cg.new_Pvariable(config[CONF_ID])
    await cg.register_component(var, config)
-    await esp32_ble_tracker.register_ble_device(var, config)
+    await ble_device_base.register_ble_device(var, config)

    cg.add(var.set_address(config[CONF_MAC_ADDRESS].as_hex))

@@ -2,6 +2,8 @@ import esphome.codegen as cg
 from esphome.components import ble_client, time
 import esphome.config_validation as cv
 from esphome.const import CONF_ID, CONF_RECEIVE_TIMEOUT, CONF_TIME_ID
+from esphome.cpp_generator import MockObj
+from esphome.types import ConfigType

 CODEOWNERS = ["@jhansche"]
 DEPENDENCIES = ["ble_client"]
@@ -32,12 +34,12 @@ BEDJET_CLIENT_SCHEMA = cv.Schema(
 )


-async def register_bedjet_child(var, config):
+async def register_bedjet_child(var: MockObj, config: ConfigType) -> None:
    parent = await cg.get_variable(config[CONF_BEDJET_ID])
    cg.add(parent.register_child(var))


-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
    var = cg.new_Pvariable(config[CONF_ID])
    await cg.register_component(var, config)
    await ble_client.register_ble_node(var, config)
@@ -2,6 +2,7 @@ import esphome.codegen as cg
 from esphome.components import climate
 import esphome.config_validation as cv
 from esphome.const import CONF_HEAT_MODE, CONF_TEMPERATURE_SOURCE
+from esphome.types import ConfigType

 from .. import BEDJET_CLIENT_SCHEMA, bedjet_ns, register_bedjet_child

@@ -37,7 +38,7 @@ CONFIG_SCHEMA = (
 )


-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
    var = await climate.new_climate(config)
    await cg.register_component(var, config)
    await register_bedjet_child(var, config)
@@ -1,6 +1,7 @@
 import esphome.codegen as cg
 from esphome.components import fan
 import esphome.config_validation as cv
+from esphome.types import ConfigType

 from .. import BEDJET_CLIENT_SCHEMA, bedjet_ns, register_bedjet_child

@@ -16,7 +17,7 @@ CONFIG_SCHEMA = (
 )


-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
    var = await fan.new_fan(config)
    await cg.register_component(var, config)
    await register_bedjet_child(var, config)
@@ -9,6 +9,7 @@ from esphome.const import (
    STATE_CLASS_MEASUREMENT,
    UNIT_CELSIUS,
 )
+from esphome.types import ConfigType

 from .. import BEDJET_CLIENT_SCHEMA, bedjet_ns, register_bedjet_child

@@ -38,7 +39,7 @@ CONFIG_SCHEMA = cv.Schema(
 ).extend(BEDJET_CLIENT_SCHEMA)


-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
    var = cg.new_Pvariable(config[CONF_ID])
    await cg.register_component(var, config)
    await register_bedjet_child(var, config)
@@ -300,46 +300,12 @@ void BekenSPILEDStripLightOutput::write_state(light::LightState *state) {
 }

 light::ESPColorView BekenSPILEDStripLightOutput::get_view_internal(int32_t index) const {
-  int32_t r = 0, g = 0, b = 0;
-  switch (this->rgb_order_) {
-    case ORDER_RGB:
-      r = 0;
-      g = 1;
-      b = 2;
-      break;
-    case ORDER_RBG:
-      r = 0;
-      g = 2;
-      b = 1;
-      break;
-    case ORDER_GRB:
-      r = 1;
-      g = 0;
-      b = 2;
-      break;
-    case ORDER_GBR:
-      r = 2;
-      g = 0;
-      b = 1;
-      break;
-    case ORDER_BGR:
-      r = 2;
-      g = 1;
-      b = 0;
-      break;
-    case ORDER_BRG:
-      r = 1;
-      g = 2;
-      b = 0;
-      break;
-  }
-  uint8_t multiplier = this->is_rgbw_ || this->is_wrgb_ ? 4 : 3;
-  uint8_t white = this->is_wrgb_ ? 0 : 3;
-
-  return {this->buf_ + (index * multiplier) + r + this->is_wrgb_,
-          this->buf_ + (index * multiplier) + g + this->is_wrgb_,
-          this->buf_ + (index * multiplier) + b + this->is_wrgb_,
-          this->is_rgbw_ || this->is_wrgb_ ? this->buf_ + (index * multiplier) + white : nullptr,
+  const light::ChannelColors &colors = this->channel_colors_;
+  uint8_t *led = this->buf_ + (index * colors.bytes_per_led());
+  return {led + colors.r,
+          led + colors.g,
+          led + colors.b,
+          colors.has_white() ? led + colors.w : nullptr,
          &this->effect_data_[index],
          &this->correction_};
 }
@@ -349,35 +315,12 @@ void BekenSPILEDStripLightOutput::dump_config() {
                "Beken SPI LED Strip:\n"
                "  Pin: %u",
                this->pin_);
-  const char *rgb_order;
-  switch (this->rgb_order_) {
-    case ORDER_RGB:
-      rgb_order = "RGB";
-      break;
-    case ORDER_RBG:
-      rgb_order = "RBG";
-      break;
-    case ORDER_GRB:
-      rgb_order = "GRB";
-      break;
-    case ORDER_GBR:
-      rgb_order = "GBR";
-      break;
-    case ORDER_BGR:
-      rgb_order = "BGR";
-      break;
-    case ORDER_BRG:
-      rgb_order = "BRG";
-      break;
-    default:
-      rgb_order = "UNKNOWN";
-      break;
-  }
+  char channel_colors[5];
  ESP_LOGCONFIG(TAG,
-                "  RGB Order: %s\n"
+                "  Channel colors: %s\n"
                "  Max refresh rate: %" PRIu32 "\n"
                "  Number of LEDs: %u",
-                rgb_order, this->max_refresh_rate_.value_or(0), this->num_leds_);
+                this->channel_colors_.to_string(channel_colors), this->max_refresh_rate_.value_or(0), this->num_leds_);
 }

 float BekenSPILEDStripLightOutput::get_setup_priority() const { return setup_priority::HARDWARE; }
@@ -3,6 +3,7 @@
 #ifdef USE_BK72XX

 #include "esphome/components/light/addressable_light.h"
+#include "esphome/components/light/channel_colors.h"
 #include "esphome/components/light/light_output.h"
 #include "esphome/core/color.h"
 #include "esphome/core/component.h"
@@ -10,15 +11,6 @@

 namespace esphome::beken_spi_led_strip {

-enum RGBOrder : uint8_t {
-  ORDER_RGB,
-  ORDER_RBG,
-  ORDER_GRB,
-  ORDER_GBR,
-  ORDER_BGR,
-  ORDER_BRG,
-};
-
 class BekenSPILEDStripLightOutput final : public light::AddressableLight {
 public:
  void setup() override;
@@ -28,7 +20,7 @@ class BekenSPILEDStripLightOutput final : public light::AddressableLight {
  int32_t size() const override { return this->num_leds_; }
  light::LightTraits get_traits() override {
    auto traits = light::LightTraits();
-    if (this->is_rgbw_ || this->is_wrgb_) {
+    if (this->channel_colors_.has_white()) {
      traits.set_supported_color_modes({light::ColorMode::RGB_WHITE, light::ColorMode::WHITE});
    } else {
      traits.set_supported_color_modes({light::ColorMode::RGB});
@@ -38,16 +30,13 @@ class BekenSPILEDStripLightOutput final : public light::AddressableLight {

  void set_pin(uint8_t pin) { this->pin_ = pin; }
  void set_num_leds(uint16_t num_leds) { this->num_leds_ = num_leds; }
-  void set_is_rgbw(bool is_rgbw) { this->is_rgbw_ = is_rgbw; }
-  void set_is_wrgb(bool is_wrgb) { this->is_wrgb_ = is_wrgb; }
+  void set_channel_colors(light::ChannelColors channel_colors) { this->channel_colors_ = channel_colors; }

  /// Set a maximum refresh rate in µs as some lights do not like being updated too often.
  void set_max_refresh_rate(uint32_t interval_us) { this->max_refresh_rate_ = interval_us; }

  void set_led_params(uint8_t bit0, uint8_t bit1, uint32_t spi_frequency);

-  void set_rgb_order(RGBOrder rgb_order) { this->rgb_order_ = rgb_order; }
-
  void clear_effect_data() override {
    for (int i = 0; i < this->size(); i++)
      this->effect_data_[i] = 0;
@@ -58,7 +47,7 @@ class BekenSPILEDStripLightOutput final : public light::AddressableLight {
 protected:
  light::ESPColorView get_view_internal(int32_t index) const override;

-  size_t get_buffer_size_() const { return this->num_leds_ * (this->is_rgbw_ || this->is_wrgb_ ? 4 : 3); }
+  size_t get_buffer_size_() const { return this->num_leds_ * this->channel_colors_.bytes_per_led(); }

  uint8_t *buf_{nullptr};
  uint8_t *effect_data_{nullptr};
@@ -66,13 +55,11 @@ class BekenSPILEDStripLightOutput final : public light::AddressableLight {

  uint8_t pin_;
  uint16_t num_leds_;
-  bool is_rgbw_;
-  bool is_wrgb_;

  uint32_t spi_frequency_{6666666};
  uint8_t bit0_{0xE0};
  uint8_t bit1_{0xFC};
-  RGBOrder rgb_order_;
+  light::ChannelColors channel_colors_{0, 1, 2, light::ChannelColors::NO_WHITE};

  uint32_t last_refresh_{0};
  optional<uint32_t> max_refresh_rate_{};
@@ -3,6 +3,7 @@ from dataclasses import dataclass
 from esphome import pins
 import esphome.codegen as cg
 from esphome.components import libretiny, light
+from esphome.components.const import CONF_CHANNEL_COLORS, CONF_IS_WRGB
 import esphome.config_validation as cv
 from esphome.const import (
    CONF_CHIPSET,
@@ -13,6 +14,7 @@ from esphome.const import (
    CONF_PIN,
    CONF_RGB_ORDER,
 )
+from esphome.types import ConfigType

 CODEOWNERS = ["@Mat931"]
 DEPENDENCIES = ["libretiny"]
@@ -22,17 +24,6 @@ BekenSPILEDStripLightOutput = beken_spi_led_strip_ns.class_(
    "BekenSPILEDStripLightOutput", light.AddressableLight
 )

-RGBOrder = beken_spi_led_strip_ns.enum("RGBOrder")
-
-RGB_ORDERS = {
-    "RGB": RGBOrder.ORDER_RGB,
-    "RBG": RGBOrder.ORDER_RBG,
-    "GRB": RGBOrder.ORDER_GRB,
-    "GBR": RGBOrder.ORDER_GBR,
-    "BGR": RGBOrder.ORDER_BGR,
-    "BRG": RGBOrder.ORDER_BRG,
-}
-

@dataclass
 class LEDStripTimings:
@@ -57,8 +48,6 @@ CHIPSETS = {
 }


-CONF_IS_WRGB = "is_wrgb"
-
 SUPPORTED_PINS = {
    libretiny.const.FAMILY_BK7231N: [16],
    libretiny.const.FAMILY_BK7231T: [16],
@@ -79,10 +68,9 @@ def _validate_pin(value):
    return value


-def _validate_num_leds(value):
-    max_num_leds = 165  # 170
-    if value[CONF_IS_RGBW] or value[CONF_IS_WRGB]:
-        max_num_leds = 123  # 127
+def _validate_num_leds(value: ConfigType) -> ConfigType:
+    # A white channel makes each LED one byte wider, so fewer of them fit in the DMA buffer.
+    max_num_leds = 123 if "W" in value[CONF_CHANNEL_COLORS] else 165  # 127 / 170
    if value[CONF_NUM_LEDS] > max_num_leds:
        raise cv.Invalid(
            f"The maximum number of LEDs for this configuration is {max_num_leds}.",
@@ -99,18 +87,23 @@ CONFIG_SCHEMA = cv.All(
                pins.internal_gpio_output_pin_number, _validate_pin
            ),
            cv.Required(CONF_NUM_LEDS): cv.positive_not_null_int,
-            cv.Required(CONF_RGB_ORDER): cv.enum(RGB_ORDERS, upper=True),
+            cv.Optional(CONF_CHANNEL_COLORS): light.validate_channel_colors,
+            # Deprecated in favour of CONF_CHANNEL_COLORS, remove in 2027.3.0
+            cv.Optional(CONF_RGB_ORDER): cv.one_of(*light.RGB_ORDERS, upper=True),
+            cv.Optional(CONF_IS_RGBW): cv.boolean,
+            cv.Optional(CONF_IS_WRGB): cv.boolean,
            cv.Optional(CONF_MAX_REFRESH_RATE): cv.positive_time_period_microseconds,
            cv.Required(CONF_CHIPSET): cv.one_of(*CHIPSETS, upper=True),
-            cv.Optional(CONF_IS_RGBW, default=False): cv.boolean,
-            cv.Optional(CONF_IS_WRGB, default=False): cv.boolean,
        }
    ),
+    light.migrate_channel_colors(
+        removed_in="2027.3.0", component="beken_spi_led_strip"
+    ),
    _validate_num_leds,
 )


-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
    var = cg.new_Pvariable(config[CONF_OUTPUT_ID])
    await light.register_light(var, config)
    await cg.register_component(var, config)
@@ -130,6 +123,6 @@ async def to_code(config):
        )
    )

-    cg.add(var.set_rgb_order(config[CONF_RGB_ORDER]))
-    cg.add(var.set_is_rgbw(config[CONF_IS_RGBW]))
-    cg.add(var.set_is_wrgb(config[CONF_IS_WRGB]))
+    cg.add(
+        var.set_channel_colors(light.channel_colors_struct(config[CONF_CHANNEL_COLORS]))
+    )
@@ -5,6 +5,7 @@ from esphome.automation import Condition, maybe_simple_id
 import esphome.codegen as cg
 from esphome.components import mqtt, web_server, zigbee
 from esphome.components.const import CONF_ON_STATE_CHANGE
+from esphome.config_helpers import filter_source_files_from_defines
 import esphome.config_validation as cv
 from esphome.const import (
    CONF_DELAY,
@@ -560,6 +561,11 @@ _CALLBACK_AUTOMATIONS = (
 async def _build_binary_sensor_automations(var, config):
    await automation.build_callback_automations(var, config, _CALLBACK_AUTOMATIONS)

+    if config.get(CONF_ON_CLICK) or config.get(CONF_ON_DOUBLE_CLICK):
+        cg.add_define("USE_BINARY_SENSOR_CLICK_TRIGGER")
+    if config.get(CONF_ON_MULTI_CLICK):
+        cg.add_define("USE_BINARY_SENSOR_MULTI_CLICK_TRIGGER")
+
    for conf in config.get(CONF_ON_CLICK, []):
        trigger = cg.new_Pvariable(
            conf[CONF_TRIGGER_ID], var, conf[CONF_MIN_LENGTH], conf[CONF_MAX_LENGTH]
@@ -673,3 +679,15 @@ async def to_code(config):
 async def binary_sensor_invalidate_state_to_code(config, action_id, template_arg, args):
    paren = await cg.get_variable(config[CONF_ID])
    return cg.new_Pvariable(action_id, template_arg, paren)
+
+
+# automation.cpp only implements the click/double_click/multi_click triggers
+FILTER_SOURCE_FILES = filter_source_files_from_defines(
+    {
+        "automation.cpp": (
+            "USE_BINARY_SENSOR_CLICK_TRIGGER",
+            "USE_BINARY_SENSOR_MULTI_CLICK_TRIGGER",
+        ),
+        "filter.cpp": "USE_BINARY_SENSOR_FILTER",
+    }
+)
@@ -1,8 +1,13 @@
+#include "esphome/core/defines.h"
+#if defined(USE_BINARY_SENSOR_CLICK_TRIGGER) || defined(USE_BINARY_SENSOR_MULTI_CLICK_TRIGGER)
+
 #include "automation.h"
 #include "esphome/core/log.h"

 namespace esphome::binary_sensor {

+#ifdef USE_BINARY_SENSOR_MULTI_CLICK_TRIGGER
+
 static const char *const TAG = "binary_sensor.automation";

 // MultiClickTrigger timeout IDs.
@@ -120,6 +125,9 @@ void MultiClickTriggerBase::trigger_() {
  this->trigger();
 }

+#endif  // USE_BINARY_SENSOR_MULTI_CLICK_TRIGGER
+
+#ifdef USE_BINARY_SENSOR_CLICK_TRIGGER
 bool match_interval(uint32_t min_length, uint32_t max_length, uint32_t length) {
  if (max_length == 0) {
    return length >= min_length;
@@ -127,4 +135,8 @@ bool match_interval(uint32_t min_length, uint32_t max_length, uint32_t length) {
    return length >= min_length && length <= max_length;
  }
 }
+#endif  // USE_BINARY_SENSOR_CLICK_TRIGGER
+
 }  // namespace esphome::binary_sensor
+
+#endif  // USE_BINARY_SENSOR_CLICK_TRIGGER || USE_BINARY_SENSOR_MULTI_CLICK_TRIGGER
@@ -51,7 +51,11 @@ def _set_core_data(config):
    return config


-CONFIG_SCHEMA = libretiny.BASE_SCHEMA
+# extend({}) makes this platform's own schema instance: BASE_SCHEMA is shared
+# by every LibreTiny platform, and prepending this platform's _set_core_data
+# onto the shared object would run it for every platform's validation once two
+# platform modules are imported in one process (device-builder, tests).
+CONFIG_SCHEMA = libretiny.BASE_SCHEMA.extend({})

 PIN_SCHEMA = libretiny.gpio.BASE_PIN_SCHEMA

@@ -4,12 +4,15 @@ The platform analog of esp32_ble / rp2040_ble: owns the Beken BDK BLE stack
 bring-up and the controller BLE address. Consumers (bk72xx_ble_tracker) build
 on this component and contain no SDK calls of their own.

-Supported SoCs (BLE 5.x): BK7231N/BK7236 (BLE 5.1), BK7238/BK7252N/BK7253
-(BLE 5.2), and any future BLE-5.x SoC. Capability is detected at compile time,
-not by a chip list: the C++ guards on `__has_include("ble_api.h")` — the Beken
-BLE 5.x public API header, which the LibreTiny beken-72xx builder ships only
-for BLE-5.x SoCs. BK7231T/BK7251/BK7271 (BLE 4.2) and BK7231Q (no BLE) fail
-with a clear #error.
+Supported SoCs (BLE 5.x): BK7231N/BK7236 (BLE 5.1), BK7252N/BK7253 (BLE 5.2),
+and any future BLE-5.x SoC. BK7238 (BLE 5.2) is blocked for now: with BLE
+compiled in, the Beken SDK erases the bootloader flash sector at boot because
+LibreTiny's partition table has no BLE bonding entry (esphome#18646,
+libretiny-eu/libretiny#408). Known non-5.x families and BK7238 are rejected in
+to_code. Unknown families are capability-checked at compile time via
+`__has_include("app_ble.h")`, a header only on the BLE 5.x include path
+(ble_api.h ships for every SoC, so it cannot be the probe). A non-5.x build
+fails with a clear #error.

 No framework patch is needed: the LibreTiny beken-72xx builder already compiles
 and links the BLE 5.x stack (CFG_SUPPORT_BLE=1 + CFG_BLE_VERSION=BLE_VERSION_5_x;
@@ -21,9 +24,16 @@ import logging

 import esphome.codegen as cg
 from esphome.components import libretiny
-from esphome.components.libretiny.const import FAMILY_BK7231N, FAMILY_BK7238
+from esphome.components.libretiny.const import (
+    FAMILY_BK7231N,
+    FAMILY_BK7231Q,
+    FAMILY_BK7231T,
+    FAMILY_BK7238,
+    FAMILY_BK7251,
+)
 import esphome.config_validation as cv
 from esphome.const import CONF_ENABLE_ON_BOOT, CONF_ID
+from esphome.core import EsphomeError
 from esphome.types import ConfigType

 DEPENDENCIES = ["bk72xx"]
@@ -45,7 +55,44 @@ CONFIG_SCHEMA = cv.Schema(
 ).extend(cv.COMPONENT_SCHEMA)


+# Once per registered scan listener; sizes the controller's StaticVector
+# listener storage.
+request_scan_listener_slot = cg.slot_counter("BK72XX_BLE_SCAN_LISTENER_COUNT")
+
+
+def _unsupported_family_message(family: str) -> str | None:
+    if family in (FAMILY_BK7231T, FAMILY_BK7251):
+        return (
+            f"bk72xx_ble does not support {family}: this SoC has the Beken BLE 4.2 "
+            "stack; a BLE 5.x SoC such as BK7231N or BK7238 is required"
+        )
+    if family == FAMILY_BK7231Q:
+        return "bk72xx_ble does not support BK7231Q: this SoC has no BLE"
+    if family == FAMILY_BK7238:
+        return (
+            "bk72xx_ble is disabled on BK7238: with BLE compiled in, the Beken SDK "
+            "erases the bootloader flash sector at boot and the device can no longer "
+            "start (see https://github.com/esphome/esphome/issues/18646); support "
+            "returns once the LibreTiny partition table fix "
+            "(libretiny-eu/libretiny#408) is released"
+        )
+    return None
+
+
+def _final_validate(config: ConfigType) -> None:
+    # Warn only: a hard error here would break the validate-only CI fixtures,
+    # which run on a BLE 4.2 board. The hard error is raised at codegen.
+    if msg := _unsupported_family_message(libretiny.get_libretiny_family()):
+        _LOGGER.warning("%s (this configuration cannot compile)", msg)
+
+
+FINAL_VALIDATE_SCHEMA = _final_validate
+
+
 async def to_code(config: ConfigType) -> None:
+    if msg := _unsupported_family_message(libretiny.get_libretiny_family()):
+        raise EsphomeError(msg)
+
    var = cg.new_Pvariable(config[CONF_ID])
    await cg.register_component(var, config)

@@ -77,18 +124,7 @@ async def to_code(config: ConfigType) -> None:
    # BK7231N, but NOT on BK7238 (its BLE stack has no such symbol; the address is
    # derived from the WiFi MAC instead — the BDK's own fallback). Tell the C++
    # which path is available so it doesn't reference a missing symbol.
-    family = libretiny.get_libretiny_family()
-    if family == FAMILY_BK7231N:
+    if libretiny.get_libretiny_family() == FAMILY_BK7231N:
        cg.add_define("BK72XX_BLE_HAS_COMMON_BDADDR")
-    elif family == FAMILY_BK7238:
-        # ESPHome's LibreTiny disables BLE on BK7238 because the SDK can hang at
-        # WiFi STA startup when BLE init runs. This component re-enables BLE, so
-        # warn loudly: BK7238 is accepted but not hardware-verified and may be
-        # WiFi-unstable with BLE on.
-        _LOGGER.warning(
-            "bk72xx_ble on BK7238: enabling BLE is known to risk a WiFi STA startup "
-            "hang on this family and is not yet hardware-verified. Expect possible "
-            "instability."
-        )

    cg.add_define("USE_BK72XX_BLE")
@@ -0,0 +1,119 @@
+// Every SDK call the scan reconciler makes. The BDK's own start hardcodes
+// passive (the active bit is commented out in both stacks), so
+// bdk_scan_start() packs the GAPM_ACTIVITY_START_CMD itself, field-for-field
+// the SDK's app_ble_start_scaning() except that prop takes the mode, armed
+// through the SDK's own operation bookkeeping. The component pins
+// beken-bdk 3.0.78; the static asserts catch a layout change on a bump.
+
+#include "bdk_scan.h"
+
+#ifdef USE_BK72XX_BLE
+
+// Same SDK gate as bk72xx_ble.cpp (which carries the explanatory #error).
+#if !defined(CLANG_TIDY) && __has_include("ble_api.h") && __has_include("app_ble.h")
+
+extern "C" {
+#include "app_ble.h"     // app_ble_env, app_ble_run, app_ble_reset, actv_state_t,
+                         // app_ble_actv_state_get, app_ble_env_state_get,
+                         // app_ble_get_idle_actv_idx_handle, UNKNOW_ACT_IDX,
+                         // bk_ble_* (via ble_api_5_x.h)
+#include "kernel_msg.h"  // KERNEL_MSG_ALLOC, kernel_msg_send
+#if __has_include("gapm_msg.h")
+#include "gapm_msg.h"  // BLE 5.2 (BK7238/BK7252N): gapm_activity_start_cmd, GAPM_SCAN_*
+#else
+#include "gapm_task.h"  // BLE 5.1 (BK7231N/BK7236): same declarations, older header name
+#endif
+}
+
+#include "esphome/core/log.h"
+
+namespace esphome::bk72xx_ble {
+
+static const char *const TAG = "bk72xx_ble";
+
+// Pin the SDK surface this file depends on: a beken-bdk bump that moves these
+// must fail the build, not corrupt the kernel message.
+static_assert(GAPM_SCAN_PROP_PHY_1M_BIT == (1 << 0) && GAPM_SCAN_PROP_ACTIVE_1M_BIT == (1 << 2) &&
+                  sizeof(struct gapm_scan_param) == 16 && sizeof(struct gapm_scan_wd_op_param) == 4,
+              "beken-bdk GAPM scan layout changed; revalidate bdk_scan_start() "
+              "against the SDK's app_ble_start_scaning()");
+static_assert(INVALID_ACTIVITY_IDX == UNKNOW_ACT_IDX,
+              "beken-bdk activity sentinel changed; revalidate the scan reconciler");
+static_assert(GAPM_REPORT_TYPE_SCAN_RSP_EXT == 2 && GAPM_REPORT_TYPE_SCAN_RSP_LEG == 3 &&
+                  GAPM_REPORT_INFO_SCAN_ADV_BIT == (1 << 5),
+              "beken-bdk GAPM report info changed; revalidate the tracker's demux constants");
+
+bool bdk_scan_ready() { return app_ble_env_state_get() == APP_BLE_READY; }
+
+BdkActivityState bdk_scan_state(uint8_t activity_idx) {
+  if (activity_idx == INVALID_ACTIVITY_IDX)
+    return BdkActivityState::IDLE;
+  switch (app_ble_actv_state_get(activity_idx)) {
+    case ACTV_IDLE:
+      return BdkActivityState::IDLE;
+    case ACTV_SCAN_CREATED:
+      return BdkActivityState::CREATED;
+    case ACTV_SCAN_STARTED:
+      return BdkActivityState::STARTED;
+    default:
+      return BdkActivityState::OTHER;
+  }
+}
+
+uint8_t bdk_scan_acquire_activity() {
+  uint8_t idx = app_ble_get_idle_actv_idx_handle(SCAN_ACTV);
+  if (idx == INVALID_ACTIVITY_IDX)
+    ESP_LOGE(TAG, "Scan start failed: no idle activity handle");
+  return idx;
+}
+
+BdkOpResult bdk_scan_create(uint8_t activity_idx) {
+  ble_err_t ret = bk_ble_create_scaning(activity_idx, nullptr);
+  if (ret == ERR_SUCCESS)
+    return BdkOpResult::OK;
+  if (ret == ERR_BLE_STATUS)
+    return BdkOpResult::BUSY;
+  ESP_LOGE(TAG, "Scan activity create failed (err %d)", static_cast<int>(ret));
+  return BdkOpResult::FAILED;
+}
+
+BdkOpResult bdk_scan_start(uint8_t activity_idx, uint16_t interval, uint16_t window, bool active) {
+  app_ble_run(activity_idx, BLE_START_SCAN, 1 << BLE_OP_START_SCAN_POS, nullptr);
+  struct gapm_activity_start_cmd *cmd =
+      KERNEL_MSG_ALLOC(GAPM_ACTIVITY_START_CMD, TASK_BLE_GAPM, TASK_BLE_APP, gapm_activity_start_cmd);
+  if (cmd == nullptr) {
+    app_ble_reset();  // the SDK's own failure path for an unsent operation
+    ESP_LOGE(TAG, "Scan start failed: kernel message allocation");
+    return BdkOpResult::FAILED;
+  }
+  cmd->operation = GAPM_START_ACTIVITY;
+  cmd->actv_idx = app_ble_env.actvs[activity_idx].gap_advt_idx;
+  cmd->u_param.scan_param.type = GAPM_SCAN_TYPE_OBSERVER;
+  cmd->u_param.scan_param.prop = GAPM_SCAN_PROP_PHY_1M_BIT | (active ? GAPM_SCAN_PROP_ACTIVE_1M_BIT : 0);
+  cmd->u_param.scan_param.scan_param_1m.scan_intv = interval;
+  cmd->u_param.scan_param.scan_param_1m.scan_wd = window;
+  cmd->u_param.scan_param.scan_param_coded.scan_intv = 0;
+  cmd->u_param.scan_param.scan_param_coded.scan_wd = 0;
+  cmd->u_param.scan_param.dup_filt_pol = 0;
+  cmd->u_param.scan_param.rsvd = 0;
+  cmd->u_param.scan_param.duration = 0;  // scan until stopped
+  cmd->u_param.scan_param.period = 10;   // matches the SDK's passive start
+  kernel_msg_send(cmd);
+  return BdkOpResult::OK;
+}
+
+BdkOpResult bdk_scan_release(uint8_t activity_idx, bool created, int *err_out) {
+  ble_err_t ret = created ? bk_ble_delete_scaning(activity_idx, nullptr) : bk_ble_scan_stop(activity_idx, nullptr);
+  *err_out = static_cast<int>(ret);
+  if (ret == ERR_SUCCESS)
+    return BdkOpResult::OK;
+  // DEBUG on purpose: the reconciler WARNs once per streak and the stuck
+  // ERROR carries this code — a per-retry ERROR would be unbounded.
+  ESP_LOGD(TAG, "Scan release %s (err %d)", ret == ERR_BLE_STATUS ? "rejected" : "failed", static_cast<int>(ret));
+  return ret == ERR_BLE_STATUS ? BdkOpResult::BUSY : BdkOpResult::FAILED;
+}
+
+}  // namespace esphome::bk72xx_ble
+
+#endif  // !CLANG_TIDY && ble_api.h && app_ble.h
+#endif  // USE_BK72XX_BLE
@@ -0,0 +1,51 @@
+#pragma once
+
+#include "esphome/core/defines.h"
+
+#ifdef USE_BK72XX_BLE
+
+#include <cstdint>
+
+namespace esphome::bk72xx_ble {
+
+/// Activity index value marking "no scan activity", the BDK's own convention
+/// (asserted against its symbol in bdk_scan.cpp).
+inline constexpr uint8_t INVALID_ACTIVITY_IDX = 0xFF;
+
+/// Scan-relevant controller activity states, read live from the SDK.
+enum class BdkActivityState : uint8_t {
+  IDLE,     ///< No activity (or one whose create failed).
+  CREATED,  ///< Created but not started.
+  STARTED,  ///< Scanning.
+  OTHER,    ///< A non-scan or transitional state; settles on a later read.
+};
+
+/// Outcome of a BDK scan operation request.
+enum class BdkOpResult : uint8_t {
+  OK,      ///< Accepted; completion is asynchronous.
+  BUSY,    ///< Another controller operation is in flight; retry later.
+  FAILED,  ///< Rejected.
+};
+
+/// True when no controller operation is in flight (APP_BLE_READY).
+bool bdk_scan_ready();
+/// Live state of the given activity; INVALID_ACTIVITY_IDX reads as IDLE.
+BdkActivityState bdk_scan_state(uint8_t activity_idx);
+/// Claim an idle activity slot; INVALID_ACTIVITY_IDX when none is free.
+uint8_t bdk_scan_acquire_activity();
+/// Create the scan activity (asynchronous); started once CREATED is observed.
+BdkOpResult bdk_scan_create(uint8_t activity_idx);
+/// Start a created activity: the packed GAPM start, taking the scan mode the
+/// BDK's own start path hardcodes away. Fire-and-forget; FAILED when the
+/// kernel message could not be allocated (the armed SDK operation is rolled
+/// back).
+BdkOpResult bdk_scan_start(uint8_t activity_idx, uint16_t interval, uint16_t window, bool active);
+/// Release the activity: delete when never started (a stop would be
+/// rejected), stop otherwise. BUSY on a transient rejection (retry), FAILED
+/// on any other error; err_out receives the SDK code (0 on success).
+/// Teardown is asynchronous — observe IDLE to confirm.
+BdkOpResult bdk_scan_release(uint8_t activity_idx, bool created, int *err_out);
+
+}  // namespace esphome::bk72xx_ble
+
+#endif  // USE_BK72XX_BLE
@@ -5,7 +5,8 @@
 // talks to the Beken BDK BLE stack:
 //   - one-time stack bring-up (ble_set_notice_cb() + ble_entry()),
 //   - the controller BLE address,
-//   - the raw controller scan primitives (bk_ble_scan_start/stop),
+//   - the scan reconciler (request, pacing, bring-up budget) over the
+//     bdk_scan surface,
 //   - the scan-report ring: the BDK notice callback (BLE task) takes a report
 //     from a fixed pool and pushes it on a lock-free SPSC queue; loop() drains,
 //     dispatches on the main task and returns reports to the pool — the same
@@ -20,32 +21,39 @@

 #include "bk72xx_ble.h"  // pulls esphome/core/defines.h for USE_BK72XX_BLE

+#include "bdk_scan.h"  // the raw BDK scan surface (state reads, starts, release)
+
 #ifdef USE_BK72XX_BLE

 #include <cstring>

+#include "esphome/core/application.h"
 #include "esphome/core/hal.h"
 #include "esphome/core/helpers.h"  // get_mac_address_raw()
 #include "esphome/core/log.h"

 // ---------------------------------------------------------------------------
 // SDK-capability gate (not a chip allowlist).
-// This component drives the Beken BLE *5.x* controller via its public API,
-// `ble_api.h`, which the LibreTiny beken-72xx builder ships only for the
-// BLE-5.x SoCs (it selects the `ble_pub` 5.x stack from CFG_BLE_VERSION; the
-// 4.2 SoCs build a different, older API with no ble_api.h). Gate on the header
-// itself so any BLE-5.x Beken chip — present or future — is supported without a
-// hard-coded list, and a non-5.x build fails here with a clear message instead
-// of a cryptic "ble_api.h: No such file or directory".
+// This component drives the Beken BLE *5.x* controller. `ble_api.h` cannot be
+// the probe: it ships for every SoC (driver/include) and merely switches on
+// CFG_BLE_VERSION internally. `app_ble.h` is on the include path only when the
+// LibreTiny beken-72xx builder selects a 5.x stack, so gating on it supports
+// any BLE-5.x chip — present or future — without a hard-coded list, and a
+// non-5.x build fails here with a clear message instead of a cryptic
+// "app_ble.h: No such file or directory".
 // ---------------------------------------------------------------------------
 #if defined(CLANG_TIDY)
 // The clang-tidy environment does not carry the full Beken BDK BLE 5.x API
 // (its ble_api.h variant lacks parts of the 5.x surface), so there is nothing
 // accurate to analyze the SDK calls against — skip the file under analysis.
 #define BK72XX_BLE_NO_SDK
-#elif !__has_include("ble_api.h")
+#elif !__has_include("ble_api.h") || !__has_include("app_ble.h")
+// Also skip the SDK body: #error does not stop the preprocessor, and on a 4.2
+// SoC ble_api.h exists, so without the guard the 5.x symbols would fail one by
+// one and bury this message.
+#define BK72XX_BLE_NO_SDK
 #error \
-    "bk72xx_ble requires a BLE 5.x Beken SDK (ble_api.h). Supported SoCs: BK7231N/BK7236 (BLE 5.1) and BK7238/BK7252N/BK7253 (BLE 5.2). BK7231T/BK7251/BK7271 (BLE 4.2) and BK7231Q (no BLE) are not supported."
+    "bk72xx_ble requires a BLE 5.x Beken SDK (app_ble.h). Supported SoCs: BK7231N/BK7236 (BLE 5.1) and BK7238/BK7252N/BK7253 (BLE 5.2). BK7231T/BK7251/BK7271 (BLE 4.2) and BK7231Q (no BLE) are not supported."
 #endif

 #ifndef BK72XX_BLE_NO_SDK
@@ -57,9 +65,8 @@
 // are C headers consumed from C++ (a standard C-header-from-C++ pattern).
 // ---------------------------------------------------------------------------
 extern "C" {
-#include "ble_api.h"  // bk_ble_scan_start/stop, ble_entry, ble_set_notice_cb,
-                      // app_ble_get_idle_actv_idx_handle, struct scan_param,
-                      // recv_adv_t, ble_notice_t, BLE_5_REPORT_ADV, SCAN_ACTV
+#include "ble_api.h"  // ble_set_notice_cb, recv_adv_t, ble_notice_t,
+                      // BLE_5_REPORT_ADV (scan primitives live in bdk_scan.cpp)
 #ifdef BK72XX_BLE_HAS_COMMON_BDADDR
 #include "common_bt_defines.h"  // struct bd_addr
 // The controller's public BLE address, populated by the BDK during ble_entry().
@@ -76,6 +83,12 @@ namespace esphome::bk72xx_ble {

 static const char *const TAG = "bk72xx_ble";

+static constexpr uint32_t RECONCILE_RETRY_MS = 10;              // pump floor for fast loops
+static constexpr uint32_t RECONCILE_REJECTED_RETRY_MS = 500;    // retry gate after a rejected release
+static constexpr uint32_t RECONCILE_PENDING_TIMEOUT_MS = 2000;  // bring-up budget before FAILED
+static constexpr uint32_t SCAN_LIVENESS_CHECK_MS = 1000;        // settled-scan re-check cadence
+static constexpr uint32_t TEARDOWN_STUCK_ERROR_MS = 30000;      // stuck-teardown ERROR (stop also goes FAILED)
+
 // The BDK notice callback is a plain C function pointer with no user argument,
 // so it reaches the (single) component instance through a file-static pointer.
 static BK72xxBLE *s_ble = nullptr;  // NOLINT(cppcoreguidelines-avoid-non-const-global-variables)
@@ -95,21 +108,22 @@ static void ble_notice_callback(ble_notice_t notice, void *param) {
  const recv_adv_t *info = reinterpret_cast<const recv_adv_t *>(param);
  // rssi is a signed dBm carried in a uint8_t; cast through int8_t (standard for
  // a signed dBm value packed in a uint8_t).
-  s_ble->enqueue_scan_report(info->adv_addr, static_cast<int8_t>(info->rssi), info->adv_addr_type, info->data,
-                             info->data_len);
+  s_ble->enqueue_scan_report(info->adv_addr, static_cast<int8_t>(info->rssi), info->adv_addr_type,
+                             static_cast<uint8_t>(info->evt_type), info->data, info->data_len);
 }

-void BK72xxBLE::enqueue_scan_report(const uint8_t *mac, int8_t rssi, uint8_t addr_type, const uint8_t *data,
-                                    uint16_t data_len) {
+void BK72xxBLE::enqueue_scan_report(const uint8_t *mac, int8_t rssi, uint8_t addr_type, uint8_t evt_type,
+                                    const uint8_t *data, uint16_t data_len) {
  BLEScanReport *report = this->report_pool_.allocate();
  if (report == nullptr) {
    // Pool exhausted — the queue is full; count and drop.
    this->report_queue_.increment_dropped_count();
    return;
  }
-  memcpy(report->mac, mac, 6);
+  memcpy(report->mac, mac, MAC_ADDRESS_SIZE);
  report->rssi = rssi;
  report->addr_type = addr_type;
+  report->evt_type = evt_type;
  report->data_len =
      (data_len <= sizeof(report->data)) ? static_cast<uint8_t>(data_len) : static_cast<uint8_t>(sizeof(report->data));
  memcpy(report->data, data, report->data_len);
@@ -123,6 +137,9 @@ void BK72xxBLE::enqueue_scan_report(const uint8_t *mac, int8_t rssi, uint8_t add

 void BK72xxBLE::setup() {
  s_ble = this;
+  // The report pool grows lazily on purpose: the BDK notice callback runs in
+  // task context (malloc-safe, unlike rp2040's IRQ path), and typical traffic
+  // stays far below the pool cap, so not warming contains RAM.
  // Resolve the MAC early so get_mac_lsb_first() is valid for consumers before
  // the stack is up (it is re-read once ble_entry() has run).
  this->resolve_mac_();
@@ -173,14 +190,40 @@ void BK72xxBLE::enable() {
 }

 void BK72xxBLE::loop() {
+  // Keep reconciling toward the requested scan state (e.g. complete a stop
+  // that arrived while a controller operation was in flight), and re-check a
+  // settled scan at low frequency: a controller-side drop re-enters the
+  // bring-up, and the budget's FAILED feeds the tracker's recovery.
+  // Keep driving until settled: any PENDING, plus a terminal stop whose slot
+  // must still be freed. A FAILED scan request is the one combination not
+  // re-driven here — that belongs to the tracker's backoff.
+  const uint32_t pump_now = App.get_loop_component_start_time();
+  if (this->last_result_ == ScanOpResult::PENDING ||
+      (!this->scan_wanted_ && this->last_result_ == ScanOpResult::FAILED)) {
+    const uint32_t gate = (this->release_warned_ || this->last_result_ == ScanOpResult::FAILED)
+                              ? RECONCILE_REJECTED_RETRY_MS
+                              : RECONCILE_RETRY_MS;
+    if (pump_now - this->last_advance_ms_ >= gate)
+      this->advance_();
+  } else if (this->scan_wanted_ && this->last_result_ == ScanOpResult::SETTLED &&
+             pump_now - this->last_advance_ms_ >= SCAN_LIVENESS_CHECK_MS) {
+    // Re-check a settled scan; scan_start() refills the bring-up budget.
+    // WARN: the only report of a drop that recovers inside its budget.
+    if (this->scan_start(this->requested_.interval, this->requested_.window, this->requested_.active) !=
+        ScanOpResult::SETTLED)
+      ESP_LOGW(TAG, "Controller dropped the scan; restarting");
+  }
+
  // Drain the lock-free ring filled by the BLE task; all per-report work runs
  // here on the main task, then the report returns to the pool.
  BLEScanReport *report = this->report_queue_.pop();
  if (report == nullptr)
    return;
  do {
+#ifdef BK72XX_BLE_SCAN_LISTENER_COUNT
    for (auto *listener : this->scan_listeners_)
      listener->on_scan_report(*report);
+#endif
    this->report_pool_.release(report);
  } while ((report = this->report_queue_.pop()) != nullptr);

@@ -191,7 +234,7 @@ void BK72xxBLE::loop() {
    ESP_LOGW(TAG, "Dropped %u scan reports due to queue overflow", dropped);
 }

-void BK72xxBLE::get_mac_lsb_first(uint8_t out[6]) const {
+void BK72xxBLE::get_mac_lsb_first(uint8_t out[MAC_ADDRESS_SIZE]) const {
  for (int i = 0; i < 6; i++)
    out[i] = this->ble_mac_[i];
 }
@@ -224,7 +267,7 @@ void BK72xxBLE::resolve_mac_() {
    }
  }
  if (nonzero) {
-    memcpy(this->ble_mac_, common_default_bdaddr.addr, 6);
+    memcpy(this->ble_mac_, common_default_bdaddr.addr, MAC_ADDRESS_SIZE);
    return;
  }
 #endif
@@ -236,54 +279,238 @@ void BK72xxBLE::resolve_mac_() {
  // (verified against the BK7231N BLE-5.1 and BK7252N/BK7238 BLE-5.2 SDK sources), so it
  // matches on every device, including the last-byte == 0xFF edge that a 24-bit increment
  // would carry differently.
-  uint8_t wifi_mac[6];
+  uint8_t wifi_mac[MAC_ADDRESS_SIZE];
  get_mac_address_raw(wifi_mac);  // MSB-first
-  const uint8_t ble[6] = {wifi_mac[0], wifi_mac[1], wifi_mac[2],
-                          wifi_mac[3], wifi_mac[4], static_cast<uint8_t>(wifi_mac[5] + 1)};
+  const uint8_t ble[MAC_ADDRESS_SIZE] = {wifi_mac[0], wifi_mac[1], wifi_mac[2],
+                                         wifi_mac[3], wifi_mac[4], static_cast<uint8_t>(wifi_mac[5] + 1)};
  // Store LSB-first to match recv_adv_t adv_addr ordering.
  for (int i = 0; i < 6; i++)
    this->ble_mac_[i] = ble[5 - i];
 }

 // ---------------------------------------------------------------------------
-// Controller scan primitives
+// Scan reconciler
 // ---------------------------------------------------------------------------

-bool BK72xxBLE::scan_start(uint16_t interval, uint16_t window) {
+// Episode boundary: fresh teardown deadline and error bookkeeping.
+void BK72xxBLE::reset_teardown_episode_() {
+  this->teardown_since_ms_ = 0;
+  this->restarting_ = false;
+  this->last_release_err_ = 0;
+}
+
+ScanOpResult BK72xxBLE::scan_start(uint16_t interval, uint16_t window, bool active) {
  if (!this->is_active())
    this->enable();

-  if (this->scan_actv_idx_ != 0xFF) {
-    // Already scanning — stop first so this call cleanly restarts with the new
-    // parameters (the BDK cannot start a second scan on a busy activity).
-    this->scan_stop();
+  const ScanParams params{active, interval, window};
+  // A new episode refills the budget and gets a fresh teardown deadline; a
+  // re-call observing an in-flight bring-up (last result PENDING) must not.
+  if (this->last_result_ != ScanOpResult::PENDING || !this->scan_wanted_ || params != this->requested_) {
+    this->pending_since_ms_ = App.get_loop_component_start_time();
+    this->reset_teardown_episode_();
  }
+  this->scan_wanted_ = true;
+  this->requested_ = params;
+  return this->advance_();
+}

-  struct scan_param sp;
-  memset(&sp, 0, sizeof(sp));
-  sp.channel_map = 7;  // advertising channels 37/38/39
-  sp.interval = interval;
-  sp.window = window;
+void BK72xxBLE::scan_stop() {
+  if (this->scan_wanted_) {
+    // A stamp inherited from a stuck restart would fail the stop on its
+    // first advance.
+    this->reset_teardown_episode_();
+  }
+  this->scan_wanted_ = false;
+  this->advance_();
+}

-  this->scan_actv_idx_ = app_ble_get_idle_actv_idx_handle(SCAN_ACTV);
-  if (this->scan_actv_idx_ == 0xFF) {
-    ESP_LOGE(TAG, "Scan start failed: no idle activity handle");
+bool BK72xxBLE::flush_pending_stop(uint32_t timeout_ms) {
+  // millis() on both sides: the loop clock is frozen while this blocks.
+  const uint32_t start = millis();
+  while (!this->scan_wanted_ && this->last_result_ == ScanOpResult::PENDING) {
+    if (millis() - start >= timeout_ms)
+      return false;
+    delay(RECONCILE_RETRY_MS);
+    this->advance_();
+  }
+  return this->last_result_ == ScanOpResult::SETTLED;
+}
+
+// Teardown is asynchronous: the handle is kept until an IDLE observation
+// confirms the radio is idle. A rejection WARNs once per failure streak and
+// widens the pump gate; the epilogue owns the stuck-teardown deadline.
+void BK72xxBLE::release_activity_(BdkActivityState state) {
+  const BdkOpResult result =
+      bdk_scan_release(this->scan_activity_idx_, state == BdkActivityState::CREATED, &this->last_release_err_);
+  if (result == BdkOpResult::OK) {
+    this->release_warned_ = false;
+    return;
+  }
+  if (!this->release_warned_) {
+    // A hard error carries its code immediately; the 30 s stuck ERROR follows
+    // if it persists.
+    if (result == BdkOpResult::FAILED) {
+      ESP_LOGW(TAG, "Scan activity release failed (err %d); retrying", this->last_release_err_);
+    } else {
+      ESP_LOGW(TAG, "Scan activity release rejected; retrying");
+    }
+    this->release_warned_ = true;
+  }
+}
+
+// Stamp/track the teardown episode; once past the deadline, ERROR (re-logged
+// each interval) and report stuck.
+bool BK72xxBLE::teardown_stuck_(uint32_t now) {
+  if (this->teardown_since_ms_ == 0) {
+    this->teardown_since_ms_ = now;
+    this->teardown_stuck_log_ms_ = now;  // first ERROR fires at the deadline
    return false;
  }
-  ble_err_t ret = bk_ble_scan_start(this->scan_actv_idx_, &sp, nullptr);
-  if (ret != ERR_SUCCESS) {
-    ESP_LOGE(TAG, "Scan start failed (err %d)", static_cast<int>(ret));
-    this->scan_actv_idx_ = 0xFF;
+  if (now - this->teardown_since_ms_ < TEARDOWN_STUCK_ERROR_MS)
    return false;
+  if (now - this->teardown_stuck_log_ms_ >= TEARDOWN_STUCK_ERROR_MS) {
+    if (this->last_release_err_ != 0) {
+      ESP_LOGE(TAG, "Scan teardown cannot proceed; scanner is stuck (release err %d)", this->last_release_err_);
+    } else {
+      // No rejected release this episode: stuck waiting on the controller.
+      ESP_LOGE(TAG, "Scan teardown cannot proceed; scanner is stuck (controller busy)");
+    }
+    this->teardown_stuck_log_ms_ = now;
  }
  return true;
 }

-void BK72xxBLE::scan_stop() {
-  if (this->scan_actv_idx_ != 0xFF) {
-    bk_ble_scan_stop(this->scan_actv_idx_, nullptr);
-    this->scan_actv_idx_ = 0xFF;
+// One SDK operation per call toward the latched request; controller state is
+// read live each time (it changes on the BLE task, so nothing is mirrored).
+// The epilogue owns all deadlines and episode bookkeeping.
+ScanOpResult BK72xxBLE::advance_() {
+  if (!this->scan_wanted_ && this->scan_activity_idx_ == INVALID_ACTIVITY_IDX) {
+    // Nothing to do; also keeps SDK reads off the pre-enable() path.
+    this->last_result_ = ScanOpResult::SETTLED;
+    return ScanOpResult::SETTLED;
  }
+  const BdkActivityState state = bdk_scan_state(this->scan_activity_idx_);
+  const bool ready = bdk_scan_ready();
+  ScanOpResult result = this->scan_wanted_ ? this->advance_start_(state, ready) : this->advance_stop_(state, ready);
+
+  const uint32_t now = App.get_loop_component_start_time();
+  this->last_advance_ms_ = now;
+  if (result == ScanOpResult::SETTLED || (state == BdkActivityState::IDLE && ready)) {
+    // Any teardown episode is over (IDLE observed with the controller
+    // settled, or e.g. a mode flip that settled back without ever reaching
+    // IDLE). An IDLE read while an operation is in flight proves nothing —
+    // a stop deferred there must keep its episode running.
+    this->reset_teardown_episode_();
+    this->release_warned_ = false;
+  }
+  if (this->restarting_ && (state == BdkActivityState::IDLE || state == BdkActivityState::CREATED)) {
+    // The mode-change release is observed complete; the rest is a normal
+    // bring-up on a fresh budget.
+    this->restarting_ = false;
+    this->pending_since_ms_ = now;
+  }
+  // Not chained to the clear above: a bring-up waiting at IDLE (create still
+  // in flight) must keep spending its budget.
+  if (result == ScanOpResult::PENDING) {
+    if (this->scan_wanted_ && state != BdkActivityState::STARTED && !this->restarting_) {
+      // A downed radio spends the bring-up budget; exhausting it hands
+      // recovery to the tracker's backoff.
+      if (now - this->pending_since_ms_ >= RECONCILE_PENDING_TIMEOUT_MS) {
+        ESP_LOGE(TAG, "Scan bring-up did not settle; giving up until the next start");
+        result = ScanOpResult::FAILED;
+      }
+    } else {
+      // A teardown is pending: a stop, or a mode-change release still in
+      // flight (restarting_); either way the bring-up budget waits.
+      if (this->scan_wanted_)
+        this->pending_since_ms_ = now;
+      if (this->teardown_stuck_(now)) {
+        // Terminal for stop AND restart: the tracker's backoff owns recovery
+        // (a stop's release keeps re-driving from loop(); a restart is
+        // re-requested through scan_start() with a fresh deadline).
+        result = ScanOpResult::FAILED;
+      }
+    }
+  }
+  this->last_result_ = result;
+  return result;
+}
+
+ScanOpResult BK72xxBLE::advance_stop_(BdkActivityState state, bool ready) {
+  if (state == BdkActivityState::IDLE && ready) {
+    // Fully torn down (or never created): the radio is idle. IDLE is trusted
+    // only when the controller is settled — mid-create the slot still reads
+    // IDLE, and dropping the handle then would leak the activity once the
+    // create lands.
+    this->scan_activity_idx_ = INVALID_ACTIVITY_IDX;
+    return ScanOpResult::SETTLED;
+  }
+  if (!ready) {
+    // Acting mid-operation could delete an activity whose start lands
+    // afterwards, leaking the slot with the radio on; wait.
+    if (this->last_result_ == ScanOpResult::SETTLED)
+      ESP_LOGD(TAG, "Scan stop deferred (controller busy)");
+    return ScanOpResult::PENDING;
+  }
+  // Settled, so CREATED unambiguously means "never started".
+  this->release_activity_(state);
+  return ScanOpResult::PENDING;  // confirmed once IDLE is observed
+}
+
+ScanOpResult BK72xxBLE::advance_start_(BdkActivityState state, bool ready) {
+  if (state == BdkActivityState::STARTED) {
+    if (this->applied_ == this->requested_)
+      return ScanOpResult::SETTLED;
+    // Running with different mode or parameters: tear down (the SDK stop
+    // chain also deletes the activity) and recreate on a later advance.
+    if (ready) {
+      this->release_activity_(state);
+      // Invalidate so a flip back to the old params cannot SETTLE against the
+      // activity being deleted (interval 0 never matches a real request).
+      this->applied_.interval = 0;
+      this->restarting_ = true;
+    }
+    return ScanOpResult::PENDING;
+  }
+  if (!ready) {
+    if (this->last_result_ == ScanOpResult::SETTLED)
+      ESP_LOGD(TAG, "Scan start deferred (controller busy)");
+    return ScanOpResult::PENDING;
+  }
+  if (state == BdkActivityState::CREATED) {
+    // Fire-and-forget: SETTLED only once a later advance observes the scan
+    // running, so a rejected start is retried rather than silently dead. On
+    // failure the created activity is intact; keep the handle.
+    if (bdk_scan_start(this->scan_activity_idx_, this->requested_.interval, this->requested_.window,
+                       this->requested_.active) != BdkOpResult::OK)
+      return ScanOpResult::FAILED;
+    this->applied_ = this->requested_;
+    return ScanOpResult::PENDING;
+  }
+  if (state == BdkActivityState::OTHER)
+    return ScanOpResult::PENDING;  // transitional; settles on a later read
+
+  // IDLE and ready: acquire a slot and create. A kept index is deliberately
+  // reused: SDK delete returns the slot to idle and create requires an idle
+  // slot, so it equals a fresh acquire — while clearing here would orphan a
+  // create still in flight (the BUSY race below).
+  if (this->scan_activity_idx_ == INVALID_ACTIVITY_IDX) {
+    this->scan_activity_idx_ = bdk_scan_acquire_activity();
+    if (this->scan_activity_idx_ == INVALID_ACTIVITY_IDX)
+      return ScanOpResult::FAILED;
+  }
+  switch (bdk_scan_create(this->scan_activity_idx_)) {
+    case BdkOpResult::BUSY:  // raced the BLE task; keep the index, the retry resumes this slot
+    case BdkOpResult::OK:
+      return ScanOpResult::PENDING;
+    case BdkOpResult::FAILED:
+      break;
+  }
+  // Safe to clear (unlike BUSY): acquire is a pure search, so a rejected
+  // create leaves the slot IDLE for re-acquire.
+  this->scan_activity_idx_ = INVALID_ACTIVITY_IDX;
+  return ScanOpResult::FAILED;
 }

 }  // namespace esphome::bk72xx_ble
@@ -6,10 +6,12 @@

 #include "esphome/core/component.h"
 #include "esphome/core/event_pool.h"
+#include "esphome/core/helpers.h"
 #include "esphome/core/lock_free_queue.h"

 #include <cstdint>
-#include <vector>
+
+#include "bdk_scan.h"

 namespace esphome::bk72xx_ble {

@@ -19,11 +21,32 @@ enum class BLEComponentState : uint8_t {
  ACTIVE,
 };

+/// Outcome of one reconciliation step.
+enum class ScanOpResult : uint8_t {
+  SETTLED,  ///< The request is reached: scan observed running, or stopped
+            ///< with the activity fully released.
+  PENDING,  ///< A step is in flight; loop() keeps advancing — call
+            ///< scan_start() again to learn the outcome.
+  FAILED,   ///< The controller rejected a step; retry later.
+};
+
+/// One scan request: mode plus timing, in BLE units (0.625 ms).
+struct ScanParams {
+  bool active;
+  uint16_t interval;
+  uint16_t window;
+  bool operator==(const ScanParams &) const = default;
+};
+
 /// One advertisement report from the controller.
 struct BLEScanReport {
-  uint8_t mac[6];  // LSB-first, as the controller delivers it
-  int8_t rssi;     // signed dBm
+  uint8_t mac[MAC_ADDRESS_SIZE];  // LSB-first, as the controller delivers it
+  int8_t rssi;                    // signed dBm
  uint8_t addr_type;
+  // GAPM report info byte (recv_adv_t.evt_type): bits 0-2 report type
+  // (1 = legacy adv, 3 = legacy scan response), bit 5 scannable — lets the
+  // tracker's merger tell the two frames apart.
+  uint8_t evt_type;
  uint8_t data_len;  // bytes valid in data[]
  uint8_t data[62];  // legacy advertisement (31) + scan response (31)

@@ -60,25 +83,48 @@ class BK72xxBLE final : public Component {
  void set_enable_on_boot(bool enable_on_boot) { this->enable_on_boot_ = enable_on_boot; }

  /// Controller BLE address, least-significant octet first (BLE convention).
-  void get_mac_lsb_first(uint8_t out[6]) const;
+  void get_mac_lsb_first(uint8_t out[MAC_ADDRESS_SIZE]) const;

+#ifdef BK72XX_BLE_SCAN_LISTENER_COUNT
  /// Register a consumer for scan reports (delivered on the main task via loop()).
+  /// Storage is codegen-sized: the consumer's codegen requests a slot via
+  /// request_scan_listener_slot(), which emits BK72XX_BLE_SCAN_LISTENER_COUNT.
  void register_scan_listener(BLEScanListener *listener) { this->scan_listeners_.push_back(listener); }
+#endif

-  /// Start the controller scan. Interval/window are in BLE units (0.625 ms).
-  /// Enables the stack first if needed. Returns false on controller failure.
-  bool scan_start(uint16_t interval, uint16_t window);
-  /// Stop the controller scan (no-op when not scanning).
+  /// Request a scan (interval/window in 0.625 ms BLE units); enables the
+  /// stack first if needed. PENDING until the scan is observed running —
+  /// loop() keeps advancing, call again to learn the outcome.
+  ScanOpResult scan_start(uint16_t interval, uint16_t window, bool active);
+  /// Request the scanner stopped and the activity released; steps that
+  /// cannot run yet are completed from loop().
  void scan_stop();
+  /// Drive a requested stop until the radio is observed idle, bounded by
+  /// timeout_ms (for OTA). Returns false if it still has not settled.
+  bool flush_pending_stop(uint32_t timeout_ms);
+  /// Last reconciliation outcome; on FAILED the consumer's retry policy owns
+  /// recovery.
+  ScanOpResult last_scan_result() const { return this->last_result_; }

  /// Internal: buffer one controller report (BDK notice callback, BLE task
  /// context — bounded copy under the scheduler lock, nothing else).
-  void enqueue_scan_report(const uint8_t *mac, int8_t rssi, uint8_t addr_type, const uint8_t *data, uint16_t data_len);
+  void enqueue_scan_report(const uint8_t *mac, int8_t rssi, uint8_t addr_type, uint8_t evt_type, const uint8_t *data,
+                           uint16_t data_len);

 protected:
  void resolve_mac_();
+  ScanOpResult advance_();
+  ScanOpResult advance_stop_(BdkActivityState state, bool ready);
+  ScanOpResult advance_start_(BdkActivityState state, bool ready);
+  bool teardown_stuck_(uint32_t now);
+  void reset_teardown_episode_();
+  void release_activity_(BdkActivityState state);

-  std::vector<BLEScanListener *> scan_listeners_;
+#ifdef BK72XX_BLE_SCAN_LISTENER_COUNT
+  // Codegen-sized: no heap allocation, no std::vector template instantiation —
+  // the same StaticVector pattern as the tracker's ble_device_base listeners.
+  StaticVector<BLEScanListener *, BK72XX_BLE_SCAN_LISTENER_COUNT> scan_listeners_;
+#endif
  // Report ring: the BDK notice callback (BLE task) allocates a report from the
  // pool, fills it and pushes the pointer; loop() pops, dispatches and releases.
  // Lock-free SPSC, zero allocation at steady state — the esp32_ble pattern.
@@ -87,10 +133,24 @@ class BK72xxBLE final : public Component {
  // allocate() returns nullptr before push() can fail. This prevents leaking a
  // pool slot on a failed push and keeps release() off the producer path.
  esphome::EventPool<BLEScanReport, MAX_SCAN_REPORT_QUEUE_SIZE - 1> report_pool_;
-  uint8_t ble_mac_[6]{0};  // LSB-first (BLE convention)
-  uint8_t scan_actv_idx_{0xFF};
-  BLEComponentState state_{BLEComponentState::STATE_OFF};
+  // Largest-to-smallest: padding only at the tail, absorbed by future byte fields.
+  uint32_t last_advance_ms_{0};
+  uint32_t pending_since_ms_{0};          // bring-up budget anchor; refilled on request change
+  uint32_t teardown_since_ms_{0};         // unfinished teardown episode start; 0 = none
+  uint32_t teardown_stuck_log_ms_{0};     // last stuck-teardown ERROR; re-logged each TEARDOWN_STUCK_ERROR_MS
+  int last_release_err_{0};               // SDK code of the episode's last failed release; 0 = none
+  ScanParams requested_{};                // latched by scan_start()
+  ScanParams applied_{};                  // last params we commanded; mismatch with requested_ restarts
+  uint8_t ble_mac_[MAC_ADDRESS_SIZE]{0};  // LSB-first (BLE convention)
+  uint8_t scan_activity_idx_{INVALID_ACTIVITY_IDX};
+  bool scan_wanted_{false};     // the latched request is to scan (vs stopped)
+  bool release_warned_{false};  // gates the release WARN; widens the pump gate
+  bool restarting_{false};      // mode-change release in flight; teardown deadline governs until released
  bool enable_on_boot_{false};
+  // PENDING means advance_() has more to do; loop() drives it, paced and
+  // (for a bring-up) bounded.
+  ScanOpResult last_result_{ScanOpResult::SETTLED};
+  BLEComponentState state_{BLEComponentState::STATE_OFF};
 };

 }  // namespace esphome::bk72xx_ble
@@ -12,19 +12,31 @@ Scan modes:
                      Use this when the radio is dedicated to BLE.
  continuous: false — a started scan runs for `duration` ms, then stops. The
                      FIRST start is external too: nothing in this component
-                      starts a non-continuous scan on boot, so until the
-                      automation actions land (follow-up PR) the radio stays
-                      idle. start_scan() is called from code (e.g. an api
-                      client-connected automation) so the single-core radio
-                      can service WiFi in between scans.
+                      starts a non-continuous scan on boot — the radio stays
+                      idle until bk72xx_ble_tracker.start_scan fires (e.g.
+                      from an api client-connected automation), so the
+                      single-core radio can service WiFi in between scans.
 """

+from esphome import automation
 import esphome.codegen as cg
 from esphome.components import bk72xx_ble, ble_device_base, ota
-from esphome.components.const import CONF_SCAN_PARAMETERS, CONF_WINDOW
+from esphome.components.ble_device_base import automation as ble_automation
+from esphome.components.const import CONF_ON_SCAN_END, CONF_SCAN_PARAMETERS, CONF_WINDOW
 import esphome.config_validation as cv
-from esphome.const import CONF_CONTINUOUS, CONF_DURATION, CONF_ID, CONF_INTERVAL
-from esphome.core import CORE, CoroPriority, coroutine_with_priority
+from esphome.const import (
+    CONF_ACTIVE,
+    CONF_CONTINUOUS,
+    CONF_DURATION,
+    CONF_ID,
+    CONF_INTERVAL,
+    CONF_MANUFACTURER_ID,
+    CONF_ON_BLE_ADVERTISE,
+    CONF_ON_BLE_MANUFACTURER_DATA_ADVERTISE,
+    CONF_ON_BLE_SERVICE_DATA_ADVERTISE,
+    CONF_SERVICE_UUID,
+)
+from esphome.core import ID
 from esphome.types import ConfigType

 CONF_BK72XX_BLE_ID = "bk72xx_ble_id"
@@ -33,119 +45,146 @@ DEPENDENCIES = ["bk72xx"]
 AUTO_LOAD = ["ble_device_base", "bk72xx_ble"]
 CODEOWNERS = ["@Bl00d-B0b"]

+ble_device_base.register_hub_provider("bk72xx_ble_tracker")
+
 bk72xx_ble_tracker_ns = cg.esphome_ns.namespace("bk72xx_ble_tracker")
 BK72xxBLETracker = bk72xx_ble_tracker_ns.class_(
    "BK72xxBLETracker", ble_device_base.BLEHub, cg.Component
 )

+StartScanAction = bk72xx_ble_tracker_ns.class_("StartScanAction", automation.Action)
+StopScanAction = bk72xx_ble_tracker_ns.class_("StopScanAction", automation.Action)

-def to_ble_units(value: cv.TimePeriod) -> int:
-    """Convert a scan time to the controller's 0.625 ms units.
-
-    Used by both validation and codegen so what is validated is exactly what is
-    programmed — the truncation here is what makes the duty-cycle check below
-    meaningful.
-    """
-    return value.total_microseconds // 625
+ESPBTAdvertiseTrigger = ble_automation.ESPBTAdvertiseTrigger
+BLEServiceDataAdvertiseTrigger = ble_automation.BLEServiceDataAdvertiseTrigger
+BLEManufacturerDataAdvertiseTrigger = ble_automation.BLEManufacturerDataAdvertiseTrigger
+BLEEndOfScanTrigger = ble_automation.BLEEndOfScanTrigger


-def validate_scan_parameters(config: ConfigType) -> ConfigType:
-    """Reject impossible window/interval/duration combinations at config time.
-
-    Mirrors esp32_ble_tracker: the controller cannot scan for longer than the
-    interval, and a too-short duration would end the scan period almost
-    immediately. Catching it here gives a clear error instead of a runtime
-    controller failure and the 1/sec retry loop.
-    """
-    duration = config[CONF_DURATION]
-    interval = config[CONF_INTERVAL]
-    window = config[CONF_WINDOW]
-
-    if window > interval:
-        raise cv.Invalid(
-            f"Scan window ({window}) needs to be smaller than scan interval ({interval})"
-        )
-
-    # BLE scan interval/window are programmed in 0.625 ms units as a 16-bit value; the
-    # controller only accepts 2.5 ms .. 10240 ms (0x0004 .. 0x4000). Reject out-of-range
-    # values here instead of letting the unit conversion silently overflow.
-    for name, value in (("interval", interval), ("window", window)):
-        if value.total_microseconds < 2500 or value.total_microseconds > 10_240_000:
-            raise cv.Invalid(
-                f"Scan {name} ({value}) must be between 2.5 ms and 10240 ms"
-            )
-
-    # Validate what actually reaches the controller: both values are truncated to
-    # whole 0.625 ms units, so a window/interval pair that differs by less than one
-    # unit collapses to the same value — silently programming a 100 % duty cycle
-    # (radio permanently on) from a config that asked for less.
-    interval_units = to_ble_units(interval)
-    window_units = to_ble_units(window)
-    if window_units == interval_units and window < interval:
-        raise cv.Invalid(
-            f"Scan window ({window}) and interval ({interval}) both round to "
-            f"{interval_units} x 0.625 ms, which the controller scans at a 100 % duty "
-            f"cycle. Separate them by at least 0.625 ms."
-        )
-
-    if interval.total_microseconds * 3 > duration.total_microseconds:
-        raise cv.Invalid(
-            f"Scan duration ({duration}) must cover at least three scan intervals "
-            f"({interval}): the scanner listens on one of the three BLE advertising "
-            f"channels per interval, so a shorter duration can miss devices entirely."
-        )
-
-    return config
-
-
-SCAN_PARAMETERS_SCHEMA = cv.All(
-    cv.Schema(
-        {
-            cv.Optional(CONF_DURATION, default="5min"): cv.positive_time_period_seconds,
-            # interval/window default to the BK reference scan rate — 100 ms / 30 ms,
-            # a 30 % duty cycle. Converted to the controller's 0.625 ms BLE units in
-            # to_code(). (LN882H's SDK recommends a different 100 / 50 ms = 50 %.)
-            cv.Optional(CONF_INTERVAL, default="100ms"): cv.positive_time_period,
-            cv.Optional(CONF_WINDOW, default="30ms"): cv.positive_time_period,
-            cv.Optional(CONF_CONTINUOUS, default=True): cv.boolean,
-        }
-    ),
-    validate_scan_parameters,
-)
+# interval defaults to the BK reference scan rate — 100 ms with the shared 30 ms
+# window, a 30 % duty cycle. Converted to the controller's 0.625 ms BLE units in
+# to_code(). (LN882H's SDK recommends a different 100 / 50 ms = 50 %.)
+SCAN_PARAMETERS_SCHEMA = ble_device_base.scan_parameters_schema("100ms")

 CONFIG_SCHEMA = cv.Schema(
    {
        cv.GenerateID(): cv.declare_id(BK72xxBLETracker),
        cv.GenerateID(CONF_BK72XX_BLE_ID): cv.use_id(bk72xx_ble.BK72xxBLE),
        cv.Optional(CONF_SCAN_PARAMETERS, default={}): SCAN_PARAMETERS_SCHEMA,
+        cv.Optional(CONF_ON_BLE_ADVERTISE): ble_automation.advertise_trigger_schema(
+            ESPBTAdvertiseTrigger
+        ),
+        cv.Optional(
+            CONF_ON_BLE_SERVICE_DATA_ADVERTISE
+        ): ble_automation.uuid_trigger_schema(
+            BLEServiceDataAdvertiseTrigger,
+            {cv.Required(CONF_SERVICE_UUID): ble_device_base.bt_uuid},
+        ),
+        cv.Optional(
+            CONF_ON_BLE_MANUFACTURER_DATA_ADVERTISE
+        ): ble_automation.uuid_trigger_schema(
+            BLEManufacturerDataAdvertiseTrigger,
+            {cv.Required(CONF_MANUFACTURER_ID): ble_device_base.bt_uuid},
+        ),
+        cv.Optional(CONF_ON_SCAN_END): ble_automation.scan_end_trigger_schema(
+            BLEEndOfScanTrigger
+        ),
    }
 ).extend(cv.COMPONENT_SCHEMA)


-# Runs at FINAL priority so every BLE sensor has registered through
-# ble_device_base (and any tracker-owned listeners have been counted) before
-# the StaticVector size is emitted. Same pattern as esp32_ble_tracker.
-@coroutine_with_priority(CoroPriority.FINAL)
-async def _emit_listener_count() -> None:
-    count = ble_device_base.get_listener_count()
-    if count > 0:
-        cg.add_define("ESPHOME_BLE_DEVICE_BASE_LISTENER_COUNT", count)
+@automation.register_action(
+    "bk72xx_ble_tracker.start_scan",
+    StartScanAction,
+    cv.Schema(
+        {
+            cv.GenerateID(): cv.use_id(BK72xxBLETracker),
+            # Optional with no default, unlike esp32_ble_tracker: omitting it
+            # keeps whatever scan_parameters.continuous configured, instead of
+            # silently forcing one-shot.
+            cv.Optional(CONF_CONTINUOUS): cv.templatable(cv.boolean),
+        }
+    ),
+    synchronous=True,
+)
+async def start_scan_action_to_code(
+    config: ConfigType,
+    action_id: ID,
+    template_arg: cg.TemplateArguments,
+    args: list,
+) -> cg.MockObj:
+    var = cg.new_Pvariable(action_id, template_arg)
+    await cg.register_parented(var, config[CONF_ID])
+    if (continuous := config.get(CONF_CONTINUOUS)) is not None:
+        template_ = await cg.templatable(continuous, args, cg.bool_)
+        cg.add(var.set_continuous(template_))
+    return var
+
+
+@automation.register_action(
+    "bk72xx_ble_tracker.stop_scan",
+    StopScanAction,
+    automation.maybe_simple_id(
+        cv.Schema(
+            {
+                cv.GenerateID(): cv.use_id(BK72xxBLETracker),
+            }
+        )
+    ),
+    synchronous=True,
+)
+async def stop_scan_action_to_code(
+    config: ConfigType,
+    action_id: ID,
+    template_arg: cg.TemplateArguments,
+    args: list,
+) -> cg.MockObj:
+    var = cg.new_Pvariable(action_id, template_arg)
+    await cg.register_parented(var, config[CONF_ID])
+    return var


 async def to_code(config: ConfigType) -> None:
+    # Selects the BLEHub alias arm in ble_device_base/ble_hub_impl.h.
+    cg.add_define("USE_BK72XX_BLE_TRACKER")
+    # Compiles the shared adv + scan-response merge (the BDK delivers the pair
+    # as separate reports).
+    cg.add_define("USE_BLE_SCAN_RESPONSE_MERGER")
+
    var = cg.new_Pvariable(config[CONF_ID])
    await cg.register_component(var, config)

    parent = await cg.get_variable(config[CONF_BK72XX_BLE_ID])
    cg.add(var.set_parent(parent))
+    # The tracker registers itself as a controller scan listener in setup();
+    # request the codegen-sized StaticVector slot for it.
+    bk72xx_ble.request_scan_listener_slot()

    # Get notified when an OTA update starts, to pause scanning (esp32_ble_tracker parity)
    ota.request_ota_state_listeners()

    scan = config[CONF_SCAN_PARAMETERS]
-    cg.add(var.set_scan_interval(to_ble_units(scan[CONF_INTERVAL])))
-    cg.add(var.set_scan_window(to_ble_units(scan[CONF_WINDOW])))
+    cg.add(var.set_scan_interval(ble_device_base.to_ble_units(scan[CONF_INTERVAL])))
+    cg.add(var.set_scan_window(ble_device_base.to_ble_units(scan[CONF_WINDOW])))
    cg.add(var.set_scan_duration(scan[CONF_DURATION].total_milliseconds))
-    cg.add(var.set_scan_continuous(scan[CONF_CONTINUOUS]))
+    cg.add(var.set_configured_continuous(scan[CONF_CONTINUOUS]))
+    cg.add(var.set_scan_active(scan[CONF_ACTIVE]))

-    CORE.add_job(_emit_listener_count)
+    for conf in config.get(CONF_ON_BLE_ADVERTISE, []):
+        await ble_automation.advertise_trigger_to_code(conf, var)
+
+    for trigger_key, uuid_key, setter_prefix in (
+        (CONF_ON_BLE_SERVICE_DATA_ADVERTISE, CONF_SERVICE_UUID, "set_service_uuid"),
+        (
+            CONF_ON_BLE_MANUFACTURER_DATA_ADVERTISE,
+            CONF_MANUFACTURER_ID,
+            "set_manufacturer_uuid",
+        ),
+    ):
+        for conf in config.get(trigger_key, []):
+            await ble_automation.uuid_trigger_to_code(
+                conf, var, uuid_key, setter_prefix
+            )
+
+    for conf in config.get(CONF_ON_SCAN_END, []):
+        await ble_automation.scan_end_trigger_to_code(conf, var)
--- a/Show More
+++ b/Show More