diff --git a/homeassistant/components/auth/indieauth.py b/homeassistant/components/auth/indieauth.py
index 8e5e9812da7dca..e16e9eef8afec5 100644
--- a/homeassistant/components/auth/indieauth.py
+++ b/homeassistant/components/auth/indieauth.py
@@ -1,7 +1,9 @@
 """Helpers to resolve client ID/secret."""
 
 from html.parser import HTMLParser
+from http import HTTPStatus
 from ipaddress import ip_address
+import json
 import logging
 from typing import override
 from urllib.parse import ParseResult, urljoin, urlparse
@@ -14,6 +16,9 @@
 
 _LOGGER = logging.getLogger(__name__)
 
+# We limit reads of a client_id page to the first 10kB.
+MAX_FETCH_BYTES = 10240
+
 
 async def verify_redirect_uri(
     hass: HomeAssistant, client_id: str, redirect_uri: str
@@ -24,7 +29,10 @@ async def verify_redirect_uri(
     except ValueError:
         return False
 
-    redirect_parts = _parse_url(redirect_uri)
+    try:
+        redirect_parts = _parse_url(redirect_uri)
+    except ValueError:
+        return False
 
     # Verify redirect url and client url have same scheme and domain.
     is_valid = (
@@ -53,7 +61,15 @@ async def verify_redirect_uri(
     # IndieAuth 4.2.2 allows for redirect_uri to be on different domain
     # but needs to be specified in link tag when fetching `client_id`.
     redirect_uris = await fetch_redirect_uris(hass, client_id)
-    return redirect_uri in redirect_uris
+    if redirect_uri in redirect_uris:
+        return True
+    _LOGGER.debug(
+        "redirect_uri %s is not among the advertised redirect uris %s for client_id %s",
+        redirect_uri,
+        redirect_uris,
+        client_id,
+    )
+    return False
 
 
 class LinkTagParser(HTMLParser):
@@ -63,7 +79,7 @@ def __init__(self, rel: str) -> None:
         """Initialize a link tag parser."""
         super().__init__()
         self.rel = rel
-        self.found: list[str | None] = []
+        self.found: list[str] = []
 
     @override
     def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
@@ -73,48 +89,115 @@ def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None
 
         attributes: dict[str, str | None] = dict(attrs)
 
-        if attributes.get("rel") == self.rel:
-            self.found.append(attributes.get("href"))
+        # Skip tags with a missing or empty href: urljoin resolves those to
+        # the client_id URL itself instead of naming a redirect target.
+        if attributes.get("rel") == self.rel and (href := attributes.get("href")):
+            self.found.append(href)
+
+
+def _reject_json_constant(constant: str) -> None:
+    """Reject NaN/Infinity/-Infinity, which RFC 8259 does not allow."""
+    raise ValueError(f"Invalid JSON constant: {constant}")
+
+
+def _is_valid_metadata_client_id(url: str) -> bool:
+    """Validate a client_id URL for the metadata-document fallback.
+
+    The client identifier URL must be https with a path component and no
+    fragment (a bare trailing # counts as a fragment component). The remaining
+    client identifier rules are enforced upstream by _parse_client_id.
+    """
+    try:
+        parts = urlparse(url)
+        # urlparse defers port validation until the attribute is accessed.
+        _ = parts.port
+    except ValueError:
+        return False
+    return parts.scheme == "https" and bool(parts.path) and "#" not in url
+
+
+def _is_valid_metadata_redirect_uri(redirect_uri: str) -> bool:
+    """Validate a client ID metadata document redirect_uris entry.
+
+    Entries must be absolute, fragment-free URIs: a non-empty scheme (so
+    private-use schemes like app:/callback stay valid) and no fragment per
+    RFC 6749 3.1.2 (a bare trailing # counts as a fragment component).
+    """
+    try:
+        parts = urlparse(redirect_uri)
+        # urlparse defers port validation until the attribute is accessed.
+        _ = parts.port
+    except ValueError:
+        return False
+    return bool(parts.scheme) and "#" not in redirect_uri
 
 
 async def fetch_redirect_uris(hass: HomeAssistant, url: str) -> list[str]:
-    """Find link tag with redirect_uri values.
+    """Find the redirect_uri values that a client_id advertises.
+
+    We support two formats, checked in this order:
 
     IndieAuth 4.2.2
 
     The client SHOULD publish one or more <link> tags or Link HTTP headers with
     a rel attribute of redirect_uri at the client_id URL.
 
-    We limit to the first 10kB of the page.
+    OAuth Client ID Metadata Document
+    (draft-ietf-oauth-client-id-metadata-document)
+
+    The client_id URL returns a JSON document with a redirect_uris array. As we
+    advertise client_id_metadata_document_supported in the authorization server
+    metadata, we fall back to this format when no link tags are found.
+
+    We read roughly the first 10kB of the page and a fetch error yields no
+    redirect uris.
 
     We do not implement extracting redirect uris from headers.
     """
-    parser = LinkTagParser("redirect_uri")
-    chunks = 0
+    body: bytes = b""
+    status: int | None = None
+    redirected = False
     try:
         async with (
             aiohttp.ClientSession() as session,
             session.get(url, timeout=aiohttp.ClientTimeout(total=5)) as resp,
         ):
+            status = resp.status
+            redirected = bool(resp.history)
             async for data in resp.content.iter_chunked(1024):
-                parser.feed(data.decode())
-                chunks += 1
+                body += data
 
-                if chunks == 10:
+                if len(body) >= MAX_FETCH_BYTES:
                     break
 
     except TimeoutError:
         _LOGGER.error("Timeout while looking up redirect_uri %s", url)
+        return []
     except aiohttp.client_exceptions.ClientSSLError:
         _LOGGER.error("SSL error while looking up redirect_uri %s", url)
+        return []
     except aiohttp.client_exceptions.ClientOSError as ex:
         _LOGGER.error("OS error while looking up redirect_uri %s: %s", url, ex.strerror)
+        return []
     except aiohttp.client_exceptions.ClientConnectionError:
         _LOGGER.error(
             "Low level connection error while looking up redirect_uri %s", url
         )
+        return []
     except aiohttp.client_exceptions.ClientError:
         _LOGGER.error("Unknown error while looking up redirect_uri %s", url)
+        return []
+
+    if redirect_uris := _parse_link_tag_redirect_uris(url, body):
+        return redirect_uris
+
+    return _parse_metadata_document_redirect_uris(url, body, status, redirected)
+
+
+def _parse_link_tag_redirect_uris(url: str, body: bytes) -> list[str]:
+    """Find <link rel="redirect_uri"> values in the client_id page body."""
+    parser = LinkTagParser("redirect_uri")
+    parser.feed(body.decode(errors="replace"))
 
     # Authorization endpoints verifying that a redirect_uri is allowed for use
     # by a client MUST look for an exact match of the given redirect_uri in the
@@ -123,6 +206,77 @@ async def fetch_redirect_uris(hass: HomeAssistant, url: str) -> list[str]:
     return [urljoin(url, found) for found in parser.found]
 
 
+def _parse_metadata_document_redirect_uris(
+    url: str, body: bytes, status: int | None, redirected: bool
+) -> list[str]:
+    """Parse the client_id page body as an OAuth Client ID Metadata Document.
+
+    Per draft-ietf-oauth-client-id-metadata-document the document only counts
+    when the client_id URL is https with a path and no fragment, the response
+    was a direct 200 (not redirected), the document's client_id round-trips,
+    and every redirect_uris entry is an absolute, fragment-free URI matched
+    exactly. The url and its document are client-controlled and fetched
+    unauthenticated, so rejections log at DEBUG (higher levels would be a
+    log-flood vector).
+    """
+    # A body at the read cap may be truncated; a truncated prefix must not be
+    # trusted even if it happens to be parseable.
+    if (
+        len(body) >= MAX_FETCH_BYTES
+        or status != HTTPStatus.OK
+        or redirected
+        or not _is_valid_metadata_client_id(url)
+    ):
+        _LOGGER.debug(
+            "Not treating %s as a client ID metadata document: body length %s,"
+            " status %s, redirected %s (client_id must be a fragment-free https"
+            " URL with a path)",
+            url,
+            len(body),
+            status,
+            redirected,
+        )
+        return []
+
+    try:
+        # Strict decode (RFC 8259 requires UTF-8): the link tag parser's
+        # lenient replacement decode would mask invalid bytes as U+FFFD.
+        document = json.loads(body.decode(), parse_constant=_reject_json_constant)
+    except UnicodeDecodeError:
+        _LOGGER.debug("Client ID metadata document at %s is not valid UTF-8", url)
+        return []
+    except ValueError:
+        _LOGGER.debug("Client ID metadata document at %s is not valid JSON", url)
+        return []
+
+    if not isinstance(document, dict):
+        _LOGGER.debug("Client ID metadata document at %s is not a JSON object", url)
+        return []
+
+    if document.get("client_id") != url:
+        _LOGGER.debug(
+            "Client ID metadata document at %s client_id does not match the"
+            " document URL",
+            url,
+        )
+        return []
+
+    # redirect_uris entries are returned unmodified for RFC 6749 exact matching
+    # rather than resolving relative references.
+    redirect_uris = document.get("redirect_uris")
+    if not isinstance(redirect_uris, list) or not all(
+        isinstance(redirect_uri, str) and _is_valid_metadata_redirect_uri(redirect_uri)
+        for redirect_uri in redirect_uris
+    ):
+        _LOGGER.debug(
+            "Client ID metadata document at %s has missing or invalid redirect_uris",
+            url,
+        )
+        return []
+
+    return redirect_uris
+
+
 def verify_client_id(client_id: str) -> bool:
     """Verify that the client id is valid."""
     try:
diff --git a/homeassistant/components/auth/login_flow.py b/homeassistant/components/auth/login_flow.py
index 6594acd9eec45c..ead83de8d029bf 100644
--- a/homeassistant/components/auth/login_flow.py
+++ b/homeassistant/components/auth/login_flow.py
@@ -137,12 +137,11 @@ async def get(self, request: web.Request) -> web.Response:
             "authorization_endpoint": f"{url_prefix}/auth/authorize",
             "token_endpoint": f"{url_prefix}/auth/token",
             "revocation_endpoint": f"{url_prefix}/auth/revoke",
-            # Home Assistant already accepts URL-based client_ids via
-            # IndieAuth without prior registration, which is compatible with
-            # draft-ietf-oauth-client-id-metadata-document. This flag
-            # advertises that support to encourage clients to use it. The
-            # metadata document is not actually fetched as IndieAuth doesn't
-            # require it.
+            # Home Assistant accepts URL-based client_ids via IndieAuth without
+            # prior registration, and discovers allowed redirect URIs from link
+            # tags or a Client ID Metadata Document served at the client_id URL.
+            # This flag advertises that support
+            # (draft-ietf-oauth-client-id-metadata-document).
             "client_id_metadata_document_supported": True,
             "response_types_supported": ["code"],
             "service_documentation": (
diff --git a/homeassistant/components/bluetooth/manifest.json b/homeassistant/components/bluetooth/manifest.json
index 6b02775717a661..4c85a25b9d2711 100644
--- a/homeassistant/components/bluetooth/manifest.json
+++ b/homeassistant/components/bluetooth/manifest.json
@@ -19,8 +19,8 @@
     "bleak-retry-connector==4.6.3",
     "bluetooth-adapters==2.4.0",
     "bluetooth-auto-recovery==1.6.4",
-    "bluetooth-data-tools==1.29.18",
+    "bluetooth-data-tools==1.29.21",
     "dbus-fast==5.0.22",
-    "habluetooth==6.26.5"
+    "habluetooth==6.26.7"
   ]
 }
diff --git a/homeassistant/components/esphome/entry_data.py b/homeassistant/components/esphome/entry_data.py
index 00a56ca6dc058d..97cdc0353dfb8a 100644
--- a/homeassistant/components/esphome/entry_data.py
+++ b/homeassistant/components/esphome/entry_data.py
@@ -522,7 +522,9 @@ def async_on_disconnect(self) -> None:
         """
         self.available = False
         if self.bluetooth_device:
-            self.bluetooth_device.available = False
+            # Fails pending BLE slot waiters and clears the dead
+            # session's allocations in addition to closing the gate.
+            self.bluetooth_device.async_set_unavailable()
         # Make a copy since calling the disconnect callbacks
         # may also try to discard/remove themselves.
         for disconnect_cb in self.disconnect_callbacks.copy():
diff --git a/homeassistant/components/esphome/manifest.json b/homeassistant/components/esphome/manifest.json
index f428fc5b9d65b8..5a509cff34d402 100644
--- a/homeassistant/components/esphome/manifest.json
+++ b/homeassistant/components/esphome/manifest.json
@@ -19,7 +19,7 @@
   "requirements": [
     "aioesphomeapi==45.12.0",
     "esphome-dashboard-api==1.4.0",
-    "bleak-esphome==3.9.7"
+    "bleak-esphome==4.0.0"
   ],
   "zeroconf": ["_esphomelib._tcp.local."]
 }
diff --git a/homeassistant/components/ld2410_ble/manifest.json b/homeassistant/components/ld2410_ble/manifest.json
index 806d8edecb32fd..995bdaa09669e4 100644
--- a/homeassistant/components/ld2410_ble/manifest.json
+++ b/homeassistant/components/ld2410_ble/manifest.json
@@ -20,5 +20,5 @@
   "documentation": "https://www.home-assistant.io/integrations/ld2410_ble",
   "integration_type": "device",
   "iot_class": "local_push",
-  "requirements": ["bluetooth-data-tools==1.29.18", "ld2410-ble==0.1.1"]
+  "requirements": ["bluetooth-data-tools==1.29.21", "ld2410-ble==0.1.1"]
 }
diff --git a/homeassistant/components/led_ble/manifest.json b/homeassistant/components/led_ble/manifest.json
index 6489e7711e3f09..1f03d9099426f4 100644
--- a/homeassistant/components/led_ble/manifest.json
+++ b/homeassistant/components/led_ble/manifest.json
@@ -36,5 +36,5 @@
   "documentation": "https://www.home-assistant.io/integrations/led_ble",
   "integration_type": "device",
   "iot_class": "local_polling",
-  "requirements": ["bluetooth-data-tools==1.29.18", "led-ble==1.1.11"]
+  "requirements": ["bluetooth-data-tools==1.29.21", "led-ble==1.1.11"]
 }
diff --git a/homeassistant/components/private_ble_device/manifest.json b/homeassistant/components/private_ble_device/manifest.json
index 386dcb0ac9b78e..46e45a2cc242bf 100644
--- a/homeassistant/components/private_ble_device/manifest.json
+++ b/homeassistant/components/private_ble_device/manifest.json
@@ -7,5 +7,5 @@
   "documentation": "https://www.home-assistant.io/integrations/private_ble_device",
   "integration_type": "device",
   "iot_class": "local_push",
-  "requirements": ["bluetooth-data-tools==1.29.18"]
+  "requirements": ["bluetooth-data-tools==1.29.21"]
 }
