diff --git a/esphome/components/dashboard_import/__init__.py b/esphome/components/dashboard_import/__init__.py
index c27669d77e..906e9762bf 100644
--- a/esphome/components/dashboard_import/__init__.py
+++ b/esphome/components/dashboard_import/__init__.py
@@ -4,7 +4,6 @@ import re
 import secrets
 from typing import Any
 
-import requests
 from ruamel.yaml import YAML
 
 from esphome import git
@@ -13,7 +12,7 @@ from esphome.components.packages import validate_source_shorthand
 import esphome.config_validation as cv
 from esphome.const import CONF_ESPHOME, CONF_PROJECT, CONF_REF, CONF_WIFI
 import esphome.final_validate as fv
-from esphome.happy_eyeballs import ensure_happy_eyeballs
+from esphome.net_retry import fetch_with_retry, http_request
 from esphome.types import ConfigType
 from esphome.yaml_util import dump
 
@@ -111,14 +110,20 @@ def import_config(
 
     if git_file.query and "full_config" in git_file.query:
         url = git_file.raw_url
-        try:
-            ensure_happy_eyeballs()
-            req = requests.get(url, timeout=30)
+
+        # Deferred so config-time imports of this component stay light;
+        # http_request does the lazy import for the request itself.
+        import requests
+
+        def _fetch() -> str:
+            req = http_request("GET", url, timeout=30)
             req.raise_for_status()
+            return req.text
+
+        try:
+            contents = fetch_with_retry(url, _fetch, what="Import")
         except requests.exceptions.RequestException as e:
             raise ValueError(f"Error while fetching {url}: {e}") from e
-
-        contents = req.text
         yaml = YAML()
         loaded_yaml = yaml.load(contents)
         if (
diff --git a/esphome/components/nrf52/framework.py b/esphome/components/nrf52/framework.py
index cffc630fcc..5e2cf197fb 100644
--- a/esphome/components/nrf52/framework.py
+++ b/esphome/components/nrf52/framework.py
@@ -5,16 +5,14 @@ from pathlib import Path
 import platform
 import shutil
 import sys
-import tempfile
 
 from esphome.build_helpers.tools_cache import SDK_NRF_TOOLS_CACHE, tools_cache_path
 import esphome.config_validation as cv
 from esphome.const import KEY_CORE, KEY_FRAMEWORK_VERSION
 from esphome.core import CORE, EsphomeError
 from esphome.framework_helpers import (
-    archive_extract_all,
     create_venv,
-    download_from_mirrors,
+    download_and_extract,
     get_python_env_executable_path,
     rmdir,
     run_command_ok,
@@ -338,34 +336,37 @@ def check_and_install() -> None:
     if not sentinel.exists():
         rmdir(toolchains_dir, msg=f"Clean up {TOOLCHAIN_VERSION} toolchain environment")
         sysname, machine, extension = _get_toolchain_platform_info()
-        with tempfile.NamedTemporaryFile() as tmp:
-            _LOGGER.info("Downloading Zephyr SDK %s minimal ...", TOOLCHAIN_VERSION)
-            download_from_mirrors(
-                SDK_NG_MINIMAL_MIRRORS,
-                {
-                    "VERSION": TOOLCHAIN_VERSION,
-                    "sysname": sysname,
-                    "machine": machine,
-                    "extension": extension,
-                },
-                tmp.file,
-            )
-            archive_extract_all(tmp.file, toolchains_dir, progress_header="Extracting")
-        with tempfile.NamedTemporaryFile() as tmp:
-            _LOGGER.info("Downloading %s toolchain ...", TOOLCHAIN_VERSION)
-            download_from_mirrors(
+        substitutions = {
+            "VERSION": TOOLCHAIN_VERSION,
+            "sysname": sysname,
+            "machine": machine,
+            "extension": extension,
+        }
+        # Downloaded next to the destination (not a temp file) so an
+        # interrupted download's .part file resumes on the next run.
+        for mirrors, extract_dir, what, slug in (
+            (SDK_NG_MINIMAL_MIRRORS, toolchains_dir, "Zephyr SDK minimal", "minimal"),
+            (
                 SDK_NG_TOOLCHAIN_MIRRORS,
-                {
-                    "VERSION": TOOLCHAIN_VERSION,
-                    "sysname": sysname,
-                    "machine": machine,
-                    "extension": extension,
-                },
-                tmp.file,
-            )
-            archive_extract_all(
-                tmp.file,
                 toolchains_dir / "arm-zephyr-eabi",
+                "toolchain",
+                "toolchain",
+            ),
+        ):
+            _LOGGER.info("Downloading %s %s ...", TOOLCHAIN_VERSION, what)
+            download_and_extract(
+                mirrors,
+                substitutions,
+                toolchains_dir.with_name(f"{toolchains_dir.name}.{slug}.archive"),
+                extract_dir,
                 progress_header="Extracting",
             )
+        # Best-effort prune of resume leftovers, including a previous
+        # TOOLCHAIN_VERSION's orphans; the SDK archives are hundreds of MB.
+        # A locked file must not discard the just-completed install.
+        for leftover in toolchains_dir.parent.glob("*.archive.part*"):
+            try:
+                leftover.unlink()
+            except OSError as err:
+                _LOGGER.debug("Could not remove %s: %s", leftover, err)
         sentinel.touch()
diff --git a/esphome/espidf/framework.py b/esphome/espidf/framework.py
index bebe513f75..25fbc1aa52 100644
--- a/esphome/espidf/framework.py
+++ b/esphome/espidf/framework.py
@@ -21,8 +21,8 @@ from esphome.build_helpers.tools_cache import IDF_TOOLS_CACHE, tools_cache_path
 from esphome.core import Version
 from esphome.framework_helpers import (
     PathType,
-    archive_extract_all,
     create_venv,
+    download_and_extract,
     download_from_mirrors,
     download_with_resume,
     failure_reason,
@@ -901,20 +901,13 @@ def _check_esphome_idf_framework_install(
             # a temp file) so an interrupted download resumes on the next
             # run; the cache is pruned after a successful install anyway.
             tarball_path = get_idf_tools_path() / "dist" / f"esp-idf-{version}.tar.xz"
-            download_from_mirrors(mirrors, substitutions, tarball_path)
-
-            _LOGGER.info("Extracting ESP-IDF %s framework ...", version)
-            try:
-                with tarball_path.open("rb") as tarball:
-                    archive_extract_all(
-                        tarball, framework_path, progress_header="Extracting"
-                    )
-            finally:
-                # Success: drop the archive rather than caching ~70MB twice.
-                # Failure: a corrupt archive (e.g. torn by an unclean
-                # shutdown) must not be reused — without a checksum only a
-                # failed extraction can expose it, so force a re-download.
-                tarball_path.unlink(missing_ok=True)
+            download_and_extract(
+                mirrors,
+                substitutions,
+                tarball_path,
+                framework_path,
+                progress_header="Extracting",
+            )
         extracted_marker.touch()
 
     # Idempotent post-extract patch: written every invocation so a build
diff --git a/esphome/external_files.py b/esphome/external_files.py
index 58be4a7c26..ff3b5baf7a 100644
--- a/esphome/external_files.py
+++ b/esphome/external_files.py
@@ -5,6 +5,7 @@ from concurrent.futures import ThreadPoolExecutor
 import contextlib
 from dataclasses import dataclass, field
 from datetime import UTC, datetime
+from functools import partial
 import hashlib
 import logging
 import os
@@ -14,9 +15,8 @@ import time
 import esphome.config_validation as cv
 from esphome.const import CONF_FILE, CONF_TYPE, CONF_URL, __version__
 from esphome.core import CORE, EsphomeError, TimePeriodSeconds
-from esphome.happy_eyeballs import ensure_happy_eyeballs
 from esphome.helpers import write_file
-from esphome.net_retry import fetch_with_retry
+from esphome.net_retry import fetch_with_retry, http_request
 from esphome.types import ConfigType
 
 _LOGGER = logging.getLogger(__name__)
@@ -143,7 +143,6 @@ def has_remote_file_changed(
     # Deferred so configs with no remote files skip the heavy import.
     import requests
 
-    ensure_happy_eyeballs()
     if local_file_path.exists():
         _LOGGER.debug("has_remote_file_changed: File exists at %s", local_file_path)
         try:
@@ -165,9 +164,7 @@ def has_remote_file_changed(
             # the GET's own retry.
             response = fetch_with_retry(
                 url,
-                lambda: requests.head(
-                    url, headers=headers, timeout=timeout, allow_redirects=True
-                ),
+                partial(http_request, "HEAD", url, headers=headers, timeout=timeout),
                 what="Revalidation",
             )
 
@@ -282,7 +279,6 @@ def download_content(
             ) from failure.cause
         # The file appeared since the failure; revalidate normally.
         del run_data.failed_paths[path]
-    ensure_happy_eyeballs()
     if CORE.skip_external_update and path.exists():
         _LOGGER.debug("Skipping update for %s (refresh disabled)", url)
         run_data.unchecked_paths.add(path)
@@ -304,7 +300,8 @@ def download_content(
     _LOGGER.debug("Saving to %s", path)
 
     def _fetch() -> tuple[requests.Response, bytes]:
-        req = requests.get(
+        req = http_request(
+            "GET",
             url,
             timeout=timeout,
             headers={"User-agent": f"ESPHome/{__version__} (https://esphome.io)"},
@@ -371,7 +368,6 @@ def download_content_many(
     unique = list(seen.values())
     if not unique:
         return
-    ensure_happy_eyeballs()
     _LOGGER.info("Checking %d %s for updates", len(unique), description)
 
     def _download_one(file: RemoteFile) -> None:
diff --git a/esphome/framework_helpers.py b/esphome/framework_helpers.py
index 28fd9856a6..aab7acc0e8 100644
--- a/esphome/framework_helpers.py
+++ b/esphome/framework_helpers.py
@@ -15,9 +15,12 @@ import threading
 import time
 from typing import IO, TYPE_CHECKING
 
-from esphome.happy_eyeballs import ensure_happy_eyeballs
 from esphome.helpers import ProgressBar, rmtree
-from esphome.net_retry import NETWORK_MAX_ATTEMPTS, is_transient_download_error
+from esphome.net_retry import (
+    NETWORK_MAX_ATTEMPTS,
+    http_request,
+    is_transient_download_error,
+)
 
 if TYPE_CHECKING:
     import requests
@@ -624,12 +627,10 @@ def _open_ranged(
     Raises on connect errors and HTTP error statuses; the response is closed
     on failure.
     """
-    import requests
-
     headers = {"Range": f"bytes={offset}-"} if offset else {}
     if offset and validator:
         headers["If-Range"] = validator
-    resp = requests.get(url, stream=True, timeout=timeout, headers=headers)
+    resp = http_request("GET", url, stream=True, timeout=timeout, headers=headers)
     if offset and resp.status_code == 416:
         resp.close()
         return None, offset
@@ -965,8 +966,6 @@ def download_with_resume(
 
     from esphome.core import EsphomeError
 
-    ensure_happy_eyeballs()
-
     dest = Path(dest)
     part = _part_path(dest)
     meta = part.with_name(part.name + ".meta")
@@ -1102,20 +1101,9 @@ def failure_reason(e: BaseException) -> str:
     return str(e).split(" for url: ", maxsplit=1)[0] or repr(e)
 
 
-def _spent_attempts_error(e: Exception, attempts: int) -> Exception:
-    """Wrap a failure whose mirror already consumed download attempts, so
-    the sweep classifies it as permanent."""
-    from esphome.core import EsphomeError
-
-    err = EsphomeError(f"failed after {attempts} attempts: {failure_reason(e)}")
-    err.__cause__ = e
-    return err
-
-
 def _try_mirrors_once(
     urls: list[str],
-    path_target: Path | None,
-    f: IO[bytes] | None,
+    path_target: Path,
     timeout: int,
     failures: list[tuple[str, Exception]],
     progress: Callable[[int], None] | None = None,
@@ -1134,109 +1122,73 @@ def _try_mirrors_once(
     for url in urls:
         _LOGGER.debug("Trying to download from %s", url)
 
-        # Path targets delegate to download_with_resume so a partial
-        # download persists (and resumes) across esphome runs.
-        if path_target is not None:
-            try:
-                download_with_resume(
-                    url,
-                    path_target,
-                    attempts=_MIRROR_ATTEMPTS,
-                    timeout=timeout,
-                    # Pre-body failures (connect/HTTP errors) fall to the
-                    # next mirror immediately; only mid-stream drops
-                    # retry-with-resume on the same URL.
-                    retry_connect_errors=False,
-                    progress=progress,
-                )
-                return url
-            except (requests.RequestException, OSError, EsphomeError) as e:
-                # Everything download_with_resume classifies as a download
-                # failure; programming errors propagate.
-                _LOGGER.debug("Failed to download %s: %s", url, str(e))
-                failures.append((url, e))
-                continue
-
-        # File-like targets download here; mid-stream failures retry the
-        # same mirror with resume (see download_with_resume) instead of
-        # starting over. There is no checksum to verify a resumed file
-        # against, so a stitch is only trusted when the server proves
-        # consistency: the If-Range validator guarantees 206 only for
-        # unchanged content, and the expected total length (when the first
-        # response carried one) guards against short or shifted bodies.
-        # Without a validator the retry restarts from zero.
-        offset = 0
-        expected_total = 0
-        validator = None
-        for attempt in range(_MIRROR_ATTEMPTS):
-            try:
-                resp, offset = _open_ranged(url, offset, timeout, validator)
-            except (requests.RequestException, OSError) as e:
-                # Connect/HTTP error, no bytes flowed — next mirror. Wrap
-                # when earlier attempts were already spent on this mirror.
-                _LOGGER.debug("Failed to download %s: %s", url, str(e))
-                failures.append(
-                    (url, _spent_attempts_error(e, attempt + 1) if attempt else e)
-                )
-                break
-
-            try:
-                # A None response means HTTP 416: the file already holds
-                # every byte the server has (a drop after the last byte);
-                # only the length check below remains.
-                if resp is not None:
-                    with resp:
-                        if offset == 0:
-                            validator = _response_validator(resp)
-                            expected_total = _content_length(resp)
-                        _stream_response_to_file(resp, f, offset, progress=progress)
-
-                if expected_total and f.tell() != expected_total:
-                    raise EsphomeError(
-                        f"size mismatch: expected {expected_total}, got {f.tell()}"
-                    )
-                if not expected_total:
-                    # Same trust decision as download_with_resume's
-                    # unverifiable promotion; surface it at the same level.
-                    _LOGGER.debug(
-                        "Downloaded %s without any way to verify completeness",
-                        url,
-                    )
-
-                _LOGGER.debug("Downloaded successfully from: %s", url)
-
-                # Reset file pointer and return
-                f.seek(0)
-                return url
-
-            except (requests.RequestException, OSError, EsphomeError) as e:
-                # Mid-stream drop: keep the received bytes and retry this
-                # mirror from the current position — but only when the
-                # server gave a validator to resume against safely AND a
-                # total length to prove the stitched file complete (the
-                # length check above is the only verification here).
-                _LOGGER.debug("Failed to download %s: %s", url, str(e))
-                if validator and expected_total:
-                    offset = f.tell()
-                else:
-                    _LOGGER.debug(
-                        "Restarting %s from zero: cannot prove a "
-                        "resumed file complete (validator=%s, total=%s)",
-                        url,
-                        validator is not None,
-                        expected_total,
-                    )
-                    offset = 0
-                if attempt == _MIRROR_ATTEMPTS - 1:
-                    failures.append((url, _spent_attempts_error(e, _MIRROR_ATTEMPTS)))
+        # Delegate to download_with_resume so a partial download persists
+        # (and resumes) across esphome runs.
+        try:
+            download_with_resume(
+                url,
+                path_target,
+                attempts=_MIRROR_ATTEMPTS,
+                timeout=timeout,
+                # Pre-body failures (connect/HTTP errors) fall to the
+                # next mirror immediately; only mid-stream drops
+                # retry-with-resume on the same URL.
+                retry_connect_errors=False,
+                progress=progress,
+            )
+            return url
+        except (requests.RequestException, OSError, EsphomeError) as e:
+            # Everything download_with_resume classifies as a download
+            # failure; programming errors propagate.
+            _LOGGER.debug("Failed to download %s: %s", url, str(e))
+            failures.append((url, e))
 
     return None
 
 
+def download_and_extract(
+    mirrors: list[str],
+    substitutions: dict[str, str],
+    archive_path: PathType,
+    extract_dir: PathType,
+    timeout: int = 30,
+    progress_header: str | None = None,
+    progress: Callable[[int], None] | None = None,
+) -> str:
+    """Download an archive from ``mirrors`` to ``archive_path``, extract it
+    into ``extract_dir``, and delete the archive.
+
+    The archive should live next to its destination (not in a temp dir) so
+    an interrupted download's ``.part`` file resumes on the next run. The
+    archive is deleted whether extraction succeeds or fails: a
+    complete-but-corrupt file (e.g. torn by an unclean shutdown) must not
+    poison the next run, and without a checksum only a failed extraction
+    can expose it.
+
+    Returns the source URL the download came from.
+    """
+    archive_path = Path(archive_path)
+    url = download_from_mirrors(
+        mirrors, substitutions, archive_path, timeout=timeout, progress=progress
+    )
+    try:
+        archive_extract_all(archive_path, extract_dir, progress_header=progress_header)
+    finally:
+        # Best-effort: an AV handle on the just-written archive (Windows)
+        # must not replace the real extraction error or fail a successful
+        # extraction. A surviving archive is harmless; download_with_resume
+        # re-verifies or re-downloads it next run.
+        try:
+            archive_path.unlink(missing_ok=True)
+        except OSError as err:
+            _LOGGER.debug("Could not remove archive %s: %s", archive_path, err)
+    return url
+
+
 def download_from_mirrors(
     mirrors: list[str],
     substitutions: dict[str, str],
-    target: io.RawIOBase | IO[bytes] | PathType,
+    target: PathType,
     timeout: int = 30,
     progress: Callable[[int], None] | None = None,
 ) -> str:
@@ -1246,7 +1198,7 @@ def download_from_mirrors(
     Args:
         mirrors: list of mirror URLs
         substitutions: Dictionary of substitutions to apply to URLs
-        target: Target file path or file-like object
+        target: Target file path
         timeout: Download timeout in seconds
         progress: Passed through to the download (see ``download_with_resume``);
             replaces the built-in per-file bar
@@ -1258,9 +1210,8 @@ def download_from_mirrors(
     ``substitutions`` are skipped, so callers can offer templates that only
     apply to some downloads.
 
-    A path target downloads through ``download_with_resume``, so an
-    interrupted download resumes on the next esphome run; a file-like target
-    only resumes mid-stream drops within this call.
+    The target downloads through ``download_with_resume``, so an
+    interrupted download resumes on the next esphome run.
 
     When every mirror fails and at least one failure is transient (dropped
     connection, timeout, HTTP 429/5xx), the whole list is retried with a
@@ -1274,21 +1225,11 @@ def download_from_mirrors(
     """
     from esphome.core import EsphomeError
 
-    ensure_happy_eyeballs()
+    if not isinstance(target, (str, os.PathLike)):
+        raise TypeError(f"target must be a str or Path: {type(target)}")
+    path_target = Path(target)
 
-    # 1. Classify the target: filesystem path or open file object
-    path_target: Path | None = None
-    f: IO[bytes] | None = None
-    if isinstance(target, (str, os.PathLike)):
-        path_target = Path(target)
-    elif isinstance(target, (io.RawIOBase, io.IOBase)):
-        f = target
-    else:
-        raise TypeError(
-            f"target must be str, Path, or file-like object: {type(target)}"
-        )
-
-    # 2. Resolve the mirror templates (invariant across retry sweeps)
+    # 1. Resolve the mirror templates (invariant across retry sweeps)
     urls: list[str] = []
     skipped: list[tuple[str, str]] = []
     for mirror in mirrors:
@@ -1307,7 +1248,7 @@ def download_from_mirrors(
             _LOGGER.warning("Skipping malformed mirror URL template %s: %r", mirror, e)
             skipped.append((mirror, f"skipped ({e!r})"))
 
-    # 3. Sweep the mirror list, retrying transient failures with backoff:
+    # 2. Sweep the mirror list, retrying transient failures with backoff:
     # a single pass keeps mirror failover fast, re-sweeping keeps one
     # network blip from failing the build when only one mirror applies.
     failures: list[tuple[str, Exception]] = []
@@ -1315,7 +1256,7 @@ def download_from_mirrors(
         sweep_failures: list[tuple[str, Exception]] = []
         if (
             url := _try_mirrors_once(
-                urls, path_target, f, timeout, sweep_failures, progress
+                urls, path_target, timeout, sweep_failures, progress
             )
         ) is not None:
             return url
@@ -1342,14 +1283,11 @@ def download_from_mirrors(
             # steady during the backoff instead of rewinding to zero
             done = 0
             if progress is not None:
-                if f is not None:
-                    done = f.tell()
-                else:
-                    part = _part_path(path_target)
-                    done = part.stat().st_size if part.is_file() else 0
+                part = _part_path(path_target)
+                done = part.stat().st_size if part.is_file() else 0
             _cancellable_sleep(delay, progress, done)
 
-    # 4. Report every attempted URL if all mirrors failed. failures spans
+    # 3. Report every attempted URL if all mirrors failed. failures spans
     # all sweeps (deduplicated by URL and reason), so neither an early
     # mirror's failure nor an earlier sweep's failure mode is hidden.
     if failures:
diff --git a/esphome/happy_eyeballs.py b/esphome/happy_eyeballs.py
index ebfb94f1f9..35092e7daa 100644
--- a/esphome/happy_eyeballs.py
+++ b/esphome/happy_eyeballs.py
@@ -12,6 +12,7 @@ from __future__ import annotations
 
 import logging
 import socket
+import threading
 from typing import TYPE_CHECKING, Any
 
 if TYPE_CHECKING:
@@ -27,20 +28,27 @@ HAPPY_EYEBALLS_DELAY = 0.25
 _THREAD_WAIT_BUFFER = 5.0
 
 
+# Serialises the check-then-patch so concurrent first calls (download worker
+# threads fanning out) build the replacement exactly once.
+_PATCH_LOCK = threading.Lock()
+
+
 def ensure_happy_eyeballs() -> None:
     """Make urllib3 (and therefore requests) connect with Happy Eyeballs.
 
-    Idempotent; call before performing requests-based downloads.
+    Idempotent and thread-safe; call before performing requests-based
+    downloads.
     """
     stock: Callable[..., socket.socket] | None = None
     try:
         import urllib3.util.connection
 
-        stock = urllib3.util.connection.create_connection
-        if getattr(stock, "_esphome_patched", False):
-            return
+        with _PATCH_LOCK:
+            stock = urllib3.util.connection.create_connection
+            if getattr(stock, "_esphome_patched", False):
+                return
 
-        urllib3.util.connection.create_connection = _make_create_connection()
+            urllib3.util.connection.create_connection = _make_create_connection()
     except (ImportError, AttributeError) as err:  # urllib3 internals moved
         # WARNING: degraded mode brings back the stalls this module prevents.
         _LOGGER.warning(
diff --git a/esphome/net_retry.py b/esphome/net_retry.py
index f7e6e601ea..b91b333114 100644
--- a/esphome/net_retry.py
+++ b/esphome/net_retry.py
@@ -1,4 +1,4 @@
-"""Retry policy for HTTP downloads.
+"""Retry policy and raw HTTP entry point for downloads.
 
 Kept import-light on purpose: this module is imported at config time, so it
 must not pull in requests (a heavy import, ~85ms) at module scope.
@@ -9,6 +9,12 @@ from __future__ import annotations
 from collections.abc import Callable
 import logging
 import time
+from typing import TYPE_CHECKING, Literal
+
+from esphome.happy_eyeballs import ensure_happy_eyeballs
+
+if TYPE_CHECKING:
+    import requests
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -112,3 +118,34 @@ def fetch_with_retry[T](url: str, fetch: Callable[[], T], what: str = "Download"
             )
             time.sleep(delay)
     return fetch()
+
+
+def http_request(
+    method: Literal["GET", "HEAD"],
+    url: str,
+    *,
+    timeout: float | tuple[float, float],
+    stream: bool = False,
+    headers: dict[str, str] | None = None,
+    allow_redirects: bool = True,
+) -> requests.Response:
+    """Perform one HTTP request with the Happy Eyeballs patch in place.
+
+    Every ESPHome file download funnels through here so the urllib3 patch
+    and the lazy requests import live in exactly one place. Status handling,
+    retries and streaming stay with the caller. The web server OTA and log
+    clients bypass this on purpose: they iterate already-resolved device
+    addresses themselves, so the patch buys them nothing.
+    """
+    import requests
+
+    ensure_happy_eyeballs()
+    # Dispatched through requests.get/head/... (not requests.request) so
+    # tests patching those entry points keep working.
+    return getattr(requests, method.lower())(
+        url,
+        timeout=timeout,
+        stream=stream,
+        headers=headers or {},
+        allow_redirects=allow_redirects,
+    )
diff --git a/esphome/platformio/library.py b/esphome/platformio/library.py
index 50e1408b13..1792647d6b 100644
--- a/esphome/platformio/library.py
+++ b/esphome/platformio/library.py
@@ -24,7 +24,6 @@ import logging
 import os
 from pathlib import Path, PurePosixPath
 import re
-import tempfile
 from typing import Any
 from urllib.parse import urlsplit, urlunsplit
 from urllib.request import url2pathname
@@ -32,8 +31,7 @@ from urllib.request import url2pathname
 from esphome import git
 from esphome.core import CORE, EsphomeError, Library
 from esphome.framework_helpers import (
-    archive_extract_all,
-    download_from_mirrors,
+    download_and_extract,
     failure_reason,
     rmdir,
     run_batch_downloads,
@@ -147,18 +145,21 @@ class URLSource(Source):
         if not extracted_marker.is_file() or force:
             rmdir(path, msg=f"Clean up library directory {path}")
 
-            # Download in temporary file
-            with tempfile.NamedTemporaryFile() as tmp:
-                if progress is None:
-                    # A batch caller draws one combined bar and logs the list
-                    _LOGGER.info("Downloading %s ...", self.url)
-                _LOGGER.debug("Location: %s", path)
+            if progress is None:
+                # A batch caller draws one combined bar and logs the list
+                _LOGGER.info("Downloading %s ...", self.url)
+            _LOGGER.debug("Location: %s", path)
 
-                download_from_mirrors([self.url], {}, tmp.file, progress=progress)
-
-                _LOGGER.debug("Extracting archive to %s ...", path)
-                archive_extract_all(tmp.file, path)
-                extracted_marker.touch()
+            # The sibling archive path lets an interrupted download's .part
+            # file survive and resume on the next esphome run.
+            download_and_extract(
+                [self.url],
+                {},
+                path.with_name(f"{path.name}.archive"),
+                path,
+                progress=progress,
+            )
+            extracted_marker.touch()
         return path
 
     def __str__(self):
diff --git a/tests/unit_tests/test_dashboard_import.py b/tests/unit_tests/test_dashboard_import.py
index 427bee0f86..46a2fa5db0 100644
--- a/tests/unit_tests/test_dashboard_import.py
+++ b/tests/unit_tests/test_dashboard_import.py
@@ -10,8 +10,10 @@ during the adoption flow and depend on the output's ``esphome.name``
 from __future__ import annotations
 
 from pathlib import Path
+from unittest.mock import MagicMock, patch
 
 import pytest
+import requests as req
 import yaml as pyyaml
 
 from esphome.components.dashboard_import import import_config
@@ -201,3 +203,56 @@ def test_import_refuses_to_overwrite_existing_yaml(tmp_path: Path) -> None:
         )
     # Original content survives unchanged.
     assert yaml_path.read_text() == "# user's hand-edited config\n"
+
+
+def _full_config_kwargs(yaml_path: Path) -> dict:
+    return {
+        "path": str(yaml_path),
+        "name": "kitchen",
+        "friendly_name": None,
+        "project_name": "acme.kitchen-light",
+        "import_url": "github://acme/firmware/kitchen.yaml@main?full_config",
+    }
+
+
+def test_full_config_import_fetches_and_writes_contents(tmp_path: Path) -> None:
+    yaml_path = tmp_path / "kitchen.yaml"
+    resp = MagicMock(text="esphome:\n  name: orig\n")
+    with patch(
+        "esphome.components.dashboard_import.http_request", return_value=resp
+    ) as mock_req:
+        import_config(**_full_config_kwargs(yaml_path))
+    assert yaml_path.read_text() == "esphome:\n  name: orig\n"
+    assert mock_req.call_args[0][0] == "GET"
+
+
+def test_full_config_import_retries_transient_errors(tmp_path: Path) -> None:
+    """The fetch goes through the shared retry policy: a transient network
+    error is retried instead of failing the adoption immediately."""
+    yaml_path = tmp_path / "kitchen.yaml"
+    resp = MagicMock(text="esphome:\n  name: orig\n")
+    with (
+        patch(
+            "esphome.components.dashboard_import.http_request",
+            side_effect=[req.ConnectionError("reset"), resp],
+        ),
+        patch("esphome.net_retry.time.sleep") as mock_sleep,
+    ):
+        import_config(**_full_config_kwargs(yaml_path))
+    assert yaml_path.exists()
+    mock_sleep.assert_called_once_with(2)
+
+
+def test_full_config_import_wraps_permanent_errors_in_value_error(
+    tmp_path: Path,
+) -> None:
+    """device-builder depends on the ValueError contract for fetch failures."""
+    resp = MagicMock()
+    resp.raise_for_status.side_effect = req.HTTPError(
+        "404", response=MagicMock(status_code=404)
+    )
+    with (
+        patch("esphome.components.dashboard_import.http_request", return_value=resp),
+        pytest.raises(ValueError, match="Error while fetching"),
+    ):
+        import_config(**_full_config_kwargs(tmp_path / "kitchen.yaml"))
diff --git a/tests/unit_tests/test_espidf_framework.py b/tests/unit_tests/test_espidf_framework.py
index 81e70d2959..42a00ef0d7 100644
--- a/tests/unit_tests/test_espidf_framework.py
+++ b/tests/unit_tests/test_espidf_framework.py
@@ -371,10 +371,9 @@ def _fake_download_from_mirrors(
 ) -> str:
     """Stand-in for download_from_mirrors that creates path targets, since
     the framework code opens the downloaded tarball afterwards."""
-    if isinstance(target, (str, os.PathLike)):
-        path = Path(target)
-        path.parent.mkdir(parents=True, exist_ok=True)
-        path.touch()
+    path = Path(target)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.touch()
     return "https://example.com/idf.tar.xz"
 
 
@@ -384,13 +383,15 @@ def espidf_mocks(setup_core: Path):
     # archive_extract_all is mocked, so pre-create the framework dir that the
     # extracted-marker touch writes into.
     _get_framework_path(_IDF_VERSION).mkdir(parents=True, exist_ok=True)
+    # One mock covers the tarball (via framework_helpers.download_and_extract)
+    # and the constraints file (espidf-bound download_from_mirrors), so call
+    # counts and ordering assertions span the two.
+    download = MagicMock(side_effect=_fake_download_from_mirrors)
     with (
         patch("esphome.espidf.framework.rmdir") as rmdir_mock,
-        patch(
-            "esphome.espidf.framework.download_from_mirrors",
-            side_effect=_fake_download_from_mirrors,
-        ) as download,
-        patch("esphome.espidf.framework.archive_extract_all") as extract,
+        patch("esphome.framework_helpers.download_from_mirrors", download),
+        patch("esphome.espidf.framework.download_from_mirrors", download),
+        patch("esphome.framework_helpers.archive_extract_all") as extract,
         patch("esphome.espidf.framework.create_venv") as venv,
         patch("esphome.espidf.framework.run_command_ok", return_value=True) as run_ok,
         patch(
diff --git a/tests/unit_tests/test_framework_helpers.py b/tests/unit_tests/test_framework_helpers.py
index 9f72313280..8844212600 100644
--- a/tests/unit_tests/test_framework_helpers.py
+++ b/tests/unit_tests/test_framework_helpers.py
@@ -2,6 +2,7 @@
 
 # pylint: disable=protected-access
 
+import gzip
 import hashlib
 import importlib.util
 import io
@@ -31,6 +32,7 @@ from esphome.framework_helpers import (
     _zip_extract_all,
     archive_extract_all,
     create_venv,
+    download_and_extract,
     download_from_mirrors,
     download_with_resume,
     get_project_compile_flags,
@@ -1339,20 +1341,20 @@ class TestDownloadFromMirrors:
         assert url == "https://example.com/f"
         assert target.read_bytes() == b"filedata"
 
-    def test_file_object_target_reports_progress(self) -> None:
-        """The library prefetch's production path: a file-object target
-        streams through the mirror fallback and ticks the tracker."""
-        buf = io.BytesIO()
+    def test_progress_callback_reports_bytes(self, tmp_path: Path) -> None:
+        """The library prefetch's production path: the mirror download ticks
+        the caller's tracker instead of drawing its own bar."""
+        target = tmp_path / "f.bin"
         ticks: list[int] = []
         with patch(
             "requests.get",
             return_value=_mock_response(b"filedata"),
         ):
             url = download_from_mirrors(
-                ["https://example.com/f"], {}, buf, progress=ticks.append
+                ["https://example.com/f"], {}, target, progress=ticks.append
             )
         assert url == "https://example.com/f"
-        assert buf.getvalue() == b"filedata"
+        assert target.read_bytes() == b"filedata"
         assert ticks and ticks[-1] == len(b"filedata")
 
     def test_substitutions_applied_to_url(self, tmp_path: Path) -> None:
@@ -1460,8 +1462,8 @@ class TestDownloadFromMirrors:
             ei.value
         )
 
-    def test_falls_back_to_second_mirror(self) -> None:
-        buf = io.BytesIO()
+    def test_falls_back_to_second_mirror(self, tmp_path: Path) -> None:
+        target = tmp_path / "f.bin"
         with patch(
             "requests.get",
             side_effect=[_mock_response(b"", ok=False), _mock_response(b"second")],
@@ -1469,18 +1471,18 @@ class TestDownloadFromMirrors:
             url = download_from_mirrors(
                 ["https://mirror1.com/f", "https://mirror2.com/f"],
                 {},
-                buf,
+                target,
             )
         assert url == "https://mirror2.com/f"
-        assert buf.getvalue() == b"second"
+        assert target.read_bytes() == b"second"
 
-    def test_mid_stream_drop_resumes_same_mirror(self) -> None:
+    def test_mid_stream_drop_resumes_same_mirror(self, tmp_path: Path) -> None:
         """A mid-stream failure retries the same mirror with Range and
         If-Range headers, keeping the bytes already received, before falling
         to the next."""
         first = _interrupted_response(b"1234", etag='"v1"')
         first.headers = {**first.headers, "content-length": "8"}
-        buf = io.BytesIO()
+        target = tmp_path / "f.bin"
         with patch(
             "requests.get",
             side_effect=[first, _resumed_response(b"5678")],
@@ -1488,10 +1490,10 @@ class TestDownloadFromMirrors:
             url = download_from_mirrors(
                 ["https://mirror1.com/f", "https://mirror2.com/f"],
                 {},
-                buf,
+                target,
             )
         assert url == "https://mirror1.com/f"
-        assert buf.getvalue() == b"12345678"
+        assert target.read_bytes() == b"12345678"
         assert mock_get.call_count == 2
         assert mock_get.call_args_list[1][0][0] == "https://mirror1.com/f"
         # the resume is conditional on the content being unchanged
@@ -1500,48 +1502,6 @@ class TestDownloadFromMirrors:
             "If-Range": '"v1"',
         }
 
-    def test_mid_stream_drop_without_validator_restarts(self) -> None:
-        """A server offering no ETag/Last-Modified cannot be resumed safely;
-        the retry restarts from zero instead of stitching unverified bytes."""
-        buf = io.BytesIO()
-        with patch(
-            "requests.get",
-            side_effect=[_interrupted_response(b"1234"), _mock_response(b"full")],
-        ) as mock_get:
-            download_from_mirrors(["https://mirror1.com/f"], {}, buf)
-        assert buf.getvalue() == b"full"
-        assert "Range" not in mock_get.call_args_list[1][1]["headers"]
-
-    def test_drop_after_last_byte_recovers_via_416(self) -> None:
-        """A connection drop after the final body byte leaves a complete file;
-        the retry's 416 answer plus the length check turn it into success
-        instead of a wasted refetch."""
-        first = _interrupted_response(b"1234", etag='"v1"')
-        first.headers = {**first.headers, "content-length": "4"}
-        r416 = _mock_response(b"", ok=False)
-        r416.status_code = 416
-        buf = io.BytesIO()
-        with patch("requests.get", side_effect=[first, r416]) as mock_get:
-            url = download_from_mirrors(["https://mirror1.com/f"], {}, buf)
-        assert url == "https://mirror1.com/f"
-        assert buf.getvalue() == b"1234"
-        assert mock_get.call_count == 2
-
-    def test_mirror_drop_without_length_restarts(self) -> None:
-        """With no content-length there is no way to prove a stitched file
-        complete, so the retry restarts even though a validator exists."""
-        buf = io.BytesIO()
-        with patch(
-            "requests.get",
-            side_effect=[
-                _interrupted_response(b"1234", etag='"v1"'),
-                _mock_response(b"full"),
-            ],
-        ) as mock_get:
-            download_from_mirrors(["https://mirror1.com/f"], {}, buf)
-        assert buf.getvalue() == b"full"
-        assert "Range" not in mock_get.call_args_list[1][1]["headers"]
-
     def test_path_target_resumes_across_runs(self, tmp_path: Path) -> None:
         """A path target routes through download_with_resume: a part file and
         metadata from a previous run resume instead of restarting."""
@@ -1573,32 +1533,14 @@ class TestDownloadFromMirrors:
         assert url == "https://mirror2.com/f"
         assert dest.read_bytes() == b"data"
 
-    def test_resumed_short_body_fails_length_check(self) -> None:
-        """A stitched file whose final length disagrees with the advertised
-        total is rejected instead of reported as success."""
-        first = _interrupted_response(b"1234", etag='"v1"')
-        first.headers = {**first.headers, "content-length": "8"}
-        # the resume ends early (5 of 8 bytes); the poisoned part is then
-        # discarded and the fresh retry also delivers a short body
-        short_resume = _resumed_response(b"5")
-        short_fresh = _mock_response(b"56")
-        short_fresh.headers = {**short_fresh.headers, "content-length": "8"}
-        buf = io.BytesIO()
-        with (
-            patch("requests.get", side_effect=[first, short_resume, short_fresh]),
-            pytest.raises(EsphomeError, match="all mirrors"),
-        ):
-            download_from_mirrors(["https://mirror1.com/f"], {}, buf)
-
-    def test_failed_mirror_leftovers_not_kept_for_next_mirror(self) -> None:
-        """Bytes from a mirror that failed all attempts must not leak into the
-        next mirror's download (no bogus Range request, fresh content)."""
-        exhausted = [_interrupted_response(b"AAAA", etag='"a1"')]
-        for _ in range(2):
-            r = _interrupted_response(b"BB")
-            r.status_code = 206
-            exhausted.append(r)
-        buf = io.BytesIO()
+    def test_failed_mirror_leftovers_not_resumed_on_next_mirror(
+        self, tmp_path: Path
+    ) -> None:
+        """A part file left by a mirror that failed all attempts must not be
+        stitched onto the next mirror's download (its meta names the other
+        URL, so the retry restarts from zero without a Range request)."""
+        exhausted = [_interrupted_response(b"AAAA") for _ in range(3)]
+        target = tmp_path / "f.bin"
         with patch(
             "requests.get",
             side_effect=exhausted + [_mock_response(b"clean")],
@@ -1606,15 +1548,17 @@ class TestDownloadFromMirrors:
             url = download_from_mirrors(
                 ["https://mirror1.com/f", "https://mirror2.com/f"],
                 {},
-                buf,
+                target,
             )
         assert url == "https://mirror2.com/f"
-        assert buf.getvalue() == b"clean"
+        assert target.read_bytes() == b"clean"
         # the second mirror starts fresh, without a Range header
         assert mock_get.call_args_list[3][0][0] == "https://mirror2.com/f"
         assert "Range" not in mock_get.call_args_list[3][1]["headers"]
 
-    def test_all_mirrors_fail_raises_error_listing_every_attempt(self) -> None:
+    def test_all_mirrors_fail_raises_error_listing_every_attempt(
+        self, tmp_path: Path
+    ) -> None:
         with (
             patch(
                 "requests.get",
@@ -1625,7 +1569,7 @@ class TestDownloadFromMirrors:
             download_from_mirrors(
                 ["https://mirror1.com/f", "https://mirror2.com/f"],
                 {},
-                io.BytesIO(),
+                tmp_path / "out.bin",
             )
         # Every attempted URL appears in the message, and the first mirror's
         # exception (the primary URL, usually the one that matters) is chained.
@@ -1641,16 +1585,6 @@ class TestDownloadFromMirrors:
         with pytest.raises(TypeError, match="target must be"):
             download_from_mirrors(["https://example.com/f"], {}, 42)  # type: ignore[arg-type]
 
-    def test_file_like_target_written(self) -> None:
-        buf = io.BytesIO()
-        with patch(
-            "requests.get",
-            return_value=_mock_response(b"bytes"),
-        ):
-            download_from_mirrors(["https://example.com/f"], {}, buf)
-        buf.seek(0)
-        assert buf.read() == b"bytes"
-
     def test_progress_bar_shown_when_content_length_known(self, tmp_path: Path) -> None:
         r = _mock_response(b"1234567890")
         r.headers = {"content-length": "10"}
@@ -1676,13 +1610,10 @@ class TestDownloadFromMirrors:
         assert target.exists()
         assert target.read_bytes() == b""
 
-    @pytest.mark.parametrize("target_kind", ["path", "file-like"])
-    def test_transient_failure_retries_mirror_sweep(
-        self, tmp_path: Path, target_kind: str
-    ) -> None:
+    def test_transient_failure_retries_mirror_sweep(self, tmp_path: Path) -> None:
         """A transient connect error on the only applicable mirror retries the
         whole mirror list with backoff instead of failing the build."""
-        target = tmp_path / "idf.tar.xz" if target_kind == "path" else io.BytesIO()
+        target = tmp_path / "idf.tar.xz"
         with (
             patch(
                 "requests.get",
@@ -1695,33 +1626,10 @@ class TestDownloadFromMirrors:
         ):
             url = download_from_mirrors(["https://mirror1.com/f"], {}, target)
         assert url == "https://mirror1.com/f"
-        data = target.read_bytes() if target_kind == "path" else target.getvalue()
-        assert data == b"data"
+        assert target.read_bytes() == b"data"
         assert mock_get.call_count == 2
         mock_sleep.assert_called_once_with(2)
 
-    def test_backoff_tick_reports_filelike_bytes(self) -> None:
-        """For a file-like target the backoff tick carries f.tell(), so the
-        combined bar holds steady through the sweep retry."""
-        target = io.BytesIO()
-        ticks: list[int] = []
-        with (
-            patch(
-                "requests.get",
-                side_effect=[
-                    req.ConnectionError("down"),
-                    _mock_response(b"data"),
-                ],
-            ),
-            patch("esphome.framework_helpers._cancellable_sleep") as mock_sleep,
-        ):
-            download_from_mirrors(
-                ["https://mirror1.com/f"], {}, target, progress=ticks.append
-            )
-        # No bytes had streamed at backoff time, so the tick carries 0
-        assert mock_sleep.call_args == call(2, ticks.append, 0)
-        assert target.getvalue() == b"data"
-
     def test_backoff_tick_reports_partial_bytes(self, tmp_path: Path) -> None:
         """The backoff tick carries the bytes already in the part file, so a
         combined bar holds steady instead of rewinding to zero."""
@@ -1831,41 +1739,83 @@ class TestDownloadFromMirrors:
         assert isinstance(ei.value.__cause__, req.ConnectionError)
         mock_sleep.assert_called_once_with(2)
 
-    def test_exhausted_mid_stream_attempts_not_swept(self) -> None:
-        """A file-like mirror that spent all its mid-stream attempts is not
-        retried again at the sweep level (unlike a path target, it has no
-        part file to resume from on a later sweep)."""
-        buf = io.BytesIO()
+    def test_exhausted_mid_stream_attempts_not_swept(self, tmp_path: Path) -> None:
+        """A mirror that spent all its mid-stream attempts fails permanently
+        instead of re-arming the sweep, and its part file survives so the
+        next esphome run resumes it."""
         with (
             patch(
                 "requests.get",
                 side_effect=[_interrupted_response(b"1234") for _ in range(3)],
             ) as mock_get,
             patch("esphome.framework_helpers.time.sleep") as mock_sleep,
-            pytest.raises(EsphomeError, match="failed after 3 attempts"),
+            pytest.raises(EsphomeError, match="after 3 attempts"),
         ):
-            download_from_mirrors(["https://mirror1.com/f"], {}, buf)
+            download_from_mirrors(["https://mirror1.com/f"], {}, tmp_path / "out.bin")
         assert mock_get.call_count == 3
         mock_sleep.assert_not_called()
+        assert (tmp_path / "out.bin.part").exists()
+
+
+class TestDownloadAndExtract:
+    def test_downloads_extracts_and_deletes_archive(self, tmp_path: Path) -> None:
+        content = gzip.compress(
+            _make_tar([_reg("file.txt")], {"file.txt": b"data"}).getvalue()
+        )
+        dest = tmp_path / "out"
+        with patch("requests.get", return_value=_mock_response(content)):
+            url = download_and_extract(
+                ["https://example.com/lib.tar.gz"],
+                {},
+                tmp_path / "lib.archive",
+                dest,
+            )
+        assert url == "https://example.com/lib.tar.gz"
+        assert (dest / "file.txt").read_bytes() == b"data"
+        # the archive is consumed; only the extraction remains
+        assert not (tmp_path / "lib.archive").exists()
+
+    def test_locked_archive_does_not_mask_result(self, tmp_path: Path) -> None:
+        """A cleanup unlink blocked by e.g. an AV handle (Windows) must not
+        replace the extraction result; the archive simply survives."""
+        content = gzip.compress(
+            _make_tar([_reg("file.txt")], {"file.txt": b"data"}).getvalue()
+        )
+        real_unlink = Path.unlink
+
+        def locked_unlink(self: Path, missing_ok: bool = False) -> None:
+            if self.name.endswith(".archive"):
+                raise PermissionError("held by antivirus")
+            real_unlink(self, missing_ok=missing_ok)
 
-    def test_mid_stream_drop_then_connect_error_not_swept(self) -> None:
-        """A connect error on a later attempt (after a mid-stream drop spent
-        one) also counts as spent budget and does not re-arm the sweep."""
-        buf = io.BytesIO()
         with (
-            patch(
-                "requests.get",
-                side_effect=[
-                    _interrupted_response(b"1234"),
-                    req.ConnectionError("down"),
-                ],
-            ) as mock_get,
-            patch("esphome.framework_helpers.time.sleep") as mock_sleep,
-            pytest.raises(EsphomeError, match="failed after 2 attempts"),
+            patch("requests.get", return_value=_mock_response(content)),
+            patch("pathlib.Path.unlink", locked_unlink),
         ):
-            download_from_mirrors(["https://mirror1.com/f"], {}, buf)
-        assert mock_get.call_count == 2
-        mock_sleep.assert_not_called()
+            url = download_and_extract(
+                ["https://example.com/lib.tar.gz"],
+                {},
+                tmp_path / "lib.archive",
+                tmp_path / "out",
+            )
+        assert url == "https://example.com/lib.tar.gz"
+        assert (tmp_path / "out" / "file.txt").read_bytes() == b"data"
+        assert (tmp_path / "lib.archive").exists()  # left behind, harmless
+
+    def test_corrupt_archive_deleted_on_extract_failure(self, tmp_path: Path) -> None:
+        """A complete-but-corrupt archive must not survive to poison the next
+        run; without a checksum only a failed extraction can expose it."""
+        with (
+            patch("requests.get", return_value=_mock_response(b"not an archive")),
+            pytest.raises(ValueError, match="Unsupported archive format"),
+        ):
+            download_and_extract(
+                ["https://example.com/lib.tar.gz"],
+                {},
+                tmp_path / "lib.archive",
+                tmp_path / "out",
+            )
+        assert not (tmp_path / "lib.archive").exists()
 
 
 def test_importing_framework_helpers_does_not_import_requests() -> None:
diff --git a/tests/unit_tests/test_happy_eyeballs.py b/tests/unit_tests/test_happy_eyeballs.py
index 3335a8a3e3..ccb7aa3c67 100644
--- a/tests/unit_tests/test_happy_eyeballs.py
+++ b/tests/unit_tests/test_happy_eyeballs.py
@@ -4,7 +4,9 @@ from __future__ import annotations
 
 import asyncio
 from collections.abc import Generator
+from concurrent.futures import ThreadPoolExecutor
 import socket
+import threading
 from typing import Any
 from unittest.mock import Mock, patch
 
@@ -61,6 +63,41 @@ def test_ensure_happy_eyeballs_patches_and_is_idempotent(
     assert urllib3.util.connection.create_connection is patched
 
 
+def test_ensure_happy_eyeballs_concurrent_first_calls_patch_once(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Worker threads fanning out (download_content_many, run_batch_downloads)
+    may race the first call; the replacement is built exactly once."""
+    import urllib3.util.connection
+
+    from esphome import happy_eyeballs
+
+    def stock(*args: Any, **kwargs: Any) -> None:
+        pass
+
+    monkeypatch.setattr(urllib3.util.connection, "create_connection", stock)
+
+    barrier = threading.Barrier(8)
+    builds: list[int] = []
+    real_make = happy_eyeballs._make_create_connection
+
+    def counting_make() -> Any:
+        builds.append(1)
+        return real_make()
+
+    monkeypatch.setattr(happy_eyeballs, "_make_create_connection", counting_make)
+
+    def racer() -> None:
+        barrier.wait(timeout=10)
+        ensure_happy_eyeballs()
+
+    with ThreadPoolExecutor(max_workers=8) as ex:
+        list(ex.map(lambda _: racer(), range(8)))
+
+    assert builds == [1]
+    assert urllib3.util.connection.create_connection._esphome_patched
+
+
 def test_connects_and_restores_socket_state(
     create_connection: Any, listener: tuple[str, int], mock_gai: Any
 ) -> None:
diff --git a/tests/unit_tests/test_net_retry.py b/tests/unit_tests/test_net_retry.py
index c22bda5ee5..6c2a1ee05f 100644
--- a/tests/unit_tests/test_net_retry.py
+++ b/tests/unit_tests/test_net_retry.py
@@ -7,7 +7,11 @@ import pytest
 import requests as req
 
 from esphome.core import EsphomeError
-from esphome.net_retry import fetch_with_retry, is_transient_download_error
+from esphome.net_retry import (
+    fetch_with_retry,
+    http_request,
+    is_transient_download_error,
+)
 
 
 def _http_error(status: int) -> req.HTTPError:
@@ -141,3 +145,40 @@ class TestFetchWithRetry:
         assert mock_sleep.call_args_list == [call(2), call(4)]
         assert "(attempt 2/3)" in caplog.text
         assert "(attempt 3/3)" in caplog.text
+
+
+class TestHttpRequest:
+    def test_applies_happy_eyeballs_and_forwards_arguments(self) -> None:
+        with (
+            patch("esphome.net_retry.ensure_happy_eyeballs") as mock_he,
+            patch("requests.get", return_value=MagicMock()) as mock_get,
+        ):
+            resp = http_request(
+                "GET",
+                "https://example.com/f",
+                timeout=30,
+                stream=True,
+                headers={"Range": "bytes=4-"},
+            )
+        mock_he.assert_called_once_with()
+        assert resp is mock_get.return_value
+        assert mock_get.call_args == call(
+            "https://example.com/f",
+            timeout=30,
+            stream=True,
+            headers={"Range": "bytes=4-"},
+            allow_redirects=True,
+        )
+
+    def test_dispatches_head_through_requests_head(self) -> None:
+        """Dispatch goes through requests.get/head so tests patching those
+        entry points keep working."""
+        with patch("requests.head", return_value=MagicMock()) as mock_head:
+            http_request("HEAD", "https://example.com/f", timeout=(5, 30))
+        assert mock_head.call_args[1]["timeout"] == (5, 30)
+
+    def test_no_status_handling(self) -> None:
+        """Error statuses are the caller's problem; nothing raises here."""
+        resp = MagicMock(status_code=404)
+        with patch("requests.get", return_value=resp):
+            assert http_request("GET", "https://example.com/f", timeout=1) is resp
diff --git a/tests/unit_tests/test_nrf52_framework.py b/tests/unit_tests/test_nrf52_framework.py
index 65b73e37fe..b78a94a2e7 100644
--- a/tests/unit_tests/test_nrf52_framework.py
+++ b/tests/unit_tests/test_nrf52_framework.py
@@ -107,11 +107,13 @@ def mock_nrf52_ops():
         patch(
             "esphome.components.nrf52.framework.run_command_ok", return_value=True
         ) as mock_run_cmd,
+        # download_and_extract resolves its internals in framework_helpers,
+        # so the download/extract seams are patched there.
         patch(
-            "esphome.components.nrf52.framework.download_from_mirrors",
+            "esphome.framework_helpers.download_from_mirrors",
             return_value="https://example.com/tc.tar.xz",
         ) as mock_download,
-        patch("esphome.components.nrf52.framework.archive_extract_all") as mock_extract,
+        patch("esphome.framework_helpers.archive_extract_all") as mock_extract,
     ):
         yield SimpleNamespace(
             rmdir=mock_rmdir,
diff --git a/tests/unit_tests/test_platformio_library.py b/tests/unit_tests/test_platformio_library.py
index ef24f99953..0c873dc3fe 100644
--- a/tests/unit_tests/test_platformio_library.py
+++ b/tests/unit_tests/test_platformio_library.py
@@ -158,16 +158,12 @@ def test_urlsource_download_extracts_then_reuses_marker(
 ):
     monkeypatch.setattr(lib, "rmdir", lambda path, msg="": None)
     dl_calls: list[list[str]] = []
-    monkeypatch.setattr(
-        lib,
-        "download_from_mirrors",
-        lambda urls, headers, f, progress=None: dl_calls.append(urls),
-    )
 
-    def fake_extract(fileobj, path):
-        Path(path).mkdir(parents=True, exist_ok=True)
+    def fake_download_and_extract(urls, subs, archive_path, extract_dir, **kwargs):
+        dl_calls.append(urls)
+        Path(extract_dir).mkdir(parents=True, exist_ok=True)
 
-    monkeypatch.setattr(lib, "archive_extract_all", fake_extract)
+    monkeypatch.setattr(lib, "download_and_extract", fake_download_and_extract)
 
     src = URLSource("http://example.test/lib.tar.gz")
     out = src.download("mylib")
@@ -187,6 +183,25 @@ def test_urlsource_download_extracts_then_reuses_marker(
     assert "Downloading" not in caplog.text
 
 
+def test_urlsource_downloads_to_sibling_archive_path(setup_core, monkeypatch):
+    """The archive downloads to a deterministic path next to the cache dir
+    (not a random temp file), so an interrupted download's .part file
+    resumes on the next run."""
+    monkeypatch.setattr(lib, "rmdir", lambda path, msg="": None)
+    targets: list[Path] = []
+
+    def fake_download_and_extract(urls, subs, archive_path, extract_dir, **kwargs):
+        targets.append(Path(archive_path))
+        Path(extract_dir).mkdir(parents=True, exist_ok=True)
+
+    monkeypatch.setattr(lib, "download_and_extract", fake_download_and_extract)
+
+    src = URLSource("http://example.test/lib.tar.gz")
+    out = src.download("mylib")
+
+    assert targets == [out.with_name(f"{out.name}.archive")]
+
+
 def test_resolve_registry_version_raises_without_pkg_file(monkeypatch):
     registry = lib._make_registry_client()
     monkeypatch.setattr(