diff --git a/homeassistant/components/template/entity.py b/homeassistant/components/template/entity.py
index e2c2709b84eef0..85d19f013b1f2d 100644
--- a/homeassistant/components/template/entity.py
+++ b/homeassistant/components/template/entity.py
@@ -88,12 +88,13 @@ def __init__(
             )
 
         device_registry = dr.async_get(hass)
+        # Allow linking to a main or child device, but not to a composite device.
         if (
-            device_id := config.get(CONF_DEVICE_ID)
-        ) is not None and device_registry.async_is_composite_device_id(
-            device_id
-        ) is False:
-            self.device_entry = device_registry.async_get(device_id)
+            (device_id := config.get(CONF_DEVICE_ID)) is not None
+            and (device_entry := device_registry.async_get(device_id)) is not None
+            and not device_registry.async_is_composite_device_id(device_id)
+        ):
+            self.device_entry = device_entry
 
     @property
     @abstractmethod
diff --git a/homeassistant/components/template/repairs.py b/homeassistant/components/template/repairs.py
index 3a95eb11a349f9..c7e308eee879b7 100644
--- a/homeassistant/components/template/repairs.py
+++ b/homeassistant/components/template/repairs.py
@@ -41,9 +41,9 @@ async def async_step_select_device(
         errors: dict[str, str] = {}
         if user_input is not None:
             device_id = user_input.get(CONF_DEVICE_ID)
-            if (
-                device_id is None
-                or device_registry.async_is_composite_device_id(device_id) is False
+            if device_id is None or (
+                device_registry.async_get(device_id) is not None
+                and not device_registry.async_is_composite_device_id(device_id)
             ):
                 options = {**entry.options}
                 if device_id:
diff --git a/homeassistant/components/tplink/entity.py b/homeassistant/components/tplink/entity.py
index 12abb7913c0a92..02724d4ceeb0a3 100644
--- a/homeassistant/components/tplink/entity.py
+++ b/homeassistant/components/tplink/entity.py
@@ -279,7 +279,7 @@ def _async_call_update_attrs(self) -> None:
             if self._attr_available:
                 _LOGGER.warning(
                     "Unable to read data for %s %s: %s",
-                    self._device,
+                    self._device.host,
                     self.entity_id,
                     ex,
                 )
diff --git a/homeassistant/components/unifi/manifest.json b/homeassistant/components/unifi/manifest.json
index 0b4facb368cb04..eda3c5beb90439 100644
--- a/homeassistant/components/unifi/manifest.json
+++ b/homeassistant/components/unifi/manifest.json
@@ -9,5 +9,5 @@
   "iot_class": "local_push",
   "loggers": ["aiounifi"],
   "quality_scale": "silver",
-  "requirements": ["aiounifi==92"]
+  "requirements": ["aiounifi==93"]
 }
diff --git a/homeassistant/components/zwave_js/config_flow.py b/homeassistant/components/zwave_js/config_flow.py
index 457be46b9a151d..203cfe7d4c4f0f 100644
--- a/homeassistant/components/zwave_js/config_flow.py
+++ b/homeassistant/components/zwave_js/config_flow.py
@@ -791,6 +791,19 @@ async def async_step_on_supervisor(
             self.lr_s2_authenticated_key = addon_config.get(
                 CONF_ADDON_LR_S2_AUTHENTICATED_KEY, ""
             )
+
+            if self._adapter_discovered:
+                # Apply the discovered adapter to the add-on config and
+                # restart the add-on before connecting, so the server
+                # version info reflects the discovered adapter.
+                self._addon_config_updates.update(
+                    {
+                        CONF_ADDON_DEVICE: self.usb_path,
+                        CONF_ADDON_SOCKET: self.socket_path,
+                    }
+                )
+                return await self.async_step_start_addon()
+
             return await self.async_step_finish_addon_setup_user()
 
         if addon_info.state is AddonState.NOT_RUNNING:
@@ -1030,24 +1043,6 @@ async def async_step_finish_addon_setup_user(
             # with add-on data.
             return self.async_abort(reason="already_configured")
 
-        # When we came from discovery, make sure we update the add-on
-        if self._adapter_discovered and self.use_addon:
-            await self._async_set_addon_config(
-                {
-                    CONF_ADDON_DEVICE: self.usb_path,
-                    CONF_ADDON_SOCKET: self.socket_path,
-                    CONF_ADDON_S0_LEGACY_KEY: self.s0_legacy_key,
-                    CONF_ADDON_S2_ACCESS_CONTROL_KEY: self.s2_access_control_key,
-                    CONF_ADDON_S2_AUTHENTICATED_KEY: self.s2_authenticated_key,
-                    CONF_ADDON_S2_UNAUTHENTICATED_KEY: self.s2_unauthenticated_key,
-                    CONF_ADDON_LR_S2_ACCESS_CONTROL_KEY: self.lr_s2_access_control_key,
-                    CONF_ADDON_LR_S2_AUTHENTICATED_KEY: self.lr_s2_authenticated_key,
-                }
-            )
-            if self.restart_addon:
-                manager = get_addon_manager(self.hass)
-                await manager.async_stop_addon()
-
         self._abort_if_unique_id_configured(
             updates={
                 CONF_URL: self.ws_address,
@@ -1709,6 +1704,24 @@ async def async_step_esphome(
         }
         self._adapter_discovered = True
 
+        # A discovered adapter that doesn't belong to an existing add-on based
+        # entry is a different adapter, so offer to migrate the existing
+        # network to it instead of repointing the shared add-on config.
+        discovered_home_id = (
+            str(discovery_info.zwave_home_id) if discovery_info.zwave_home_id else None
+        )
+        if addon_entry := next(
+            (
+                entry
+                for entry in self._async_current_entries(include_ignore=False)
+                if entry.data.get(CONF_USE_ADDON)
+                and entry.unique_id != discovered_home_id
+            ),
+            None,
+        ):
+            self._reconfigure_config_entry = addon_entry
+            return await self.async_step_confirm_usb_migration()
+
         return await self.async_step_installation_type()
 
     async def async_revert_addon_config(self, reason: str) -> ConfigFlowResult:
diff --git a/homeassistant/package_constraints.txt b/homeassistant/package_constraints.txt
index 5fa44636f0055a..d015c15709a7e5 100644
--- a/homeassistant/package_constraints.txt
+++ b/homeassistant/package_constraints.txt
@@ -24,7 +24,7 @@ bleak-retry-connector==4.6.3
 bleak==3.0.2
 bluetooth-adapters==2.4.0
 bluetooth-auto-recovery==1.6.4
-bluetooth-data-tools==1.29.18
+bluetooth-data-tools==1.29.21
 cached-ipaddress==1.1.2
 certifi>=2021.5.30
 ciso8601==2.3.3
@@ -35,7 +35,7 @@ file-read-backwards==2.0.0
 fnv-hash-fast==2.0.3
 go2rtc-client==0.4.0
 ha-ffmpeg==3.2.2
-habluetooth==6.26.5
+habluetooth==6.26.7
 hass-nabucasa==2.2.0
 hassil==3.11.0
 home-assistant-bluetooth==2.0.0
diff --git a/requirements_all.txt b/requirements_all.txt
index a0494898fd8e47..a9c9eef3e76492 100644
--- a/requirements_all.txt
+++ b/requirements_all.txt
@@ -453,7 +453,7 @@ aiotedee==0.3.0
 aiotractive==1.0.3
 
 # homeassistant.components.unifi
-aiounifi==92
+aiounifi==93
 
 # homeassistant.components.usb
 aiousbwatcher==1.1.2
@@ -657,7 +657,7 @@ beautifulsoup4==4.13.3
 bizkaibus==0.1.1
 
 # homeassistant.components.esphome
-bleak-esphome==3.9.7
+bleak-esphome==4.0.0
 
 # homeassistant.components.bluetooth
 bleak-retry-connector==4.6.3
@@ -693,7 +693,7 @@ bluetooth-auto-recovery==1.6.4
 # homeassistant.components.ld2410_ble
 # homeassistant.components.led_ble
 # homeassistant.components.private_ble_device
-bluetooth-data-tools==1.29.18
+bluetooth-data-tools==1.29.21
 
 # homeassistant.components.bond
 bond-async==0.2.1
@@ -1228,7 +1228,7 @@ ha-xthings-cloud==1.0.5
 habiticalib==0.4.7
 
 # homeassistant.components.bluetooth
-habluetooth==6.26.5
+habluetooth==6.26.7
 
 # homeassistant.components.hanna
 hanna-cloud==0.0.7
diff --git a/tests/components/auth/test_indieauth.py b/tests/components/auth/test_indieauth.py
index 2a8d6894dc63c1..653a6b60f3fd44 100644
--- a/tests/components/auth/test_indieauth.py
+++ b/tests/components/auth/test_indieauth.py
@@ -1,8 +1,10 @@
 """Tests for the client validator."""
 
 import asyncio
+import json
 from unittest.mock import patch
 
+import aiohttp
 import pytest
 
 from homeassistant.components.auth import indieauth
@@ -167,6 +169,440 @@ async def test_find_link_tag_max_size(hass: HomeAssistant, mock_session) -> None
     assert redirect_uris == ["http://127.0.0.1:8000/wine"]
 
 
+async def test_find_link_tag_without_href(
+    hass: HomeAssistant, mock_session: AiohttpClientMocker
+) -> None:
+    """Test a redirect_uri link tag without a usable href is skipped."""
+    mock_session.get(
+        "http://127.0.0.1:8000",
+        text="""
+<!doctype html>
+<html>
+  <head>
+    <link rel="redirect_uri">
+    <link rel="redirect_uri" href="">
+    <link rel="redirect_uri" href="https://example.com/cb">
+  </head>
+</html>
+""",
+    )
+    redirect_uris = await indieauth.fetch_redirect_uris(hass, "http://127.0.0.1:8000")
+
+    assert redirect_uris == ["https://example.com/cb"]
+
+
+async def test_fetch_redirect_uris_metadata_document(
+    hass: HomeAssistant, mock_session: AiohttpClientMocker
+) -> None:
+    """Test fetching redirect uris from a client id metadata document."""
+    mock_session.get(
+        "https://example.com/client",
+        text=json.dumps(
+            {
+                "client_id": "https://example.com/client",
+                "redirect_uris": [
+                    "https://example.com/callback",
+                    "https://other.com/callback",
+                ],
+            }
+        ),
+        headers={"Content-Type": "application/json"},
+    )
+    redirect_uris = await indieauth.fetch_redirect_uris(
+        hass, "https://example.com/client"
+    )
+
+    assert redirect_uris == [
+        "https://example.com/callback",
+        "https://other.com/callback",
+    ]
+
+
+async def test_fetch_redirect_uris_metadata_document_text_plain(
+    hass: HomeAssistant, mock_session: AiohttpClientMocker
+) -> None:
+    """Test the metadata document is parsed regardless of content type."""
+    mock_session.get(
+        "https://example.com/client",
+        text=json.dumps(
+            {
+                "client_id": "https://example.com/client",
+                "redirect_uris": ["https://example.com/callback"],
+            }
+        ),
+        headers={"Content-Type": "text/plain"},
+    )
+    redirect_uris = await indieauth.fetch_redirect_uris(
+        hass, "https://example.com/client"
+    )
+
+    assert redirect_uris == ["https://example.com/callback"]
+
+
+async def test_fetch_redirect_uris_link_tag_precedence(
+    hass: HomeAssistant, mock_session: AiohttpClientMocker
+) -> None:
+    """Test link tags take precedence over metadata document parsing."""
+    mock_session.get(
+        "http://127.0.0.1:8000",
+        text="""
+<!doctype html>
+<html>
+  <head>
+    <link rel="redirect_uri" href="hass://oauth2_redirect">
+  </head>
+  <body>
+    {"redirect_uris": ["https://example.com/should-be-ignored"]}
+  </body>
+</html>
+""",
+    )
+    redirect_uris = await indieauth.fetch_redirect_uris(hass, "http://127.0.0.1:8000")
+
+    assert redirect_uris == ["hass://oauth2_redirect"]
+
+
+@pytest.mark.parametrize(
+    "text",
+    [
+        pytest.param("this is neither json nor html", id="not-json-not-html"),
+        pytest.param('["https://example.com/callback"]', id="json-array"),
+        pytest.param("42", id="json-scalar"),
+        pytest.param(
+            json.dumps({"redirect_uris": ["https://example.com/callback"]}),
+            id="missing-client-id",
+        ),
+        pytest.param(
+            json.dumps({"client_id": "https://example.com/client"}),
+            id="missing-redirect-uris",
+        ),
+        pytest.param(
+            json.dumps(
+                {
+                    "client_id": "https://example.com/client",
+                    "redirect_uris": [],
+                }
+            ),
+            id="empty-redirect-uris",
+        ),
+        pytest.param(
+            json.dumps(
+                {
+                    "client_id": "https://other.example/client",
+                    "redirect_uris": ["https://example.com/callback"],
+                }
+            ),
+            id="client-id-mismatch",
+        ),
+        pytest.param(
+            json.dumps(
+                {
+                    "client_id": "https://example.com/client",
+                    "redirect_uris": "https://example.com/callback",
+                }
+            ),
+            id="redirect-uris-not-list",
+        ),
+        pytest.param(
+            json.dumps(
+                {
+                    "client_id": "https://example.com/client",
+                    "redirect_uris": ["https://example.com/callback", 123],
+                }
+            ),
+            id="redirect-uris-non-string-entry",
+        ),
+        pytest.param(
+            json.dumps(
+                {
+                    "client_id": "https://example.com/client",
+                    "redirect_uris": ["/callback"],
+                }
+            ),
+            id="redirect-uris-relative-entry",
+        ),
+        pytest.param(
+            json.dumps(
+                {
+                    "client_id": "https://example.com/client",
+                    "redirect_uris": ["https://example.com/callback#fragment"],
+                }
+            ),
+            id="redirect-uris-fragment-entry",
+        ),
+        pytest.param(
+            json.dumps(
+                {
+                    "client_id": "https://example.com/client",
+                    "redirect_uris": ["https://["],
+                }
+            ),
+            id="redirect-uris-unparsable-entry",
+        ),
+        pytest.param(
+            json.dumps(
+                {
+                    "client_id": "https://example.com/client",
+                    "redirect_uris": ["https://example.com/callback#"],
+                }
+            ),
+            id="redirect-uris-empty-fragment-entry",
+        ),
+        pytest.param(
+            json.dumps(
+                {
+                    "client_id": "https://example.com/client",
+                    "redirect_uris": ["https://example.com:not-a-port/callback"],
+                }
+            ),
+            id="redirect-uris-invalid-port-entry",
+        ),
+        pytest.param(
+            '{"client_id": "https://example.com/client",'
+            ' "redirect_uris": ["https://example.com/callback"], "x": NaN}',
+            id="json-nan-constant",
+        ),
+    ],
+)
+async def test_fetch_redirect_uris_metadata_document_invalid(
+    hass: HomeAssistant, mock_session: AiohttpClientMocker, text: str
+) -> None:
+    """Test that invalid metadata documents yield no redirect uris."""
+    mock_session.get(
+        "https://example.com/client",
+        text=text,
+        headers={"Content-Type": "application/json"},
+    )
+
+    assert await indieauth.fetch_redirect_uris(hass, "https://example.com/client") == []
+    assert not await indieauth.verify_redirect_uri(
+        hass, "https://example.com/client", "https://other.com/callback"
+    )
+
+
+async def test_verify_redirect_uri_metadata_document(
+    hass: HomeAssistant, mock_session: AiohttpClientMocker
+) -> None:
+    """Test verifying a cross-origin redirect uri from a metadata document."""
+    client_id = "https://example.com/client"
+    mock_session.get(
+        client_id,
+        text=json.dumps(
+            {
+                "client_id": client_id,
+                "redirect_uris": ["https://other.com/callback"],
+            }
+        ),
+        headers={"Content-Type": "application/json"},
+    )
+
+    assert await indieauth.verify_redirect_uri(
+        hass, client_id, "https://other.com/callback"
+    )
+
+    assert not await indieauth.verify_redirect_uri(
+        hass, client_id, "https://other.com/not-listed"
+    )
+
+
+async def test_verify_redirect_uri_unparsable(hass: HomeAssistant) -> None:
+    """Test an unparsable requested redirect uri is rejected without raising."""
+    assert not await indieauth.verify_redirect_uri(
+        hass, "https://example.com/client", "https://["
+    )
+
+
+async def test_fetch_redirect_uris_metadata_document_invalid_utf8(
+    hass: HomeAssistant, mock_session: AiohttpClientMocker
+) -> None:
+    """Test a metadata document with invalid UTF-8 is rejected."""
+    mock_session.get(
+        "https://example.com/client",
+        content=(
+            b'{"client_id": "https://example.com/client",'
+            b' "redirect_uris": ["https://other.com/callback"], "note": "\xff"}'
+        ),
+        headers={"Content-Type": "application/json"},
+    )
+
+    assert await indieauth.fetch_redirect_uris(hass, "https://example.com/client") == []
+
+
+@pytest.mark.parametrize(
+    "client_id",
+    [
+        pytest.param("https://example.com", id="no-path"),
+        pytest.param("https://example.com/client#", id="empty-fragment"),
+    ],
+)
+async def test_fetch_redirect_uris_metadata_document_invalid_client_id(
+    hass: HomeAssistant, mock_session: AiohttpClientMocker, client_id: str
+) -> None:
+    """Test client ids violating the metadata document URL rules are ignored."""
+    mock_session.get(
+        client_id,
+        text=json.dumps(
+            {
+                "client_id": client_id,
+                "redirect_uris": ["https://other.com/callback"],
+            }
+        ),
+        headers={"Content-Type": "application/json"},
+    )
+
+    assert await indieauth.fetch_redirect_uris(hass, client_id) == []
+
+
+async def test_fetch_redirect_uris_metadata_document_not_ok(
+    hass: HomeAssistant, mock_session: AiohttpClientMocker
+) -> None:
+    """Test a metadata document not served with 200 OK is ignored."""
+    mock_session.get(
+        "https://example.com/client",
+        text=json.dumps(
+            {
+                "client_id": "https://example.com/client",
+                "redirect_uris": ["https://example.com/callback"],
+            }
+        ),
+        status=404,
+        headers={"Content-Type": "application/json"},
+    )
+
+    assert await indieauth.fetch_redirect_uris(hass, "https://example.com/client") == []
+
+
+async def test_fetch_redirect_uris_metadata_document_http_scheme(
+    hass: HomeAssistant, mock_session: AiohttpClientMocker
+) -> None:
+    """Test a metadata document served over http is ignored."""
+    client_id = "http://example.com/client"
+    mock_session.get(
+        client_id,
+        text=json.dumps(
+            {
+                "client_id": client_id,
+                "redirect_uris": ["https://other.com/callback"],
+            }
+        ),
+        headers={"Content-Type": "application/json"},
+    )
+
+    assert await indieauth.fetch_redirect_uris(hass, client_id) == []
+    assert not await indieauth.verify_redirect_uri(
+        hass, client_id, "https://other.com/callback"
+    )
+
+
+async def test_fetch_redirect_uris_metadata_document_redirected(
+    hass: HomeAssistant, mock_session: AiohttpClientMocker
+) -> None:
+    """Test a metadata document reached via a redirect is ignored."""
+    mock_session.get(
+        "https://example.com/client",
+        text=json.dumps(
+            {
+                "client_id": "https://example.com/client",
+                "redirect_uris": ["https://example.com/callback"],
+            }
+        ),
+        headers={"Content-Type": "application/json"},
+        history=(object(),),
+    )
+
+    assert await indieauth.fetch_redirect_uris(hass, "https://example.com/client") == []
+
+
+async def test_fetch_redirect_uris_metadata_document_private_use_scheme(
+    hass: HomeAssistant, mock_session: AiohttpClientMocker
+) -> None:
+    """Test a private-use scheme redirect uri is accepted as an absolute URI."""
+    mock_session.get(
+        "https://example.com/client",
+        text=json.dumps(
+            {
+                "client_id": "https://example.com/client",
+                "redirect_uris": ["app:/oauth-callback"],
+            }
+        ),
+        headers={"Content-Type": "application/json"},
+    )
+
+    assert await indieauth.fetch_redirect_uris(hass, "https://example.com/client") == [
+        "app:/oauth-callback"
+    ]
+
+
+async def test_fetch_redirect_uris_metadata_document_oversized(
+    hass: HomeAssistant, mock_session: AiohttpClientMocker
+) -> None:
+    """Test a document past the 10kB cap is rejected as an incomplete read."""
+    mock_session.get(
+        "https://example.com/client",
+        text=json.dumps(
+            {
+                "client_id": "https://example.com/client",
+                "redirect_uris": ["https://example.com/callback"],
+                "padding": "x" * 11000,
+            }
+        ),
+        headers={"Content-Type": "application/json"},
+    )
+
+    assert await indieauth.fetch_redirect_uris(hass, "https://example.com/client") == []
+
+
+async def test_fetch_redirect_uris_metadata_document_exactly_at_cap(
+    hass: HomeAssistant, mock_session: AiohttpClientMocker
+) -> None:
+    """Test a document of exactly the read cap is rejected as possibly truncated."""
+    document = {
+        "client_id": "https://example.com/client",
+        "redirect_uris": ["https://other.com/callback"],
+        "padding": "",
+    }
+    document["padding"] = "x" * (10240 - len(json.dumps(document)))
+    text = json.dumps(document)
+    assert len(text) == 10240
+
+    mock_session.get(
+        "https://example.com/client",
+        text=text,
+        headers={"Content-Type": "application/json"},
+    )
+
+    assert await indieauth.fetch_redirect_uris(hass, "https://example.com/client") == []
+
+
+async def test_fetch_redirect_uris_metadata_document_at_cap_ineligible(
+    hass: HomeAssistant, mock_session: AiohttpClientMocker
+) -> None:
+    """Test a valid document that reaches the 10kB cap is ineligible."""
+    mock_session.get(
+        "https://example.com/client",
+        text=json.dumps(
+            {
+                "client_id": "https://example.com/client",
+                "redirect_uris": [
+                    f"https://example.com/callback/{index}" for index in range(400)
+                ],
+            }
+        ),
+        headers={"Content-Type": "application/json"},
+    )
+
+    assert await indieauth.fetch_redirect_uris(hass, "https://example.com/client") == []
+
+
+async def test_fetch_redirect_uris_network_error(
+    hass: HomeAssistant, mock_session: AiohttpClientMocker
+) -> None:
+    """Test a network error yields no redirect uris without raising."""
+    mock_session.get("https://example.com/client", exc=aiohttp.ClientError())
+
+    assert await indieauth.fetch_redirect_uris(hass, "https://example.com/client") == []
+
+
 @pytest.mark.parametrize(
     "client_id",
     ["https://home-assistant.io/android", "https://home-assistant.io/iOS"],
diff --git a/tests/components/bluetooth/test_diagnostics.py b/tests/components/bluetooth/test_diagnostics.py
index a0748faf9d37f3..a76ee7ee3283d9 100644
--- a/tests/components/bluetooth/test_diagnostics.py
+++ b/tests/components/bluetooth/test_diagnostics.py
@@ -565,6 +565,14 @@ def inject_advertisement(
                         "slots": 5,
                         "source": "00:00:00:00:00:01",
                     },
+                    # Registering a connectable scanner seeds a zeroed
+                    # entry (slots=0 means no slot info reported yet).
+                    "esp32": {
+                        "allocated": [],
+                        "free": 0,
+                        "slots": 0,
+                        "source": "esp32",
+                    },
                 },
                 "adapters": {
                     "hci0": {
diff --git a/tests/components/esphome/test_bluetooth.py b/tests/components/esphome/test_bluetooth.py
index ded9a776a8ada7..9e8947049c1505 100644
--- a/tests/components/esphome/test_bluetooth.py
+++ b/tests/components/esphome/test_bluetooth.py
@@ -1,5 +1,6 @@
 """Test the ESPHome bluetooth integration."""
 
+import asyncio
 from collections.abc import Callable
 from typing import Any
 from unittest.mock import MagicMock, patch
@@ -10,6 +11,7 @@
     BluetoothScannerState,
     BluetoothScannerStateResponse,
 )
+import pytest
 
 from homeassistant.components import bluetooth
 from homeassistant.components.bluetooth import BluetoothScanningMode
@@ -343,3 +345,22 @@ def _spy_register(*args: Any, **kwargs: Any) -> Callable[[], None]:
     # habluetooth auto-mode worker is spawned at registration time.
     set_mode_mock.assert_called_once_with(BluetoothScannerMode.PASSIVE)
     assert requested_at_register == [BluetoothScanningMode.AUTO]
+
+
+async def test_bluetooth_disconnect_fails_parked_slot_waiter(
+    hass: HomeAssistant, mock_bluetooth_entry_with_raw_adv: MockESPHomeDevice
+) -> None:
+    """Test a parked BLE slot waiter fails fast when the entry disconnects."""
+    entry_data = mock_bluetooth_entry_with_raw_adv.entry.runtime_data
+    bluetooth_device = entry_data.bluetooth_device
+    assert bluetooth_device is not None
+    task = hass.async_create_task(bluetooth_device.wait_for_ble_connections_free(60.0))
+    await asyncio.sleep(0)
+    assert not task.done()
+
+    await mock_bluetooth_entry_with_raw_adv.mock_disconnect(True)
+    await hass.async_block_till_done()
+
+    with pytest.raises(TimeoutError, match="Proxy became unavailable"):
+        await task
+    assert bluetooth_device.available is False
diff --git a/tests/components/template/test_init.py b/tests/components/template/test_init.py
index af518f5a7b3021..8727784f08275b 100644
--- a/tests/components/template/test_init.py
+++ b/tests/components/template/test_init.py
@@ -492,6 +492,56 @@ def check_template_entities(
     )
 
 
+@pytest.mark.parametrize(
+    "linked_device",
+    [
+        pytest.param("main", id="main_device"),
+        pytest.param("child", id="child_device"),
+    ],
+)
+async def test_link_to_main_or_child_device(
+    hass: HomeAssistant,
+    device_registry: dr.DeviceRegistry,
+    entity_registry: er.EntityRegistry,
+    linked_device: str,
+) -> None:
+    """Test a template entity links to a selected main or child device."""
+    source_entry = MockConfigEntry()
+    source_entry.add_to_hass(hass)
+    main_device = device_registry.async_get_or_create(
+        config_entry_id=source_entry.entry_id,
+        identifiers={("test", "main")},
+    )
+    child_device = device_registry.async_get_or_create_child(
+        config_entry_id=source_entry.entry_id,
+        identifiers={("test", "child")},
+        parent_device_id=main_device.id,
+    )
+    selected_device_id = {"main": main_device, "child": child_device}[linked_device].id
+
+    template_config_entry = MockConfigEntry(
+        domain=DOMAIN,
+        options={
+            "name": "My template",
+            "state": "{{10}}",
+            "template_type": "sensor",
+            "device_id": selected_device_id,
+        },
+        title="Template",
+    )
+    template_config_entry.add_to_hass(hass)
+    assert await hass.config_entries.async_setup(template_config_entry.entry_id)
+    await hass.async_block_till_done()
+
+    template_entities = list(
+        entity_registry.entities.get_entries_for_config_entry_id(
+            template_config_entry.entry_id
+        )
+    )
+    assert len(template_entities) == 1
+    assert template_entities[0].device_id == selected_device_id
+
+
 async def test_setup_removes_stale_helper_device(
     hass: HomeAssistant,
     device_registry: dr.DeviceRegistry,
diff --git a/tests/components/template/test_repairs.py b/tests/components/template/test_repairs.py
index 5e5f004abcd63c..1be7befffd1e95 100644
--- a/tests/components/template/test_repairs.py
+++ b/tests/components/template/test_repairs.py
@@ -176,6 +176,53 @@ async def test_composite_device_id_repair_flow(
     assert entity_entry.device_id == picked_device_id
 
 
+@pytest.mark.usefixtures("split_devices")
+async def test_composite_device_id_repair_flow_links_child_device(
+    hass: HomeAssistant,
+    hass_client: ClientSessionGenerator,
+    device_registry: dr.DeviceRegistry,
+    entity_registry: er.EntityRegistry,
+    issue_registry: ir.IssueRegistry,
+) -> None:
+    """Test the repair flow accepts a child device and links the entity to it."""
+    source_entry = MockConfigEntry(domain="itg3")
+    source_entry.add_to_hass(hass)
+    parent_device = device_registry.async_get_or_create(
+        config_entry_id=source_entry.entry_id,
+        identifiers={("itg3", "parent")},
+    )
+    child_device = device_registry.async_get_or_create_child(
+        config_entry_id=source_entry.entry_id,
+        identifiers={("itg3", "child")},
+        parent_device_id=parent_device.id,
+    )
+
+    entry = await _setup_template_entry(hass, COMPOSITE_ID)
+    issue_id = f"composite_device_id_{entry.entry_id}"
+    assert issue_registry.async_get_issue(DOMAIN, issue_id)
+
+    assert await async_setup_component(hass, "repairs", {})
+    await hass.async_block_till_done()
+    client = await hass_client()
+
+    result = await start_repair_fix_flow(client, DOMAIN, issue_id)
+    assert result["type"] == FlowResultType.FORM
+    assert result["step_id"] == "select_device"
+
+    result = await process_repair_fix_flow(
+        client, result["flow_id"], json={CONF_DEVICE_ID: child_device.id}
+    )
+    assert result["type"] == FlowResultType.CREATE_ENTRY
+    await hass.async_block_till_done()
+
+    assert entry.options[CONF_DEVICE_ID] == child_device.id
+    assert not issue_registry.async_get_issue(DOMAIN, issue_id)
+
+    entity_entry = entity_registry.async_get(TEMPLATE_ENTITY_ID)
+    assert entity_entry is not None
+    assert entity_entry.device_id == child_device.id
+
+
 async def test_composite_device_id_repair_flow_ambiguity_not_resolved(
     hass: HomeAssistant,
     hass_client: ClientSessionGenerator,
diff --git a/tests/components/tplink/test_init.py b/tests/components/tplink/test_init.py
index 0d757a3fc310f9..3cb8f8bf8064bd 100644
--- a/tests/components/tplink/test_init.py
+++ b/tests/components/tplink/test_init.py
@@ -375,7 +375,8 @@ async def test_update_attrs_fails_in_init(
     assert entity
     state = hass.states.get(entity_id)
     assert state.state == STATE_UNAVAILABLE
-    assert f"Unable to read data for MockLight {entity_id}:" in caplog.text
+    assert f"Unable to read data for {IP_ADDRESS} {entity_id}:" in caplog.text
+    assert "MockLight" not in caplog.text
 
 
 async def test_update_attrs_fails_on_update(
@@ -418,7 +419,8 @@ async def test_update_attrs_fails_on_update(
     assert entity
     state = hass.states.get(entity_id)
     assert state.state == STATE_UNAVAILABLE
-    assert f"Unable to read data for MockLight {entity_id}:" in caplog.text
+    assert f"Unable to read data for {IP_ADDRESS} {entity_id}:" in caplog.text
+    assert "MockLight" not in caplog.text
     # Check only logs once
     caplog.clear()
     freezer.tick(5)
@@ -427,7 +429,7 @@ async def test_update_attrs_fails_on_update(
     assert entity
     state = hass.states.get(entity_id)
     assert state.state == STATE_UNAVAILABLE
-    assert f"Unable to read data for MockLight {entity_id}:" not in caplog.text
+    assert f"Unable to read data for {IP_ADDRESS} {entity_id}:" not in caplog.text
 
 
 async def test_feature_no_category(
diff --git a/tests/components/unifi/test_switch.py b/tests/components/unifi/test_switch.py
index e1c980a0e21096..4b223fde41f067 100644
--- a/tests/components/unifi/test_switch.py
+++ b/tests/components/unifi/test_switch.py
@@ -735,6 +735,34 @@
     "x_has_ssh_hostkey": True,
 }
 
+UPS_DEVICE_1 = deepcopy(PDU_DEVICE_1)
+UPS_DEVICE_1.update(
+    {
+        "device_id": "mock-ups",
+        "mac": "02:00:00:00:00:01",
+        "model": "USPDA2B",
+        "name": "Dummy UPS 2U Pro",
+        "type": "usp",
+        "outlet_table": [
+            {
+                "index": 1,
+                "relay_state": True,
+                "cycle_enabled": False,
+                "name": "Outlet 1",
+                "outlet_caps": 65539,
+            }
+        ],
+        "outlet_overrides": [
+            {
+                "cycle_enabled": False,
+                "name": "Outlet 1",
+                "relay_state": True,
+                "index": 1,
+            }
+        ],
+    }
+)
+
 WLAN = {
     "_id": "012345678910111213141516",
     "bc_filter_enabled": False,
@@ -1455,6 +1483,7 @@ async def test_object_oriented_network_configs(
         ([OUTLET_UP1], "plug_outlet_1", 1, 1),
         ([PDU_DEVICE_1], "dummy_usp_pdu_pro_usb_outlet_1", 1, 2),
         ([PDU_DEVICE_1], "dummy_usp_pdu_pro_outlet_2", 2, 2),
+        ([UPS_DEVICE_1], "dummy_ups_2u_pro_outlet_1", 1, 1),
     ],
 )
 async def test_outlet_switches(
diff --git a/tests/components/zwave_js/test_config_flow.py b/tests/components/zwave_js/test_config_flow.py
index ed8ad08afe3d16..2885a9b872adcd 100644
--- a/tests/components/zwave_js/test_config_flow.py
+++ b/tests/components/zwave_js/test_config_flow.py
@@ -1450,12 +1450,12 @@ async def test_esphome_discovery_intent_custom(
     assert len(mock_setup_entry.mock_calls) == 1
 
 
-@pytest.mark.usefixtures("supervisor", "addon_running", "addon_running", "addon_info")
+@pytest.mark.usefixtures("supervisor", "addon_running", "addon_info")
 async def test_esphome_discovery_intent_recommended(
     hass: HomeAssistant,
     set_addon_options: AsyncMock,
     addon_options: dict,
-    stop_addon: AsyncMock,
+    restart_addon: AsyncMock,
 ) -> None:
     """Test ESPHome discovery success path."""
     addon_options.update(
@@ -1480,6 +1480,31 @@ async def test_esphome_discovery_intent_recommended(
     assert result["step_id"] == "installation_type"
     assert result["menu_options"] == ["intent_recommended", "intent_custom"]
 
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"], {"next_step_id": "intent_recommended"}
+    )
+
+    assert result["type"] is FlowResultType.SHOW_PROGRESS
+    assert result["step_id"] == "start_addon"
+    assert set_addon_options.call_args == call(
+        "core_zwave_js",
+        AddonsOptions(
+            config={
+                "socket": "esphome://192.168.1.100:6053",
+                "s0_legacy_key": "new123",
+                "s2_access_control_key": "new456",
+                "s2_authenticated_key": "new789",
+                "s2_unauthenticated_key": "new987",
+                "lr_s2_access_control_key": "new654",
+                "lr_s2_authenticated_key": "new321",
+            }
+        ),
+    )
+
+    await hass.async_block_till_done()
+
+    assert restart_addon.call_args == call("core_zwave_js")
+
     with (
         patch(
             "homeassistant.components.zwave_js.async_setup", return_value=True
@@ -1489,9 +1514,8 @@ async def test_esphome_discovery_intent_recommended(
             return_value=True,
         ) as mock_setup_entry,
     ):
-        result = await hass.config_entries.flow.async_configure(
-            result["flow_id"], {"next_step_id": "intent_recommended"}
-        )
+        result = await hass.config_entries.flow.async_configure(result["flow_id"])
+        await hass.async_block_till_done()
 
     assert result["type"] is FlowResultType.CREATE_ENTRY
     assert result["title"] == TITLE
@@ -1509,22 +1533,6 @@ async def test_esphome_discovery_intent_recommended(
         "use_addon": True,
         "integration_created_addon": False,
     }
-    assert set_addon_options.call_args == call(
-        "core_zwave_js",
-        AddonsOptions(
-            config={
-                "socket": "esphome://192.168.1.100:6053",
-                "s0_legacy_key": "new123",
-                "s2_access_control_key": "new456",
-                "s2_authenticated_key": "new789",
-                "s2_unauthenticated_key": "new987",
-                "lr_s2_access_control_key": "new654",
-                "lr_s2_authenticated_key": "new321",
-            }
-        ),
-    )
-    assert stop_addon.call_count == 1
-    assert stop_addon.call_args == call("core_zwave_js")
     assert len(mock_setup.mock_calls) == 1
     assert len(mock_setup_entry.mock_calls) == 1
 
@@ -1584,6 +1592,109 @@ async def test_esphome_discovery_already_configured(
     assert stop_addon.call_args == call("core_zwave_js")
 
 
+@pytest.mark.usefixtures("supervisor", "addon_running", "backup_nvm", "restore_nvm")
+@pytest.mark.parametrize(
+    "esphome_discovery_info",
+    [
+        pytest.param(ESPHOME_DISCOVERY_INFO, id="different_home_id"),
+        pytest.param(ESPHOME_DISCOVERY_INFO_CLEAN, id="unknown_home_id"),
+    ],
+)
+async def test_esphome_discovery_migration(
+    hass: HomeAssistant,
+    addon_options: dict[str, Any],
+    set_addon_options: AsyncMock,
+    restart_addon: AsyncMock,
+    client: MagicMock,
+    integration: MockConfigEntry,
+    get_server_version: AsyncMock,
+    esphome_discovery_info: ESPHomeServiceInfo,
+) -> None:
+    """Test ESPHome discovery of a different adapter starts migration."""
+    addon_options["device"] = "/dev/ttyUSB0"
+    entry = integration
+    assert client.connect.call_count == 1
+    assert entry.unique_id == "3245146787"
+    hass.config_entries.async_update_entry(
+        entry,
+        data={
+            "url": "ws://localhost:3000",
+            "use_addon": True,
+            "usb_path": "/dev/ttyUSB0",
+        },
+    )
+
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN,
+        context={"source": config_entries.SOURCE_ESPHOME},
+        data=esphome_discovery_info,
+    )
+
+    assert result["type"] is FlowResultType.FORM
+    assert result["step_id"] == "confirm_usb_migration"
+    # The add-on config is not touched before the user confirms.
+    set_addon_options.assert_not_called()
+
+    result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
+
+    assert result["type"] is FlowResultType.SHOW_PROGRESS
+    assert result["step_id"] == "backup_nvm"
+
+    with patch("pathlib.Path.write_bytes") as mock_file:
+        await hass.async_block_till_done()
+        assert client.driver.controller.async_backup_nvm_raw.call_count == 1
+        assert mock_file.call_count == 1
+
+    result = await hass.config_entries.flow.async_configure(result["flow_id"])
+
+    assert result["type"] is FlowResultType.FORM
+    assert result["step_id"] == "instruct_unplug"
+
+    result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
+
+    assert entry.state is config_entries.ConfigEntryState.NOT_LOADED
+    assert result["type"] is FlowResultType.SHOW_PROGRESS
+    assert result["step_id"] == "start_addon"
+    assert set_addon_options.call_args == call(
+        "core_zwave_js",
+        AddonsOptions(
+            config={
+                CONF_ADDON_SOCKET: "esphome://192.168.1.100:6053",
+            }
+        ),
+    )
+
+    await hass.async_block_till_done()
+
+    assert restart_addon.call_args == call("core_zwave_js")
+    # The add-on start has finished and the next configure call below
+    # runs the finish step, which routes to the migration finish.
+    flow = hass.config_entries.flow.async_get(result["flow_id"])
+    assert flow["step_id"] == "finish_addon_setup"
+
+    _set_home_id(get_server_version, 3245146787)
+
+    result = await hass.config_entries.flow.async_configure(result["flow_id"])
+
+    assert result["type"] is FlowResultType.SHOW_PROGRESS
+    assert result["step_id"] == "restore_nvm"
+    assert client.connect.call_count == 2
+
+    await hass.async_block_till_done()
+    assert client.connect.call_count == 3
+    assert entry.state is config_entries.ConfigEntryState.LOADED
+    assert client.driver.controller.async_restore_nvm.call_count == 1
+
+    result = await hass.config_entries.flow.async_configure(result["flow_id"])
+
+    assert result["type"] is FlowResultType.ABORT
+    assert result["reason"] == "migration_successful"
+    assert entry.data["url"] == "ws://host1:3001"
+    assert entry.data["usb_path"] is None
+    assert entry.data["socket_path"] == "esphome://192.168.1.100:6053"
+    assert entry.data["use_addon"] is True
+
+
 @pytest.mark.usefixtures("supervisor", "addon_running", "addon_info")
 async def test_esphome_discovery_same_socket_no_reload(
     hass: HomeAssistant,
@@ -2232,7 +2343,7 @@ async def test_discovery_not_blocked_by_zeroconf_flow(hass: HomeAssistant) -> No
     assert result["reason"] == "already_in_progress"
 
 
-@pytest.mark.usefixtures("supervisor", "addon_running")
+@pytest.mark.usefixtures("supervisor", "addon_running", "restart_addon")
 async def test_usb_discovery_leaves_manual_entry_alone(
     hass: HomeAssistant,
     addon_options: dict[str, Any],
@@ -2268,10 +2379,18 @@ async def test_usb_discovery_leaves_manual_entry_alone(
         result["flow_id"], {"next_step_id": "intent_recommended"}
     )
 
+    assert result["type"] is FlowResultType.SHOW_PROGRESS
+    assert result["step_id"] == "start_addon"
+
+    await hass.async_block_till_done()
+
+    result = await hass.config_entries.flow.async_configure(result["flow_id"])
+
     assert result["type"] is FlowResultType.ABORT
     assert result["reason"] == "already_configured"
+    # The add-on config now points at the discovered adapter, but the
+    # manual entry stays untouched.
     assert entry.data == {"url": "ws://external-server:3000"}
-    set_addon_options.assert_not_called()
 
 
 @pytest.mark.usefixtures("supervisor", "addon_info")
diff --git a/tests/test_util/aiohttp.py b/tests/test_util/aiohttp.py
index 687de2ece0b163..5a62ae1672e76d 100644
--- a/tests/test_util/aiohttp.py
+++ b/tests/test_util/aiohttp.py
@@ -64,6 +64,7 @@ def request(
         side_effect=None,
         closing=None,
         timeout=None,
+        history=(),
     ):
         """Mock a request."""
         if not isinstance(url, RETYPE):
@@ -83,6 +84,7 @@ def request(
             headers=headers,
             side_effect=side_effect,
             closing=closing,
+            history=history,
         )
         self._mocks.append(resp)
         return resp
@@ -185,6 +187,7 @@ def __init__(
         headers=None,
         side_effect=None,
         closing=None,
+        history=(),
     ) -> None:
         """Initialize a fake response."""
         if json is not None:
@@ -197,6 +200,7 @@ def __init__(
         self.method = method
         self._url = url
         self.status = status
+        self.history = history
         self._response = response
         self.exc = exc
         self.side_effect = side_effect