From 58a42fe5c27bdb2158d8c444d544f473fd0e3b0a Mon Sep 17 00:00:00 2001
From: "J. Nick Koston" <nick@koston.org>
Date: Wed, 12 Aug 2026 01:44:35 -0500
Subject: [PATCH] [core] Batch remote file downloads during config validation
 (#18069)

---
 esphome/components/animation/__init__.py      |   5 +
 esphome/components/animation/image.py         |   5 +
 esphome/components/bme68x_bsec2/__init__.py   |  64 +++-
 .../components/bme68x_bsec2_i2c/__init__.py   |   7 +-
 esphome/components/esp32/__init__.py          |  50 ++-
 esphome/components/file/image.py              |  81 ++--
 esphome/components/font/__init__.py           | 246 +++++++++---
 esphome/components/gsl3670/touchscreen.py     |  22 +-
 .../components/micro_wake_word/__init__.py    |  16 +-
 esphome/components/shelly_dimmer/light.py     | 108 ++++--
 esphome/config.py                             | 128 ++++++-
 esphome/external_files.py                     | 224 +++++++++--
 esphome/loader.py                             |  18 +-
 tests/component_tests/gsl3670/test_init.py    |  22 +-
 .../components/bme68x_bsec2/__init__.py       |   0
 .../components/bme68x_bsec2/test_init.py      |  64 ++++
 tests/unit_tests/components/file/__init__.py  |   0
 .../unit_tests/components/file/test_image.py  |  75 ++++
 tests/unit_tests/components/font/__init__.py  |   0
 tests/unit_tests/components/font/test_init.py | 229 +++++++++++
 .../unit_tests/components/gsl3670/__init__.py |   0
 .../components/gsl3670/test_touchscreen.py    |  35 ++
 .../components/micro_wake_word/test_init.py   |   9 +-
 .../components/shelly_dimmer/__init__.py      |   0
 .../components/shelly_dimmer/test_light.py    | 154 ++++++++
 tests/unit_tests/test_config_prefetch.py      | 355 ++++++++++++++++++
 tests/unit_tests/test_external_files.py       | 341 +++++++++++++++--
 27 files changed, 2005 insertions(+), 253 deletions(-)
 create mode 100644 tests/unit_tests/components/bme68x_bsec2/__init__.py
 create mode 100644 tests/unit_tests/components/bme68x_bsec2/test_init.py
 create mode 100644 tests/unit_tests/components/file/__init__.py
 create mode 100644 tests/unit_tests/components/file/test_image.py
 create mode 100644 tests/unit_tests/components/font/__init__.py
 create mode 100644 tests/unit_tests/components/font/test_init.py
 create mode 100644 tests/unit_tests/components/gsl3670/__init__.py
 create mode 100644 tests/unit_tests/components/gsl3670/test_touchscreen.py
 create mode 100644 tests/unit_tests/components/shelly_dimmer/__init__.py
 create mode 100644 tests/unit_tests/components/shelly_dimmer/test_light.py
 create mode 100644 tests/unit_tests/test_config_prefetch.py

diff --git a/esphome/components/animation/__init__.py b/esphome/components/animation/__init__.py
index 0df7c56313..6da5268432 100644
--- a/esphome/components/animation/__init__.py
+++ b/esphome/components/animation/__init__.py
@@ -13,8 +13,13 @@
 import esphome.components.image as espImage
 import esphome.config_validation as cv
 
+from . import image as animation_image
 from .image import ANIMATION_CONFIG_SCHEMA, setup_animation
 
+# The deprecated top-level `animation:` shim gets the same batched
+# downloads as the `image:` platform form.
+PREFETCH_FILES = animation_image.PREFETCH_FILES
+
 AUTO_LOAD = ["image", "file"]
 CODEOWNERS = ["@syndlex"]
 DEPENDENCIES = ["display"]
diff --git a/esphome/components/animation/image.py b/esphome/components/animation/image.py
index 95875fe2b0..73d428bd20 100644
--- a/esphome/components/animation/image.py
+++ b/esphome/components/animation/image.py
@@ -1,6 +1,7 @@
 from esphome import automation
 import esphome.codegen as cg
 from esphome.components.const import CONF_LOOP
+from esphome.components.file import image as file_image
 from esphome.components.file.image import image_schema, write_image
 from esphome.components.image import Image_, validate_settings
 import esphome.config_validation as cv
@@ -8,6 +9,10 @@ from esphome.const import CONF_ID, CONF_REPEAT
 from esphome.types import ConfigType
 
 CODEOWNERS = ["@syndlex"]
+
+# The animation platform shares the file platform's remote file handling,
+# including its batch-download hook.
+PREFETCH_FILES = file_image.PREFETCH_FILES
 AUTO_LOAD = ["file"]
 DEPENDENCIES = ["display"]
 
diff --git a/esphome/components/bme68x_bsec2/__init__.py b/esphome/components/bme68x_bsec2/__init__.py
index 63f63c5da2..c12eb39d2d 100644
--- a/esphome/components/bme68x_bsec2/__init__.py
+++ b/esphome/components/bme68x_bsec2/__init__.py
@@ -1,4 +1,3 @@
-import hashlib
 from pathlib import Path
 
 from esphome import core, external_files
@@ -12,6 +11,8 @@ from esphome.const import (
     CONF_SAMPLE_RATE,
     CONF_TEMPERATURE_OFFSET,
 )
+from esphome.external_files import RemoteFile
+from esphome.types import ConfigType
 
 CODEOWNERS = ["@neffs", "@kbx81"]
 CONFLICTS_WITH = ["bme680_bsec"]
@@ -74,11 +75,7 @@ VOLTAGE_FILE_NAME = {
 
 
 def _compute_local_file_path(url: str) -> Path:
-    h = hashlib.new("sha256")
-    h.update(url.encode())
-    key = h.hexdigest()[:8]
-    base_dir = external_files.compute_local_file_dir(DOMAIN)
-    return base_dir / key
+    return external_files.compute_local_file_path(DOMAIN, url)
 
 
 def _compute_url(config: dict) -> str:
@@ -105,6 +102,42 @@ def download_bme68x_blob(config):
     return config
 
 
+# Shared by the schema and the prefetch hook so they cannot drift.
+_MODEL_VALIDATOR = cv.one_of(*MODEL_OPTIONS, lower=True)
+_ALGORITHM_OUTPUT_VALIDATOR = cv.enum(ALGORITHM_OUTPUT_OPTIONS, lower=True)
+# Key -> (validator, default) for the defaulted options that select the blob.
+_BLOB_OPTIONS = {
+    CONF_OPERATING_AGE: (cv.enum(OPERATING_AGE_OPTIONS, lower=True), "28d"),
+    CONF_SAMPLE_RATE: (cv.enum(SAMPLE_RATE_OPTIONS, upper=True), "LP"),
+    CONF_SUPPLY_VOLTAGE: (cv.enum(VOLTAGE_OPTIONS, upper=True), "3.3V"),
+}
+
+
+def _extract_blob_ref(entry: ConfigType) -> RemoteFile | None:
+    """Raw entry to its BSEC2 blob; None when a value is unrecognized.
+
+    Applies the schema defaults and validators read-only; skipped entries
+    are left to the schema validator.
+    """
+    try:
+        spec = {
+            key: validator(str(entry.get(key, default)))  # pylint: disable=not-callable
+            for key, (validator, default) in _BLOB_OPTIONS.items()
+        }
+        spec[CONF_MODEL] = _MODEL_VALIDATOR(str(entry.get(CONF_MODEL, "")))
+        if (algorithm_output := entry.get(CONF_ALGORITHM_OUTPUT)) is not None:
+            spec[CONF_ALGORITHM_OUTPUT] = _ALGORITHM_OUTPUT_VALIDATOR(
+                str(algorithm_output)
+            )
+    except cv.Invalid:
+        return None
+    url = _compute_url(spec)
+    return RemoteFile(url, _compute_local_file_path(url))
+
+
+PREFETCH_FILES = external_files.single_stage_prefetch(_extract_blob_ref)
+
+
 def validate_bme68x(config):
     if CONF_ALGORITHM_OUTPUT not in config:
         return config
@@ -128,19 +161,12 @@ CONFIG_SCHEMA_BASE = (
         {
             cv.GenerateID(): cv.declare_id(BME68xBSEC2Component),
             cv.GenerateID(CONF_RAW_DATA_ID): cv.declare_id(cg.uint8),
-            cv.Required(CONF_MODEL): cv.one_of(*MODEL_OPTIONS, lower=True),
-            cv.Optional(CONF_ALGORITHM_OUTPUT): cv.enum(
-                ALGORITHM_OUTPUT_OPTIONS, lower=True
-            ),
-            cv.Optional(CONF_OPERATING_AGE, default="28d"): cv.enum(
-                OPERATING_AGE_OPTIONS, lower=True
-            ),
-            cv.Optional(CONF_SAMPLE_RATE, default="LP"): cv.enum(
-                SAMPLE_RATE_OPTIONS, upper=True
-            ),
-            cv.Optional(CONF_SUPPLY_VOLTAGE, default="3.3V"): cv.enum(
-                VOLTAGE_OPTIONS, upper=True
-            ),
+            cv.Required(CONF_MODEL): _MODEL_VALIDATOR,
+            cv.Optional(CONF_ALGORITHM_OUTPUT): _ALGORITHM_OUTPUT_VALIDATOR,
+            **{
+                cv.Optional(key, default=default): validator
+                for key, (validator, default) in _BLOB_OPTIONS.items()
+            },
             cv.Optional(CONF_TEMPERATURE_OFFSET, default=0): cv.temperature_delta,
             cv.Optional(
                 CONF_STATE_SAVE_INTERVAL, default="6hours"
diff --git a/esphome/components/bme68x_bsec2_i2c/__init__.py b/esphome/components/bme68x_bsec2_i2c/__init__.py
index c8ca0ba022..dacd4e32ad 100644
--- a/esphome/components/bme68x_bsec2_i2c/__init__.py
+++ b/esphome/components/bme68x_bsec2_i2c/__init__.py
@@ -1,5 +1,5 @@
 import esphome.codegen as cg
-from esphome.components import i2c
+from esphome.components import bme68x_bsec2, i2c
 from esphome.components.bme68x_bsec2 import (
     CONFIG_SCHEMA_BASE,
     BME68xBSEC2Component,
@@ -13,6 +13,11 @@ AUTO_LOAD = ["bme68x_bsec2"]
 DEPENDENCIES = ["i2c"]
 MULTI_CONF = True
 
+# The user-facing domain is this module (the base component only appears
+# via AUTO_LOAD), so the batch-download hook must be re-exported here to
+# take effect.
+PREFETCH_FILES = bme68x_bsec2.PREFETCH_FILES
+
 bme68x_bsec2_i2c_ns = cg.esphome_ns.namespace("bme68x_bsec2_i2c")
 BME68xBSEC2I2CComponent = bme68x_bsec2_i2c_ns.class_(
     "BME68xBSEC2I2CComponent", BME68xBSEC2Component, i2c.I2CDevice
diff --git a/esphome/components/esp32/__init__.py b/esphome/components/esp32/__init__.py
index 2e72c78974..ada6d25db5 100644
--- a/esphome/components/esp32/__init__.py
+++ b/esphome/components/esp32/__init__.py
@@ -3280,27 +3280,45 @@ def copy_files():
         __version__,
     )
 
+    # Remote extra build files are fetched into the shared download cache in
+    # one parallel batch (conditional requests skip unchanged files), then
+    # copied into the build tree like their local counterparts.
+    sources: dict[str, Path] = {}
+    remote: list[tuple[str, str]] = []
     for file in CORE.data[KEY_ESP32][KEY_EXTRA_BUILD_FILES].values():
         name: str = file[KEY_NAME]
         path: Path = file[KEY_PATH]
         if str(path).startswith("http"):
-            import requests
-
-            from esphome.happy_eyeballs import ensure_happy_eyeballs
-
-            ensure_happy_eyeballs()
-
-            try:
-                req = requests.get(path, timeout=30)
-                req.raise_for_status()
-            except requests.exceptions.RequestException as e:
-                raise EsphomeError(
-                    f"Could not download extra build file {path}: {e}"
-                ) from e
-            CORE.relative_build_path(name).parent.mkdir(parents=True, exist_ok=True)
-            CORE.relative_build_path(name).write_bytes(req.content)
+            remote.append((name, str(path)))
         else:
-            copy_file_if_changed(path, CORE.relative_build_path(name))
+            sources[name] = path
+    if remote:
+        # Imported lazily: requests (via external_files) is a heavy import
+        # and remote extra build files are rare.
+        from esphome import external_files
+
+        downloads: list[external_files.RemoteFile] = []
+        for name, url in remote:
+            cache_path = external_files.compute_local_file_path(KEY_ESP32, url)
+            # Unverifiable bytes: an unrevalidated copy is an error, matching
+            # the old always-download behavior on network failure.
+            downloads.append(
+                external_files.RemoteFile(url, cache_path, allow_stale=False)
+            )
+            sources[name] = cache_path
+        try:
+            external_files.download_content_many(
+                downloads, description="extra build file(s)"
+            )
+        except cv.MultipleInvalid as e:
+            details = "; ".join(str(err) for err in e.errors)
+            raise EsphomeError(
+                f"Could not download extra build file(s): {details}"
+            ) from e
+        except cv.Invalid as e:
+            raise EsphomeError(f"Could not download extra build file(s): {e}") from e
+    for name, source in sources.items():
+        copy_file_if_changed(source, CORE.relative_build_path(name))
 
 
 def _decode_pc(config, addr):
diff --git a/esphome/components/file/image.py b/esphome/components/file/image.py
index 9a7c762a79..b54c3f2adf 100644
--- a/esphome/components/file/image.py
+++ b/esphome/components/file/image.py
@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 import contextlib
-import hashlib
 import io
 import logging
 from pathlib import Path
@@ -43,15 +42,13 @@ from esphome.const import (
 )
 from esphome.core import CORE, HexInt
 from esphome.cpp_generator import MockObj, MockObjClass
+from esphome.external_files import RemoteFile
 from esphome.types import ConfigType
 
 CODEOWNERS = ["@esphome/core"]
 
 _LOGGER = logging.getLogger(__name__)
 
-# If the MDI file cannot be downloaded within this time, abort.
-IMAGE_DOWNLOAD_TIMEOUT = 30  # seconds
-
 SOURCE_LOCAL = "local"
 SOURCE_WEB = "web"
 
@@ -65,16 +62,16 @@ MDI_SOURCES = {
     SOURCE_MEMORY: "https://raw.githubusercontent.com/Pictogrammers/Memory/refs/heads/main/src/svg/",
 }
 
+# Shared by the schema validator and the prefetch extractor so they cannot
+# drift.
+_MDI_ICON_RE = re.compile(r"^[a-zA-Z0-9\-]+$")
 
-def compute_local_image_path(value) -> Path:
+
+def compute_local_image_path(value: str | ConfigType) -> Path:
     url = value[CONF_URL] if isinstance(value, dict) else value
-    h = hashlib.new("sha256")
-    h.update(url.encode())
-    key = h.hexdigest()[:8]
     # Downloaded files are cached under the shared `image` domain directory so
     # the cache location is unaffected by which platform requested the file.
-    base_dir = external_files.compute_local_file_dir(DOMAIN)
-    return base_dir / key
+    return external_files.compute_local_file_path(DOMAIN, url)
 
 
 def local_path(value):
@@ -83,16 +80,20 @@ def local_path(value):
 
 
 def download_file(url, path):
-    external_files.download_content(url, path, IMAGE_DOWNLOAD_TIMEOUT)
+    # The shared NETWORK_TIMEOUT applies; a per-caller timeout would be
+    # silently ignored on a per-run memo hit anyway (memos key by path).
+    external_files.download_content(url, path)
     return str(path)
 
 
-def download_gh_svg(value, source):
-    mdi_id = value[CONF_ICON] if isinstance(value, dict) else value
+def _gh_svg_url_path(mdi_id: str, source: str) -> tuple[str, Path]:
     base_dir = external_files.compute_local_file_dir(DOMAIN) / source
-    path = base_dir / f"{mdi_id}.svg"
+    return MDI_SOURCES[source] + mdi_id + ".svg", base_dir / f"{mdi_id}.svg"
 
-    url = MDI_SOURCES[source] + mdi_id + ".svg"
+
+def download_gh_svg(value: str | ConfigType, source: str) -> str:
+    mdi_id = value[CONF_ICON] if isinstance(value, dict) else value
+    url, path = _gh_svg_url_path(mdi_id, source)
     return download_file(url, path)
 
 
@@ -101,17 +102,53 @@ def download_image(value):
     return download_file(value, compute_local_image_path(value))
 
 
-def validate_file_shorthand(value):
-    value = cv.string_strict(value)
+def _parse_remote_shorthand(value: str) -> RemoteFile | None:
+    """Parse a string `file:` shorthand to its remote file; None if local.
+
+    Raises cv.Invalid for a malformed icon name. Shared by the schema
+    validator and the prefetch extractor so they cannot drift.
+    """
     parts = value.strip().split(":")
     if len(parts) == 2 and parts[0] in MDI_SOURCES:
-        match = re.match(r"^[a-zA-Z0-9\-]+$", parts[1])
-        if match is None:
+        if _MDI_ICON_RE.match(parts[1]) is None:
             raise cv.Invalid(f"Could not parse mdi icon name from '{value}'.")
-        return download_gh_svg(parts[1], parts[0])
-
+        return RemoteFile(*_gh_svg_url_path(parts[1], parts[0]))
     if value.startswith(("http://", "https://")):
-        return download_image(value)
+        return RemoteFile(value, compute_local_image_path(value))
+    return None
+
+
+def _extract_file_ref(value: object) -> RemoteFile | None:
+    """Map a raw, pre-schema `file:` value to its remote file.
+
+    Returns None for local files and anything it does not recognize; the
+    schema validators stay authoritative.
+    """
+    if isinstance(value, str):
+        try:
+            return _parse_remote_shorthand(value)
+        except cv.Invalid:
+            return None
+    if isinstance(value, dict):
+        source = value.get(CONF_SOURCE)
+        if source == SOURCE_WEB and isinstance(url := value.get(CONF_URL), str):
+            return RemoteFile(url, compute_local_image_path(url))
+        if source in MDI_SOURCES and isinstance(icon := value.get(CONF_ICON), str):
+            return RemoteFile(*_gh_svg_url_path(icon, source))
+    return None
+
+
+def _extract_entry_ref(entry: ConfigType) -> RemoteFile | None:
+    return _extract_file_ref(entry.get(CONF_FILE))
+
+
+PREFETCH_FILES = external_files.single_stage_prefetch(_extract_entry_ref)
+
+
+def validate_file_shorthand(value):
+    value = cv.string_strict(value)
+    if (remote := _parse_remote_shorthand(value)) is not None:
+        return download_file(remote.url, remote.path)
 
     value = cv.file_(value)
     return local_path(value)
diff --git a/esphome/components/font/__init__.py b/esphome/components/font/__init__.py
index 5872b607f1..918fde5dbd 100644
--- a/esphome/components/font/__init__.py
+++ b/esphome/components/font/__init__.py
@@ -1,6 +1,5 @@
-from collections.abc import MutableMapping
+from collections.abc import Iterable, MutableMapping
 import functools
-import hashlib
 from itertools import accumulate
 import logging
 from pathlib import Path
@@ -17,7 +16,6 @@ from freetype import (
     FT_Exception,
     ft_pixel_mode_mono,
 )
-import requests
 
 from esphome import external_files
 import esphome.codegen as cg
@@ -36,7 +34,7 @@ from esphome.const import (
     CONF_WEIGHT,
 )
 from esphome.core import CORE, HexInt
-from esphome.happy_eyeballs import ensure_happy_eyeballs
+from esphome.external_files import RemoteFile
 from esphome.types import ConfigType
 
 _LOGGER = logging.getLogger(__name__)
@@ -296,46 +294,80 @@ def validate_weight_name(value):
     return FONT_WEIGHTS[cv.one_of(*FONT_WEIGHTS, lower=True, space="-")(value)]
 
 
-def _compute_local_font_path(value: dict) -> Path:
-    url = value[CONF_URL]
-    h = hashlib.new("sha256")
-    h.update(url.encode())
-    key = h.hexdigest()[:8]
-    base_dir = external_files.compute_local_file_dir(DOMAIN)
-    _LOGGER.debug("_compute_local_font_path: %s", base_dir / key)
-    return base_dir / key
+def _web_font_path(value: dict) -> Path:
+    return external_files.compute_local_file_path(DOMAIN, value[CONF_URL]) / "font.ttf"
 
 
-def download_gfont(value):
+def _gfonts_css_url(value: dict) -> str:
+    return (
+        f"https://fonts.googleapis.com/css2?family={value[CONF_FAMILY]}"
+        f":ital,wght@{int(value[CONF_ITALIC])},{value[CONF_WEIGHT]}"
+    )
+
+
+def _gfonts_cache_path(value: dict, suffix: str) -> Path:
+    name = f"{value[CONF_FAMILY]}@{value[CONF_WEIGHT]}@{value[CONF_ITALIC]}@v1"
+    return external_files.compute_local_file_dir(DOMAIN) / f"{name}.{suffix}"
+
+
+def _gfonts_ttf_path(value: dict) -> Path:
+    return _gfonts_cache_path(value, "ttf")
+
+
+def _gfonts_css_path(value: dict) -> Path:
+    return _gfonts_cache_path(value, "css")
+
+
+def _parse_gfonts_css(css: str) -> str | None:
+    """Extract the truetype URL from a Google Fonts CSS response."""
+    match = re.search(r"src:\s+url\((.+)\)\s+format\('truetype'\);", css)
+    return match.group(1) if match else None
+
+
+def download_gfont(value: ConfigType) -> ConfigType:
     if value in FONT_CACHE:
         return value
-    name = (
-        f"{value[CONF_FAMILY]}:ital,wght@{int(value[CONF_ITALIC])},{value[CONF_WEIGHT]}"
-    )
-    url = f"https://fonts.googleapis.com/css2?family={name}"
-    path = (
-        external_files.compute_local_file_dir(DOMAIN)
-        / f"{value[CONF_FAMILY]}@{value[CONF_WEIGHT]}@{value[CONF_ITALIC]}@v1.ttf"
-    )
+    path = _gfonts_ttf_path(value)
     if not external_files.is_file_recent(path, value[CONF_REFRESH]):
         _LOGGER.debug("download_gfont: path=%s", path)
+        url = _gfonts_css_url(value)
+        css_path = _gfonts_css_path(value)
         try:
-            ensure_happy_eyeballs()
-            req = requests.get(url, timeout=external_files.NETWORK_TIMEOUT)
-            req.raise_for_status()
-        except requests.exceptions.RequestException as e:
+            css_bytes = external_files.download_content(url, css_path)
+        except cv.Invalid as e:
             raise cv.Invalid(
                 f"Could not download font at {url}, please check the fonts exists "
                 f"at google fonts ({e})"
             ) from e
-        match = re.search(r"src:\s+url\((.+)\)\s+format\('truetype'\);", req.text)
-        if match is None:
+        if not (
+            external_files.is_fresh_this_run(css_path) or CORE.skip_external_update
+        ):
+            # Same rule as PREFETCH_FILES stage two: a CSS body that could
+            # not be revalidated may name a rotated ttf URL. Use the cached
+            # font instead (the failed check already warned).
+            if path.exists():
+                FONT_CACHE[value] = path
+                return value
             raise cv.Invalid(
-                f"Could not extract ttf file from gfonts response for {name}, "
-                f"please report this."
+                f"Could not refresh the Google Fonts CSS for "
+                f"{value[CONF_FAMILY]} and no cached font is available"
+            )
+        try:
+            css = css_bytes.decode("utf-8")
+        except UnicodeDecodeError as e:
+            # Do not leave an unusable body in the cache to be served again.
+            css_path.unlink(missing_ok=True)
+            raise cv.Invalid(
+                f"Bad response from Google Fonts for {value[CONF_FAMILY]}: "
+                f"not a text document"
+            ) from e
+        ttf_url = _parse_gfonts_css(css)
+        if ttf_url is None:
+            css_path.unlink(missing_ok=True)
+            raise cv.Invalid(
+                f"Could not extract ttf file from gfonts response for "
+                f"{value[CONF_FAMILY]}, please report this."
             )
-
-        ttf_url = match.group(1)
         _LOGGER.debug("download_gfont: ttf_url=%s", ttf_url)
 
         external_files.download_content(ttf_url, path)
@@ -346,11 +378,11 @@ def download_gfont(value):
     return value
 
 
-def download_web_font(value):
+def download_web_font(value: ConfigType) -> ConfigType:
     if value in FONT_CACHE:
         return value
     url = value[CONF_URL]
-    path = _compute_local_font_path(value) / "font.ttf"
+    path = _web_font_path(value)
 
     external_files.download_content(url, path)
     _LOGGER.debug("download_web_font: path=%s", path)
@@ -358,13 +390,18 @@ def download_web_font(value):
     return value
 
 
+# Shared by the schema and the prefetch extractor so they cannot drift.
+_DEFAULT_WEIGHT = "regular"
+_DEFAULT_ITALIC = False
+_DEFAULT_REFRESH = "1d"
+_WEIGHT_VALIDATOR = cv.Any(cv.int_, validate_weight_name)
+_REFRESH_VALIDATOR = cv.All(cv.string, cv.source_refresh)
+
 EXTERNAL_FONT_SCHEMA = cv.Schema(
     {
-        cv.Optional(CONF_WEIGHT, default="regular"): cv.Any(
-            cv.int_, validate_weight_name
-        ),
-        cv.Optional(CONF_ITALIC, default=False): cv.boolean,
-        cv.Optional(CONF_REFRESH, default="1d"): cv.All(cv.string, cv.source_refresh),
+        cv.Optional(CONF_WEIGHT, default=_DEFAULT_WEIGHT): _WEIGHT_VALIDATOR,
+        cv.Optional(CONF_ITALIC, default=_DEFAULT_ITALIC): cv.boolean,
+        cv.Optional(CONF_REFRESH, default=_DEFAULT_REFRESH): _REFRESH_VALIDATOR,
     }
 )
 
@@ -387,36 +424,123 @@ WEB_FONT_SCHEMA = cv.All(
 )
 
 
-def validate_file_shorthand(value):
-    value = cv.string_strict(value)
+_GFONTS_SHORTHAND_RE = re.compile(r"^gfonts://([^@]+)(@.+)?$")
+
+
+def _shorthand_to_file_dict(value: str) -> ConfigType | None:
+    """Typed-dict form of a remote font shorthand.
+
+    Shared by the schema validator and the prefetch extractor so the two
+    cannot drift. Returns None for values that are not remote shorthand
+    (i.e. local paths); raises cv.Invalid for a malformed gfonts shorthand.
+    """
     if value.startswith("gfonts://"):
-        match = re.match(r"^gfonts://([^@]+)(@.+)?$", value)
-        if match is None:
+        if (match := _GFONTS_SHORTHAND_RE.match(value)) is None:
             raise cv.Invalid("Could not parse gfonts shorthand syntax, please check it")
-        family = match.group(1)
-        weight = match.group(2)
-        data = {
+        data = {CONF_TYPE: TYPE_GFONTS, CONF_FAMILY: match.group(1)}
+        if match.group(2):
+            data[CONF_WEIGHT] = match.group(2)[1:]
+        return data
+    if value.startswith(("http://", "https://")):
+        return {CONF_TYPE: TYPE_WEB, CONF_URL: value}
+    return None
+
+
+def _extract_remote_font(value: object) -> ConfigType | None:
+    """Map a raw, pre-schema font `file:` value to a normalized remote spec.
+
+    Read-only mirror of `validate_file_shorthand` / `TYPED_FILE_SCHEMA` for
+    the prefetch hooks; returns None for local fonts and anything it does
+    not recognize. A wrong answer only wastes or misses a prefetch, the
+    schema validators stay authoritative.
+    """
+    if isinstance(value, str):
+        try:
+            value = _shorthand_to_file_dict(value)
+        except cv.Invalid:
+            return None
+    if not isinstance(value, dict):
+        return None
+    font_type = value.get(CONF_TYPE)
+    if font_type == TYPE_WEB and isinstance(url := value.get(CONF_URL), str):
+        return {CONF_TYPE: TYPE_WEB, CONF_URL: url}
+    if font_type == TYPE_GFONTS and isinstance(family := value.get(CONF_FAMILY), str):
+        try:
+            italic = cv.boolean(value.get(CONF_ITALIC, _DEFAULT_ITALIC))
+            weight = _WEIGHT_VALIDATOR(value.get(CONF_WEIGHT, _DEFAULT_WEIGHT))
+            refresh = _REFRESH_VALIDATOR(value.get(CONF_REFRESH, _DEFAULT_REFRESH))
+        except cv.Invalid:
+            return None
+        return {
             CONF_TYPE: TYPE_GFONTS,
             CONF_FAMILY: family,
+            CONF_WEIGHT: weight,
+            CONF_ITALIC: italic,
+            CONF_REFRESH: refresh,
         }
-        if weight is not None:
-            data[CONF_WEIGHT] = weight[1:]
-        return font_file_schema(data)
+    return None
 
-    if value.startswith(("http://", "https://")):
-        return font_file_schema(
-            {
-                CONF_TYPE: TYPE_WEB,
-                CONF_URL: value,
-            }
-        )
 
-    return font_file_schema(
-        {
-            CONF_TYPE: TYPE_LOCAL,
-            CONF_PATH: value,
-        }
-    )
+def _iter_remote_specs(entries: list[ConfigType]) -> Iterable[ConfigType]:
+    """Yield the remote spec of every `file:` value, including extras."""
+    for entry in entries:
+        values = [entry.get(CONF_FILE)]
+        extras = entry.get(CONF_EXTRAS)
+        if isinstance(extras, dict):
+            # The schema runs cv.ensure_list on extras, so a bare mapping
+            # is valid raw config; mirror that normalization here.
+            extras = [extras]
+        if isinstance(extras, list):
+            values.extend(
+                extra.get(CONF_FILE) for extra in extras if isinstance(extra, dict)
+            )
+        for value in values:
+            if (spec := _extract_remote_font(value)) is not None:
+                yield spec
+
+
+def PREFETCH_FILES(entries: list[ConfigType]) -> Iterable[list[RemoteFile]]:
+    """Batch-download hook: web fonts, then Google Fonts CSS, then ttf.
+
+    Stage one fetches web fonts and the CSS of stale gfonts; stage two
+    parses the now-cached CSS for the ttf URLs it names.
+    """
+    stage1: list[RemoteFile] = []
+    # Keyed by cache path: the same font at several sizes is one download,
+    # one freshness stat, and one stage-two CSS parse.
+    stale_gfonts: dict[Path, ConfigType] = {}
+    seen_web: set[Path] = set()
+    for spec in _iter_remote_specs(entries):
+        if spec[CONF_TYPE] == TYPE_WEB:
+            if (path := _web_font_path(spec)) not in seen_web:
+                seen_web.add(path)
+                stage1.append(RemoteFile(spec[CONF_URL], path))
+        elif (css_path := _gfonts_css_path(spec)) not in stale_gfonts and (
+            not external_files.is_file_recent(
+                _gfonts_ttf_path(spec), spec[CONF_REFRESH]
+            )
+        ):
+            stale_gfonts[css_path] = spec
+            stage1.append(RemoteFile(_gfonts_css_url(spec), css_path))
+    yield stage1
+
+    yield [
+        RemoteFile(ttf_url, _gfonts_ttf_path(spec))
+        for css_path, spec in stale_gfonts.items()
+        # Only trust CSS that stage one actually refreshed this run; a
+        # leftover from an earlier run may name a rotated ttf URL.
+        if external_files.is_fresh_this_run(css_path)
+        and css_path.exists()
+        and (ttf_url := _parse_gfonts_css(css_path.read_text("utf-8", "replace")))
+        is not None
+    ]
+
+
+def validate_file_shorthand(value: object) -> ConfigType:
+    value = cv.string_strict(value)
+    if (data := _shorthand_to_file_dict(value)) is None:
+        data = {CONF_TYPE: TYPE_LOCAL, CONF_PATH: value}
+    return font_file_schema(data)
 
 
 TYPED_FILE_SCHEMA = cv.typed_schema(
diff --git a/esphome/components/gsl3670/touchscreen.py b/esphome/components/gsl3670/touchscreen.py
index fc0318f076..ccccf06d69 100644
--- a/esphome/components/gsl3670/touchscreen.py
+++ b/esphome/components/gsl3670/touchscreen.py
@@ -29,6 +29,8 @@ from esphome.const import (
     CONF_URL,
 )
 from esphome.core import ID
+from esphome.external_files import RemoteFile
+from esphome.types import ConfigType
 
 DEPENDENCIES = ["i2c"]
 AUTO_LOAD = ["touchscreen"]
@@ -103,8 +105,7 @@ def _validate_firmware_data(data: bytes, source: str) -> None:
 
 def _cache_path(url: str) -> Path:
     """Cache path for a downloaded firmware blob, keyed by URL."""
-    key = hashlib.sha256(url.encode()).hexdigest()[:8]
-    return external_files.compute_local_file_dir(DOMAIN) / key
+    return external_files.compute_local_file_path(DOMAIN, url)
 
 
 def firmware_path(firmware: dict) -> Path:
@@ -156,6 +157,23 @@ FIRMWARE_SCHEMA = cv.All(
 )
 
 
+def _extract_firmware_ref(entry: ConfigType) -> RemoteFile | None:
+    firmware = entry.get(CONF_FIRMWARE)
+    if firmware is None:
+        model = str(entry.get(CONF_MODEL, "CUSTOM")).upper()
+        firmware = MODELS.get(model, {}).get(CONF_FIRMWARE)
+    if (
+        isinstance(firmware, dict)
+        and CONF_FILE not in firmware
+        and isinstance(url := firmware.get(CONF_URL), str)
+    ):
+        return RemoteFile(url, _cache_path(url))
+    return None
+
+
+PREFETCH_FILES = external_files.single_stage_prefetch(_extract_firmware_ref)
+
+
 def _config_schema(config):
     model_option = {
         cv.Optional(CONF_MODEL, default="CUSTOM"): cv.one_of(*MODELS, upper=True)
diff --git a/esphome/components/micro_wake_word/__init__.py b/esphome/components/micro_wake_word/__init__.py
index 255923f878..092c4977ce 100644
--- a/esphome/components/micro_wake_word/__init__.py
+++ b/esphome/components/micro_wake_word/__init__.py
@@ -166,12 +166,7 @@ MANIFEST_SCHEMA_V2 = cv.Schema(
 
 
 def _compute_local_file_path(config: dict) -> Path:
-    url = config[CONF_URL]
-    h = hashlib.new("sha256")
-    h.update(url.encode())
-    key = h.hexdigest()[:8]
-    base_dir = external_files.compute_local_file_dir(DOMAIN)
-    return base_dir / key
+    return external_files.compute_local_file_path(DOMAIN, config[CONF_URL])
 
 
 def _convert_manifest_v1_to_v2(v1_manifest):
@@ -389,11 +384,14 @@ def _download_http_models(config: ConfigType) -> ConfigType:
         return config
 
     external_files.download_content_many(
-        ((url, path / "manifest.json") for path, url in http_models.items()),
+        (
+            external_files.RemoteFile(url, path / "manifest.json")
+            for path, url in http_models.items()
+        ),
         description="wake word manifest(s)",
     )
 
-    model_files: list[tuple[str, Path]] = []
+    model_files: list[external_files.RemoteFile] = []
     errors: list[cv.Invalid] = []
     for path, url in http_models.items():
         try:
@@ -412,7 +410,7 @@ def _download_http_models(config: ConfigType) -> ConfigType:
                 cv.Invalid(f"Manifest file at {url} is missing the 'model' key")
             )
             continue
-        model_files.append((urljoin(url, model), path / model))
+        model_files.append(external_files.RemoteFile(urljoin(url, model), path / model))
     if errors:
         raise cv.MultipleInvalid(errors)
 
diff --git a/esphome/components/shelly_dimmer/light.py b/esphome/components/shelly_dimmer/light.py
index cd6d858067..dd99fcbc90 100644
--- a/esphome/components/shelly_dimmer/light.py
+++ b/esphome/components/shelly_dimmer/light.py
@@ -2,9 +2,7 @@ import hashlib
 from pathlib import Path
 import re
 
-import requests
-
-from esphome import pins
+from esphome import external_files, pins
 import esphome.codegen as cg
 from esphome.components import light, sensor, uart
 from esphome.components.const import CONF_SHA256
@@ -28,8 +26,9 @@ from esphome.const import (
     UNIT_VOLT,
     UNIT_WATT,
 )
-from esphome.core import CORE, HexInt
-from esphome.happy_eyeballs import ensure_happy_eyeballs
+from esphome.core import HexInt
+from esphome.external_files import RemoteFile
+from esphome.types import ConfigType
 
 DOMAIN = "shelly_dimmer"
 AUTO_LOAD = ["sensor"]
@@ -76,46 +75,85 @@ def parse_firmware_version(value):
     return major, minor
 
 
-def get_firmware(value):
+def _firmware_cache_path(name: str) -> Path:
+    return external_files.compute_local_file_dir(DOMAIN) / f"{name}_fw_stm.bin"
+
+
+def _firmware_path(url: str, sha: str | None) -> Path:
+    """Cache path for a firmware blob: sha-keyed when verifiable, else
+    URL-keyed. Shared by the validator and the prefetch hook."""
+    return _firmware_cache_path(
+        sha.lower() if sha else external_files.url_cache_key(url)
+    )
+
+
+def get_firmware(value: ConfigType) -> list[HexInt] | None:
     if not value[CONF_UPDATE]:
         return None
 
-    def dl(url):
-        try:
-            ensure_happy_eyeballs()
-            req = requests.get(url, timeout=30)
-            req.raise_for_status()
-        except requests.exceptions.RequestException as e:
-            raise cv.Invalid(f"Could not download firmware file ({url}): {e}") from e
-
-        h = hashlib.new("sha256")
-        h.update(req.content)
-        return req.content, h.hexdigest()
-
     url = value[CONF_URL]
 
-    if CONF_SHA256 in value:  # we have a hash, enable caching
-        path = Path(CORE.data_dir) / DOMAIN / (value[CONF_SHA256] + "_fw_stm.bin")
-
-        if not path.is_file():
-            firmware_data, dl_hash = dl(url)
-
-            if dl_hash != value[CONF_SHA256]:
-                raise cv.Invalid(
-                    f"Hash mismatch for {url}: {dl_hash} != {value[CONF_SHA256]}"
-                )
-
-            path.parent.mkdir(exist_ok=True, parents=True)
-            path.write_bytes(firmware_data)
-
-        else:
+    if expected := value.get(CONF_SHA256):
+        expected = expected.lower()
+        path = _firmware_path(url, expected)
+        if path.is_file():
             firmware_data = path.read_bytes()
-    else:  # no caching, download every time
-        firmware_data, dl_hash = dl(url)
+            if hashlib.sha256(firmware_data).hexdigest() == expected:
+                return [HexInt(x) for x in firmware_data]
+            # A corrupted or foreign cache entry must never be trusted just
+            # because the file exists; discard it and download again.
+            path.unlink()
+        firmware_data = external_files.download_content(url, path)
+        if (actual := hashlib.sha256(firmware_data).hexdigest()) != expected:
+            path.unlink(missing_ok=True)
+            raise cv.Invalid(f"Hash mismatch for {url}: {actual} != {expected}")
+    else:
+        # No hash to verify the bytes, so an unrevalidated copy is an
+        # error rather than a silent fallback.
+        firmware_data = external_files.download_content(
+            url,
+            _firmware_path(url, None),
+            allow_stale=False,
+        )
 
     return [HexInt(x) for x in firmware_data]
 
 
+def _extract_firmware_ref(entry: ConfigType) -> RemoteFile | None:
+    firmware = entry.get(CONF_FIRMWARE)
+    if not isinstance(firmware, dict):
+        return None
+    try:
+        # cv.boolean, not truthiness: `update: "false"` is a valid False.
+        if not cv.boolean(firmware.get(CONF_UPDATE, False)):
+            return None
+    except cv.Invalid:
+        return None
+    url = firmware.get(CONF_URL)
+    sha = firmware.get(CONF_SHA256)
+    if url is None and (known := KNOWN_FIRMWARE.get(str(firmware.get(CONF_VERSION)))):
+        url, sha = known
+    if not isinstance(url, str):
+        return None
+    if sha is not None:
+        # Reject anything but a well-formed hash; a raw string would
+        # otherwise become a path component before validation runs.
+        try:
+            sha = validate_sha256(sha)
+        except (cv.Invalid, ValueError, TypeError):
+            return None
+    path = _firmware_path(url, sha)
+    if sha is not None and path.is_file():
+        # Content-addressed and already on disk; get_firmware verifies it
+        # by hash, so there is nothing to revalidate.
+        return None
+    # No hash means no stale copies, matching the validator's policy.
+    return RemoteFile(url, path, allow_stale=sha is not None)
+
+
+PREFETCH_FILES = external_files.single_stage_prefetch(_extract_firmware_ref)
+
+
 def validate_firmware(value):
     config = value.copy()
     if CONF_URL not in config:
diff --git a/esphome/config.py b/esphome/config.py
index b747c69b3a..987bb9c96a 100644
--- a/esphome/config.py
+++ b/esphome/config.py
@@ -1,14 +1,15 @@
 from __future__ import annotations
 
 import abc
-from contextlib import contextmanager
+from collections.abc import Iterator
+from contextlib import contextmanager, suppress
 import contextvars
 import copy
 import functools
 import heapq
 import logging
 import re
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 import voluptuous as vol
 
@@ -40,6 +41,9 @@ from esphome.util import OrderedDict, safe_print
 from esphome.voluptuous_schema import ExtraKeysInvalid
 from esphome.yaml_util import ESPHomeDataBase, ESPLiteralValue, is_secret
 
+if TYPE_CHECKING:
+    from esphome.external_files import RemoteFile
+
 _LOGGER = logging.getLogger(__name__)
 
 
@@ -717,6 +721,125 @@ class AutoLoadValidationStep(ConfigValidationStep):
         )
 
 
+# Backstop against a runaway PREFETCH_FILES generator; no real component
+# needs anywhere near this many stages (font, the deepest, uses two).
+_MAX_PREFETCH_STAGES = 10
+
+
+class PrefetchRemoteFilesValidationStep(ConfigValidationStep):
+    """Batch-download remote files referenced by the raw config.
+
+    Each round, the batches yielded by every ``PREFETCH_FILES`` hook (see
+    ``ComponentManifest.prefetch_files``) download in one parallel pass, so
+    per-entry schema validators find a warm cache. Must run between
+    AutoLoadValidationStep (-1.0) and MetadataValidationStep (-2.0):
+    metadata steps push priority-0 schema steps that pop immediately, so
+    this is the last point where every raw entry list is intact. Best
+    effort: failures are logged and memoized per run; the per-entry
+    validators stay authoritative.
+    """
+
+    priority = -1.5
+
+    def run(self, result: Config) -> None:
+        active: list[tuple[str, Iterator[list[RemoteFile]]]] = []
+
+        def warn_hook_failed(name: str, err: Exception) -> None:
+            # A broken hook must not fail validation; it only loses the
+            # batching speedup.
+            _LOGGER.warning("Remote file prefetch for %s failed: %s", name, err)
+            _LOGGER.debug("Prefetch hook traceback", exc_info=err)
+
+        def start_hook(
+            name: str, manifest: ComponentManifest, entries: list[ConfigType]
+        ) -> None:
+            if (hook := manifest.prefetch_files) is None:
+                return
+            try:
+                active.append((name, iter(hook(entries))))
+            except Exception as err:  # noqa: BLE001  # pylint: disable=broad-except
+                warn_hook_failed(name, err)
+
+        for domain, conf in result.items():
+            if not isinstance(domain, str) or domain.startswith("."):
+                continue
+            if (component := get_component(domain)) is None:
+                continue
+            if component.prefetch_files is None and not component.is_platform_component:
+                continue
+            if conf is None or isinstance(conf, core.AutoLoad):
+                continue
+            entries = [
+                entry
+                for entry in (conf if isinstance(conf, list) else [conf])
+                if isinstance(entry, dict)
+            ]
+            if not entries:
+                continue
+            # A domain-level hook on a platform component receives every
+            # entry; overlap with per-platform hooks dedupes by path.
+            start_hook(domain, component, entries)
+            if not component.is_platform_component:
+                continue
+            by_platform: dict[str, list[ConfigType]] = {}
+            for entry in entries:
+                if isinstance(p_name := entry.get(CONF_PLATFORM), str):
+                    by_platform.setdefault(p_name, []).append(entry)
+            for p_name, p_entries in by_platform.items():
+                if (platform := get_platform(domain, p_name)) is not None:
+                    start_hook(f"{domain}.{p_name}", platform, p_entries)
+
+        # One stage per round; later stages can read what earlier ones
+        # fetched.
+        for _ in range(_MAX_PREFETCH_STAGES):
+            if not active:
+                break
+            items: list[RemoteFile] = []
+            still_active: list[tuple[str, Iterator[list[RemoteFile]]]] = []
+            for name, generator in active:
+                try:
+                    batch = list(next(generator))
+                except StopIteration:
+                    continue
+                except Exception as err:  # noqa: BLE001  # pylint: disable=broad-except
+                    warn_hook_failed(name, err)
+                    continue
+                items.extend(batch)
+                still_active.append((name, generator))
+            active = still_active
+            self._download(items)
+        for name, generator in active:
+            # A tripped backstop means a broken hook.
+            _LOGGER.warning(
+                "Remote file prefetch for %s stopped after %d stages",
+                name,
+                _MAX_PREFETCH_STAGES,
+            )
+            if (close := getattr(generator, "close", None)) is not None:
+                # close() runs hook code too; it must not fail validation.
+                with suppress(Exception):
+                    close()
+
+    @staticmethod
+    def _download(items: list[RemoteFile]) -> None:
+        if not items:
+            return
+        # Imported lazily: requests is a heavy import (~85ms) and is only
+        # needed when a config actually references remote files.
+        from esphome import external_files
+
+        try:
+            external_files.download_content_many(items, description="remote file(s)")
+        except cv.Invalid as err:
+            # INFO: the trace if an extractor's cache path ever drifts from
+            # its validator's, hiding the memoized failure replay.
+            _LOGGER.info("Remote file prefetch download failed: %s", err)
+        except Exception as err:  # noqa: BLE001  # pylint: disable=broad-except
+            # The batch downloader itself broke; make it visible.
+            _LOGGER.warning("Remote file prefetch failed: %s", err)
+            _LOGGER.debug("Prefetch download traceback", exc_info=err)
+
+
 class MetadataValidationStep(ConfigValidationStep):
     """Validate component metadata
 
@@ -1259,6 +1382,7 @@ def validate_config(
 
     for domain, conf in config.items():
         result.add_validation_step(LoadValidationStep(domain, conf))
+    result.add_validation_step(PrefetchRemoteFilesValidationStep())
     result.add_validation_step(IDPassValidationStep())
     result.add_validation_step(CoreFinalValidateStep())
     result.add_validation_step(PinUseValidationCheck())
diff --git a/esphome/external_files.py b/esphome/external_files.py
index 160a2b6c29..f30d429425 100644
--- a/esphome/external_files.py
+++ b/esphome/external_files.py
@@ -1,16 +1,16 @@
 from __future__ import annotations
 
-from collections.abc import Callable, Iterable
+from collections.abc import Callable, Iterable, Iterator
 from concurrent.futures import ThreadPoolExecutor
 import contextlib
+from dataclasses import dataclass, field
 from datetime import UTC, datetime
+import hashlib
 import logging
 import os
 from pathlib import Path
 import time
 
-import requests
-
 import esphome.config_validation as cv
 from esphome.const import CONF_FILE, CONF_TYPE, CONF_URL, __version__
 from esphome.core import CORE, EsphomeError, TimePeriodSeconds
@@ -21,8 +21,54 @@ from esphome.types import ConfigType
 _LOGGER = logging.getLogger(__name__)
 CODEOWNERS = ["@landonr"]
 
+DOMAIN = "external_files"
+
 NETWORK_TIMEOUT = 30
 
+
+@dataclass(frozen=True, slots=True)
+class RemoteFile:
+    """A remote file to prefetch, yielded in stages by ``PREFETCH_FILES``
+    hooks. A dataclass rather than a tuple so fields can be added later."""
+
+    url: str
+    path: Path
+    # False when nothing downstream can verify the bytes; a copy that
+    # cannot be revalidated is then an error, not a silent fallback.
+    allow_stale: bool = True
+
+
+@dataclass(frozen=True, slots=True)
+class FailedDownload:
+    """What went wrong for a cache path this run, kept for fast replay."""
+
+    url: str
+    message: str
+    cause: BaseException
+
+
+@dataclass
+class ExternalFilesRunData:
+    """Per-run download state, cleared by ``CORE.reset()`` between runs."""
+
+    # Verified fresh this run; later touches skip even the conditional HEAD.
+    fresh_paths: set[Path] = field(default_factory=set)
+    # Served from disk without revalidation; strict callers reject these.
+    stale_paths: set[Path] = field(default_factory=set)
+    # Served under skip_external_update, deliberately unchecked; skips the
+    # network like fresh_paths but never counts as verified.
+    unchecked_paths: set[Path] = field(default_factory=set)
+    # Failed with no usable copy; later touches replay the error fast.
+    failed_paths: dict[Path, FailedDownload] = field(default_factory=dict)
+
+
+def _run_data() -> ExternalFilesRunData:
+    if (data := CORE.data.get(DOMAIN)) is not None:
+        return data
+    # setdefault: first touch may race on download_content_many's workers.
+    return CORE.data.setdefault(DOMAIN, ExternalFilesRunData())
+
+
 IF_MODIFIED_SINCE = "If-Modified-Since"
 IF_NONE_MATCH = "If-None-Match"
 ETAG = "ETag"
@@ -93,6 +139,9 @@ def _write_etag(local_file_path: Path, etag: str | None) -> None:
 def has_remote_file_changed(
     url: str, local_file_path: Path, timeout: int = NETWORK_TIMEOUT
 ) -> bool:
+    # 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)
@@ -127,6 +176,9 @@ def has_remote_file_changed(
                 )
                 if (new_etag := response.headers.get(ETAG)) and new_etag != etag:
                     _write_etag(local_file_path, new_etag)
+                # A confirmed 304 supersedes any earlier failed
+                # revalidation of this file.
+                _run_data().stale_paths.discard(local_file_path)
                 return False
             _LOGGER.debug("has_remote_file_changed: File modified")
             return True
@@ -136,6 +188,9 @@ def has_remote_file_changed(
                 url,
                 e,
             )
+            # The copy is a fallback, not a verified 304; record that so
+            # callers that must not use unverified bytes can reject it.
+            _run_data().stale_paths.add(local_file_path)
             return False
 
     _LOGGER.debug("has_remote_file_changed: File doesn't exists at %s", local_file_path)
@@ -159,14 +214,81 @@ def compute_local_file_dir(domain: str) -> Path:
     return base_directory
 
 
-def download_content(url: str, path: Path, timeout: int = NETWORK_TIMEOUT) -> bytes:
+def url_cache_key(url: str) -> str:
+    """Short stable cache key for a URL."""
+    return hashlib.sha256(url.encode()).hexdigest()[:8]
+
+
+def compute_local_file_path(domain: str, url: str) -> Path:
+    """Cache path for a URL-keyed download under the domain's cache dir.
+
+    Pure (no mkdir); parent directories are created at write time.
+    """
+    return Path(CORE.data_dir) / domain / url_cache_key(url)
+
+
+def is_fresh_this_run(path: Path) -> bool:
+    """Whether `path` was verified or downloaded during this run."""
+    return path in _run_data().fresh_paths
+
+
+def download_content(
+    url: str,
+    path: Path,
+    timeout: int = NETWORK_TIMEOUT,
+    allow_stale: bool = True,
+    return_content: bool = True,
+) -> bytes:
+    """Download `url` into `path` and return the bytes, using the cache.
+
+    On network failure an on-disk copy is served with a warning, unless
+    ``allow_stale=False``. ``CORE.skip_external_update`` always serves the
+    copy. ``return_content=False`` skips the disk read on cache hits.
+    """
+
+    # Deferred so configs with no remote files skip the heavy import.
+    import requests
+
+    def _cached() -> bytes:
+        return path.read_bytes() if return_content else b""
+
+    # Memoized paths skip the network entirely; concurrent access is safe
+    # because download_content_many dedupes by path before fanning out.
+    run_data = _run_data()
+    fresh_paths = run_data.fresh_paths
+    if (path in fresh_paths or path in run_data.unchecked_paths) and path.exists():
+        return _cached()
+    if allow_stale and path in run_data.stale_paths and path.exists():
+        # Strict callers fall through to try the network themselves.
+        _LOGGER.info("Using cached copy of %s that could not be revalidated", url)
+        return _cached()
+    if (failure := run_data.failed_paths.get(path)) is not None:
+        if not path.exists():
+            if failure.url == url:
+                raise cv.Invalid(failure.message) from failure.cause
+            raise cv.Invalid(
+                f"Could not download from {url}: an earlier download of "
+                f"{failure.url} to the same cache file failed: {failure.cause}"
+            ) 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)
-        return path.read_bytes()
+        run_data.unchecked_paths.add(path)
+        return _cached()
     if not has_remote_file_changed(url, path, timeout):
+        if path in run_data.stale_paths:
+            # The HEAD fell back to the copy without confirming it.
+            if not allow_stale:
+                raise cv.Invalid(
+                    f"Could not check {url} for updates due to a network error "
+                    f"and the cached copy cannot be verified"
+                )
+            return _cached()
         _LOGGER.debug("Remote file has not changed %s", url)
-        return path.read_bytes()
+        fresh_paths.add(path)
+        return _cached()
 
     _LOGGER.info("Downloading %s", url)
     _LOGGER.debug("Saving to %s", path)
@@ -185,16 +307,24 @@ def download_content(url: str, path: Path, timeout: int = NETWORK_TIMEOUT) -> by
         data = req.content
     except requests.exceptions.RequestException as e:
         if path.exists():
+            # Memoized so a flaky host warns once per run, not per consumer.
+            run_data.stale_paths.add(path)
+            if not allow_stale:
+                raise cv.Invalid(f"Could not download from {url}: {e}") from e
             _LOGGER.warning(
                 "Could not download from %s due to network error (%s), using cached file",
                 url,
                 e,
             )
-            return path.read_bytes()
-        raise cv.Invalid(f"Could not download from {url}: {e}") from e
+            return _cached()
+        message = f"Could not download from {url}: {e}"
+        run_data.failed_paths[path] = FailedDownload(url, message, e)
+        raise cv.Invalid(message) from e
 
     write_file(path, data)
     _write_etag(path, req.headers.get(ETAG))
+    fresh_paths.add(path)
+    run_data.stale_paths.discard(path)
     return data
 
 
@@ -207,50 +337,47 @@ DEFAULT_DOWNLOAD_WORKERS = 8
 
 
 def download_content_many(
-    items: Iterable[tuple[str, Path]],
+    items: Iterable[RemoteFile],
     timeout: int = NETWORK_TIMEOUT,
     max_workers: int = DEFAULT_DOWNLOAD_WORKERS,
     description: str = "remote file(s)",
 ) -> None:
-    """Run `download_content` for each (url, path) pair concurrently.
+    """Run `download_content` for each `RemoteFile` concurrently.
 
-    `description` names the kind of files in the progress log line, e.g.
-    "wake word manifest(s)".
-
-    Wall time drops from `sum(latency)` to roughly `max(latency)` for cached
-    files where the HEAD round-trip dominates. All workers run to
-    completion before this returns; every `cv.Invalid` raised by a worker
-    is collected and surfaced together as `cv.MultipleInvalid` so the user
-    sees every broken file in a single validation pass instead of fixing
-    them one round-trip at a time.
-
-    Items are de-duplicated by `path` -- two callers asking for the same
-    cache file (e.g. the same URL referenced twice in a config) would
-    otherwise race on `download_content`'s non-atomic write. When the
-    same `path` appears more than once, the last URL wins (standard dict
-    comprehension semantics); in practice duplicate paths only arise when
-    the URL is duplicated, so the choice doesn't matter.
+    `description` names the files in the progress log line. All workers run
+    to completion; every `cv.Invalid` raised is surfaced together as
+    `cv.MultipleInvalid`. Items dedupe by `path` (avoiding write races on
+    the same cache file); the last URL wins and a strict
+    `allow_stale=False` from any duplicate is kept.
     """
-    seen: dict[Path, str] = {path: url for url, path in items}
-    if not seen:
+    seen: dict[Path, RemoteFile] = {}
+    for file in items:
+        if (prior := seen.get(file.path)) is not None and not prior.allow_stale:
+            file = RemoteFile(file.url, file.path, allow_stale=False)
+        seen[file.path] = file
+    unique = list(seen.values())
+    if not unique:
         return
     ensure_happy_eyeballs()
-    _LOGGER.info("Checking %d %s for updates", len(seen), description)
-    if len(seen) == 1:
-        path, url = next(iter(seen.items()))
-        download_content(url, path, timeout)
+    _LOGGER.info("Checking %d %s for updates", len(unique), description)
+
+    def _download_one(file: RemoteFile) -> None:
+        download_content(
+            file.url,
+            file.path,
+            timeout,
+            allow_stale=file.allow_stale,
+            return_content=False,
+        )
+
+    if len(unique) == 1:
+        _download_one(unique[0])
         return
 
-    def _download_one(path_url: tuple[Path, str]) -> None:
-        # `seen` stores entries as (path, url) so the dict can dedupe by
-        # path; flip them back to download_content's (url, path) order.
-        path, url = path_url
-        download_content(url, path, timeout)
-
-    workers = max(1, min(max_workers, len(seen)))
+    workers = max(1, min(max_workers, len(unique)))
     errors: list[cv.Invalid] = []
     with ThreadPoolExecutor(max_workers=workers) as ex:
-        futures = [ex.submit(_download_one, item) for item in seen.items()]
+        futures = [ex.submit(_download_one, file) for file in unique]
         for future in futures:
             try:
                 future.result()
@@ -263,6 +390,21 @@ def download_content_many(
     raise cv.MultipleInvalid(errors)
 
 
+def single_stage_prefetch(
+    extract: Callable[[ConfigType], RemoteFile | None],
+) -> Callable[[list[ConfigType]], Iterator[list[RemoteFile]]]:
+    """Build a one-batch ``PREFETCH_FILES`` hook from a per-entry extractor.
+
+    Covers the common case of one remote file per raw config entry;
+    components with staged downloads write their own generator.
+    """
+
+    def prefetch_files(entries: list[ConfigType]) -> Iterator[list[RemoteFile]]:
+        yield [ref for entry in entries if (ref := extract(entry)) is not None]
+
+    return prefetch_files
+
+
 # Each component that uses external_files defines its own local
 # `TYPE_WEB = "web"`; the string is repeated here rather than imported
 # because there is no canonical `TYPE_WEB` in `esphome.const` to share.
@@ -282,7 +424,7 @@ def download_web_files_in_config(
     slotted directly into a `cv.All(...)` chain.
     """
     download_content_many(
-        (conf_file[CONF_URL], path_for(conf_file))
+        RemoteFile(conf_file[CONF_URL], path_for(conf_file))
         for entry in config
         if (conf_file := entry.get(CONF_FILE, {})).get(CONF_TYPE) == WEB_TYPE
     )
diff --git a/esphome/loader.py b/esphome/loader.py
index 22db8b156a..7a659aa0a8 100644
--- a/esphome/loader.py
+++ b/esphome/loader.py
@@ -1,4 +1,4 @@
-from collections.abc import Callable
+from collections.abc import Callable, Iterable
 from contextlib import AbstractContextManager
 from dataclasses import dataclass
 import importlib
@@ -16,6 +16,7 @@ from esphome.types import ConfigType
 
 if TYPE_CHECKING:
     from esphome.cpp_generator import MockObjClass
+    from esphome.external_files import RemoteFile
 
 # `esphome.core.config` is imported lazily in `_lookup_module` when the
 # "esphome" pseudo-component is first resolved. It pulls in
@@ -135,6 +136,21 @@ class ComponentManifest:
         """
         return getattr(self.module, "FINAL_VALIDATE_SCHEMA", None)
 
+    @property
+    def prefetch_files(
+        self,
+    ) -> Callable[[list[ConfigType]], Iterable[list["RemoteFile"]]] | None:
+        """Optional `PREFETCH_FILES` hook for batched remote file downloads.
+
+        A generator called once per run with the component's raw, pre-schema
+        config entries; each yield is a stage of ``RemoteFile`` downloaded in
+        one parallel pass before schema validation, so a later stage may
+        derive URLs from earlier files' content. Best effort: skip anything
+        unrecognized. On platform components, place it on the platform
+        sub-module; a domain-module hook receives every entry.
+        """
+        return getattr(self.module, "PREFETCH_FILES", None)
+
     @property
     def legacy_config_migrate(self) -> Callable[[ConfigType], ConfigType | None] | None:
         """Optional `LEGACY_CONFIG_MIGRATE` callable on a platform component module.
diff --git a/tests/component_tests/gsl3670/test_init.py b/tests/component_tests/gsl3670/test_init.py
index 8528cf23ca..950fa389be 100644
--- a/tests/component_tests/gsl3670/test_init.py
+++ b/tests/component_tests/gsl3670/test_init.py
@@ -87,13 +87,11 @@ def test_cache_path_is_deterministic_per_url(
     monkeypatch: pytest.MonkeyPatch, tmp_path: Path
 ) -> None:
     """The cache path is derived from (and stable for) the URL."""
-    monkeypatch.setattr(
-        gsl.external_files, "compute_local_file_dir", lambda _: tmp_path
-    )
+    monkeypatch.setenv("ESPHOME_DATA_DIR", str(tmp_path))
     first = gsl._cache_path(VALID_URL)
     assert first == gsl._cache_path(VALID_URL)
     assert first != gsl._cache_path("https://example.com/other.bin")
-    assert first.parent == tmp_path
+    assert first.parent == tmp_path / "gsl3670"
 
 
 def test_firmware_path_prefers_local_file(tmp_path: Path) -> None:
@@ -106,9 +104,7 @@ def test_firmware_path_uses_cache_for_url(
     monkeypatch: pytest.MonkeyPatch, tmp_path: Path
 ) -> None:
     """A ``url`` source resolves to the cache path for that URL."""
-    monkeypatch.setattr(
-        gsl.external_files, "compute_local_file_dir", lambda _: tmp_path
-    )
+    monkeypatch.setenv("ESPHOME_DATA_DIR", str(tmp_path))
     assert gsl.firmware_path({"url": VALID_URL}) == gsl._cache_path(VALID_URL)
 
 
@@ -145,9 +141,7 @@ def test_firmware_url_downloads_and_validates(
 ) -> None:
     """A url source downloads the content and validates its structure."""
     data = _make_firmware()
-    monkeypatch.setattr(
-        gsl.external_files, "compute_local_file_dir", lambda _: tmp_path
-    )
+    monkeypatch.setenv("ESPHOME_DATA_DIR", str(tmp_path))
     monkeypatch.setattr(gsl.external_files, "download_content", lambda url, path: data)
     assert gsl._validate_firmware({"url": VALID_URL}) == {"url": VALID_URL}
 
@@ -157,9 +151,7 @@ def test_firmware_url_sha256_mismatch_rejected(
 ) -> None:
     """A configured SHA-256 that does not match the download is rejected."""
     data = _make_firmware()
-    monkeypatch.setattr(
-        gsl.external_files, "compute_local_file_dir", lambda _: tmp_path
-    )
+    monkeypatch.setenv("ESPHOME_DATA_DIR", str(tmp_path))
     monkeypatch.setattr(gsl.external_files, "download_content", lambda url, path: data)
     with pytest.raises(cv.Invalid, match="SHA-256 mismatch"):
         gsl._validate_firmware({"url": VALID_URL, "sha256": "00" * 32})
@@ -169,9 +161,7 @@ def test_firmware_url_invalid_structure_rejected(
     monkeypatch: pytest.MonkeyPatch, tmp_path: Path
 ) -> None:
     """Downloaded content that is not a valid blob is rejected."""
-    monkeypatch.setattr(
-        gsl.external_files, "compute_local_file_dir", lambda _: tmp_path
-    )
+    monkeypatch.setenv("ESPHOME_DATA_DIR", str(tmp_path))
     monkeypatch.setattr(
         gsl.external_files, "download_content", lambda url, path: b"\x00\x01\x02"
     )
diff --git a/tests/unit_tests/components/bme68x_bsec2/__init__.py b/tests/unit_tests/components/bme68x_bsec2/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/tests/unit_tests/components/bme68x_bsec2/test_init.py b/tests/unit_tests/components/bme68x_bsec2/test_init.py
new file mode 100644
index 0000000000..b34231a1aa
--- /dev/null
+++ b/tests/unit_tests/components/bme68x_bsec2/test_init.py
@@ -0,0 +1,64 @@
+"""Tests for the bme68x_bsec2 prefetch extraction."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from esphome.components import bme68x_bsec2 as bsec
+from esphome.loader import get_component
+
+
+def test_prefetch_applies_defaults(setup_core: Path) -> None:
+    [files] = list(bsec.PREFETCH_FILES([{"model": "bme680"}]))
+    assert len(files) == 1
+    assert "bme680_iaq_33v_3s_28d" in files[0].url
+    assert files[0].path == bsec._compute_local_file_path(files[0].url)
+
+
+def test_prefetch_normalizes_enum_case(setup_core: Path) -> None:
+    [files] = list(
+        bsec.PREFETCH_FILES(
+            [
+                {
+                    "model": "BME688",
+                    "sample_rate": "ulp",
+                    "supply_voltage": "1.8v",
+                    "algorithm_output": "REGRESSION",
+                    "operating_age": "4D",
+                }
+            ]
+        )
+    )
+    assert len(files) == 1
+    assert "bme688_reg_18v_300s_4d" in files[0].url
+
+
+def test_prefetch_skips_unknown_values(setup_core: Path) -> None:
+    entries = [
+        {"model": "bme999"},
+        {"model": "bme680", "sample_rate": "TURBO"},
+        {"model": "bme680", "algorithm_output": "psychic"},
+        {},
+    ]
+    assert list(bsec.PREFETCH_FILES(entries)) == [[]]
+
+
+def test_prefetch_matches_validator_url(setup_core: Path) -> None:
+    """The hook's URL equals _compute_url over the validated config shape."""
+    validated = {
+        "model": "bme688",
+        "operating_age": "28d",
+        "sample_rate": "LP",
+        "supply_voltage": "3.3V",
+        "algorithm_output": "classification",
+    }
+    [files] = list(bsec.PREFETCH_FILES([dict(validated)]))
+    assert files[0].url == bsec._compute_url(validated)
+
+
+def test_hook_is_wired_to_the_user_facing_domain() -> None:
+    """The i2c domain (the only user-facing one) exposes the hook."""
+
+    component = get_component("bme68x_bsec2_i2c")
+    assert component is not None
+    assert component.prefetch_files is bsec.PREFETCH_FILES
diff --git a/tests/unit_tests/components/file/__init__.py b/tests/unit_tests/components/file/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/tests/unit_tests/components/file/test_image.py b/tests/unit_tests/components/file/test_image.py
new file mode 100644
index 0000000000..a9c1684db3
--- /dev/null
+++ b/tests/unit_tests/components/file/test_image.py
@@ -0,0 +1,75 @@
+"""Tests for the file image platform's prefetch extraction."""
+
+from __future__ import annotations
+
+from pathlib import Path
+from unittest.mock import patch
+
+from esphome.components.file import image as file_image
+from esphome.external_files import RemoteFile
+from esphome.loader import get_component, get_platform
+
+
+def test_extract_mdi_shorthand(setup_core: Path) -> None:
+    ref = file_image._extract_file_ref("mdi:home")
+    assert ref is not None
+    assert ref.url == file_image.MDI_SOURCES["mdi"] + "home.svg"
+    assert ref.path.name == "home.svg"
+    assert ref.path.parent.name == "mdi"
+
+
+def test_extract_web_url(setup_core: Path) -> None:
+    url = "https://example.com/img.png"
+    ref = file_image._extract_file_ref(url)
+    assert ref == RemoteFile(url, file_image.compute_local_image_path(url))
+
+
+def test_extract_typed_dicts(setup_core: Path) -> None:
+    url = "https://example.com/img.png"
+    assert file_image._extract_file_ref({"source": "web", "url": url}) == RemoteFile(
+        url, file_image.compute_local_image_path(url)
+    )
+    ref = file_image._extract_file_ref({"source": "mdil", "icon": "home"})
+    assert ref is not None
+    assert ref.url == file_image.MDI_SOURCES["mdil"] + "home.svg"
+
+
+def test_extract_skips_local_and_garbage(setup_core: Path) -> None:
+    assert file_image._extract_file_ref("images/local.png") is None
+    assert file_image._extract_file_ref("mdi:not a valid icon!") is None
+    assert file_image._extract_file_ref({"source": "local", "path": "x.png"}) is None
+    assert file_image._extract_file_ref(42) is None
+    assert file_image._extract_file_ref(None) is None
+
+
+def test_prefetch_files_yields_remote_refs(setup_core: Path) -> None:
+    entries = [
+        {"file": "mdi:home"},
+        {"file": "images/local.png"},
+        {"file": "https://example.com/img.png"},
+        {"no_file_key": True},
+    ]
+    [files] = list(file_image.PREFETCH_FILES(entries))
+    assert len(files) == 2
+    assert files[0].url.endswith("home.svg")
+    assert files[1].url == "https://example.com/img.png"
+
+
+def test_extractor_matches_validator_path(setup_core: Path) -> None:
+    """The path the validator downloads to equals the extractor's path."""
+    with patch(
+        "esphome.components.file.image.external_files.download_content"
+    ) as mock_download:
+        file_image.validate_file_shorthand("mdi:home")
+
+    validated_path = mock_download.call_args[0][1]
+    assert validated_path == file_image._extract_file_ref("mdi:home").path
+
+
+def test_hook_is_wired_to_both_animation_domains() -> None:
+    """Both animation entry points expose the shared image hook."""
+
+    assert get_component("animation").prefetch_files is file_image.PREFETCH_FILES
+    assert (
+        get_platform("image", "animation").prefetch_files is file_image.PREFETCH_FILES
+    )
diff --git a/tests/unit_tests/components/font/__init__.py b/tests/unit_tests/components/font/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/tests/unit_tests/components/font/test_init.py b/tests/unit_tests/components/font/test_init.py
new file mode 100644
index 0000000000..0ea3a0e3a1
--- /dev/null
+++ b/tests/unit_tests/components/font/test_init.py
@@ -0,0 +1,229 @@
+"""Tests for the font component's prefetch extraction."""
+
+from __future__ import annotations
+
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from esphome import external_files
+from esphome.components import font
+import esphome.config_validation as cv
+from esphome.external_files import RemoteFile
+
+
+def _gspec(family: str, weight: int = 400, italic: bool = False) -> dict:
+    return {"family": family, "weight": weight, "italic": italic}
+
+
+def test_extract_gfonts_shorthand_defaults(setup_core: Path) -> None:
+    spec = font._extract_remote_font("gfonts://Roboto")
+    assert spec is not None
+    assert spec[font.CONF_FAMILY] == "Roboto"
+    assert spec[font.CONF_WEIGHT] == 400
+    assert spec[font.CONF_ITALIC] is False
+
+
+def test_extract_gfonts_shorthand_weight_variants(setup_core: Path) -> None:
+    assert font._extract_remote_font("gfonts://Roboto@bold")[font.CONF_WEIGHT] == 700
+    assert font._extract_remote_font("gfonts://Roboto@500")[font.CONF_WEIGHT] == 500
+
+
+def test_extract_gfonts_normalizes_quoted_italic(setup_core: Path) -> None:
+    """Boolean spellings the schema accepts are accepted by the extractor."""
+    spec = font._extract_remote_font(
+        {"type": "gfonts", "family": "Roboto", "italic": "true"}
+    )
+    assert spec is not None
+    assert spec[font.CONF_ITALIC] is True
+    assert (
+        font._extract_remote_font(
+            {"type": "gfonts", "family": "Roboto", "italic": "maybe"}
+        )
+        is None
+    )
+
+
+def test_extract_typed_gfonts_dict(setup_core: Path) -> None:
+    spec = font._extract_remote_font(
+        {"type": "gfonts", "family": "Roboto", "weight": "medium", "italic": True}
+    )
+    assert spec is not None
+    assert spec[font.CONF_WEIGHT] == 500
+    assert spec[font.CONF_ITALIC] is True
+
+
+def test_extract_web_font(setup_core: Path) -> None:
+    url = "https://example.com/font.ttf"
+    for value in (url, {"type": "web", "url": url}):
+        spec = font._extract_remote_font(value)
+        assert spec is not None
+        assert spec[font.CONF_URL] == url
+
+
+def test_extract_skips_local_and_garbage(setup_core: Path) -> None:
+    assert font._extract_remote_font("fonts/local.ttf") is None
+    assert font._extract_remote_font({"type": "local", "path": "x.ttf"}) is None
+    assert (
+        font._extract_remote_font({"type": "gfonts", "family": "R", "weight": "no"})
+        is None
+    )
+    assert font._extract_remote_font(42) is None
+
+
+def test_prefetch_yields_css_for_stale_gfont(setup_core: Path) -> None:
+    entries = [
+        {"file": "gfonts://Roboto"},
+        {"file": "fonts/local.ttf"},
+        {
+            "file": "https://example.com/font.ttf",
+            "extras": [{"file": "gfonts://Monocraft"}],
+        },
+    ]
+    batches = list(font.PREFETCH_FILES(entries))
+    urls = [file.url for file in batches[0]]
+    assert font._gfonts_css_url(_gspec("Roboto")) in urls
+    assert font._gfonts_css_url(_gspec("Monocraft")) in urls
+    assert "https://example.com/font.ttf" in urls
+    assert len(batches[0]) == 3
+
+
+def test_prefetch_skips_recent_ttf(setup_core: Path) -> None:
+    path = font._gfonts_ttf_path(_gspec("Roboto"))
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_bytes(b"cached ttf")
+
+    batches = list(font.PREFETCH_FILES([{"file": "gfonts://Roboto"}]))
+    assert batches == [[], []]
+
+
+def test_stage2_parses_cached_css(setup_core: Path) -> None:
+
+    css_path = font._gfonts_css_path(_gspec("Roboto"))
+    css_path.parent.mkdir(parents=True, exist_ok=True)
+    css_path.write_text(
+        "src: url(https://fonts.gstatic.com/roboto.ttf) format('truetype');"
+    )
+    # Stage two only trusts CSS confirmed fetched this run.
+    external_files._run_data().fresh_paths.add(css_path)
+
+    batches = list(font.PREFETCH_FILES([{"file": "gfonts://Roboto"}]))
+    assert batches[1] == [
+        RemoteFile(
+            "https://fonts.gstatic.com/roboto.ttf",
+            font._gfonts_ttf_path(_gspec("Roboto")),
+        )
+    ]
+
+
+def test_stage2_skips_missing_css(setup_core: Path) -> None:
+    batches = list(font.PREFETCH_FILES([{"file": "gfonts://NoCss"}]))
+    assert batches[1] == []
+
+
+def test_prefetch_handles_bare_mapping_extras(setup_core: Path) -> None:
+    """A bare-mapping extras value (valid raw config) is scanned."""
+    entries = [
+        {
+            "file": "fonts/local.ttf",
+            "extras": {"file": "gfonts://Roboto", "glyphs": "ABC"},
+        }
+    ]
+    batches = list(font.PREFETCH_FILES(entries))
+    assert [file.url for file in batches[0]] == [font._gfonts_css_url(_gspec("Roboto"))]
+
+
+def test_unparseable_gfonts_css_is_evicted(setup_core: Path) -> None:
+    """A CSS body that fails to parse is removed from the cache."""
+
+    spec = {
+        "family": "Roboto",
+        "weight": 400,
+        "italic": False,
+        "refresh": font._REFRESH_VALIDATOR("0s"),
+    }
+    css_path = font._gfonts_css_path(spec)
+    with (
+        patch(
+            "esphome.components.font.external_files.download_content",
+            return_value=b"no truetype url here",
+        ),
+        patch(
+            "esphome.components.font.external_files.is_fresh_this_run",
+            return_value=True,
+        ),
+        pytest.raises(cv.Invalid, match="please report this"),
+    ):
+        font.download_gfont(spec)
+    assert not css_path.exists()
+
+    with (
+        patch(
+            "esphome.components.font.external_files.download_content",
+            return_value=b"\xff\xfe\x00\x01binary",
+        ),
+        patch(
+            "esphome.components.font.external_files.is_fresh_this_run",
+            return_value=True,
+        ),
+        pytest.raises(cv.Invalid, match="not a text document"),
+    ):
+        font.download_gfont(spec)
+    assert not css_path.exists()
+
+
+def test_unrevalidated_gfonts_css_uses_cached_font(setup_core: Path) -> None:
+    """A CSS body that could not be revalidated is not parsed for a ttf
+    URL; the cached font is used instead."""
+    spec = {
+        "family": "Roboto",
+        "weight": 400,
+        "italic": False,
+        "refresh": font._REFRESH_VALIDATOR("0s"),
+    }
+    ttf_path = font._gfonts_ttf_path(spec)
+    ttf_path.parent.mkdir(parents=True, exist_ok=True)
+    ttf_path.write_bytes(b"cached ttf")
+    cache = MagicMock()
+    with (
+        patch.object(font, "FONT_CACHE", cache),
+        patch(
+            "esphome.components.font.external_files.download_content",
+            return_value=b"stale css",
+        ),
+    ):
+        assert font.download_gfont(spec) is spec
+    cache.__setitem__.assert_called_once_with(spec, ttf_path)
+
+
+def test_unrevalidated_gfonts_css_without_cached_font_errors(
+    setup_core: Path,
+) -> None:
+    """No verified CSS and no cached font is a clear error."""
+    spec = {
+        "family": "Roboto",
+        "weight": 500,
+        "italic": False,
+        "refresh": font._REFRESH_VALIDATOR("0s"),
+    }
+    with (
+        patch(
+            "esphome.components.font.external_files.download_content",
+            return_value=b"stale css",
+        ),
+        pytest.raises(cv.Invalid, match="no cached font"),
+    ):
+        font.download_gfont(spec)
+
+
+def test_stage2_skips_css_not_fetched_this_run(setup_core: Path) -> None:
+    """A leftover CSS from an earlier run is not trusted for stage two."""
+    css_path = font._gfonts_css_path(_gspec("Roboto"))
+    css_path.parent.mkdir(parents=True, exist_ok=True)
+    css_path.write_text(
+        "src: url(https://fonts.gstatic.com/rotated.ttf) format('truetype');"
+    )
+
+    batches = list(font.PREFETCH_FILES([{"file": "gfonts://Roboto"}]))
+    assert batches[1] == []
diff --git a/tests/unit_tests/components/gsl3670/__init__.py b/tests/unit_tests/components/gsl3670/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/tests/unit_tests/components/gsl3670/test_touchscreen.py b/tests/unit_tests/components/gsl3670/test_touchscreen.py
new file mode 100644
index 0000000000..a4b96d72da
--- /dev/null
+++ b/tests/unit_tests/components/gsl3670/test_touchscreen.py
@@ -0,0 +1,35 @@
+"""Tests for the gsl3670 touchscreen prefetch extraction."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from esphome.components.gsl3670 import touchscreen as gsl
+from esphome.external_files import RemoteFile
+
+
+def test_prefetch_explicit_url(setup_core: Path) -> None:
+    url = "https://example.com/fw.bin"
+    entries = [{"platform": "gsl3670", "firmware": {"url": url}}]
+    assert list(gsl.PREFETCH_FILES(entries)) == [
+        [RemoteFile(url, gsl._cache_path(url))]
+    ]
+
+
+def test_prefetch_model_default_firmware(setup_core: Path) -> None:
+    entries = [{"platform": "gsl3670", "model": "seeed-reterminal-d1001"}]
+    [files] = list(gsl.PREFETCH_FILES(entries))
+    assert len(files) == 1
+    assert (
+        files[0].url == gsl.MODELS["SEEED-RETERMINAL-D1001"][gsl.CONF_FIRMWARE]["url"]
+    )
+    assert files[0].path == gsl._cache_path(files[0].url)
+
+
+def test_prefetch_skips_local_file_and_custom(setup_core: Path) -> None:
+    entries = [
+        {"platform": "gsl3670", "firmware": {"file": "fw.bin"}},
+        {"platform": "gsl3670", "model": "CUSTOM"},
+        {"platform": "gsl3670"},
+    ]
+    assert list(gsl.PREFETCH_FILES(entries)) == [[]]
diff --git a/tests/unit_tests/components/micro_wake_word/test_init.py b/tests/unit_tests/components/micro_wake_word/test_init.py
index 84371ab906..96fb73b18b 100644
--- a/tests/unit_tests/components/micro_wake_word/test_init.py
+++ b/tests/unit_tests/components/micro_wake_word/test_init.py
@@ -16,6 +16,7 @@ from esphome.const import (
     CONF_TYPE,
     CONF_URL,
 )
+from esphome.external_files import RemoteFile
 
 
 @pytest.fixture
@@ -114,12 +115,16 @@ def test_download_http_models_batches_manifests_then_models(
     assert mock_download_content_many.call_count == 2
     manifest_items = list(mock_download_content_many.call_args_list[0].args[0])
     assert manifest_items == [
-        (f"https://example.com/models/{name}.json", paths[name] / "manifest.json")
+        RemoteFile(
+            f"https://example.com/models/{name}.json", paths[name] / "manifest.json"
+        )
         for name in names
     ]
     model_items = list(mock_download_content_many.call_args_list[1].args[0])
     assert model_items == [
-        (f"https://example.com/models/{name}.tflite", paths[name] / f"{name}.tflite")
+        RemoteFile(
+            f"https://example.com/models/{name}.tflite", paths[name] / f"{name}.tflite"
+        )
         for name in names
     ]
 
diff --git a/tests/unit_tests/components/shelly_dimmer/__init__.py b/tests/unit_tests/components/shelly_dimmer/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/tests/unit_tests/components/shelly_dimmer/test_light.py b/tests/unit_tests/components/shelly_dimmer/test_light.py
new file mode 100644
index 0000000000..e5440db4c9
--- /dev/null
+++ b/tests/unit_tests/components/shelly_dimmer/test_light.py
@@ -0,0 +1,154 @@
+"""Tests for the shelly_dimmer firmware download and prefetch extraction."""
+
+from __future__ import annotations
+
+import hashlib
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+
+from esphome import external_files
+from esphome.components.shelly_dimmer import light as shd
+from esphome.config_validation import Invalid
+from esphome.external_files import RemoteFile
+
+
+def _sha(data: bytes) -> str:
+    return hashlib.sha256(data).hexdigest()
+
+
+def test_prefetch_known_version(setup_core: Path) -> None:
+    entries = [{"firmware": {"version": "51.6", "update": True}}]
+    stages = list(shd.PREFETCH_FILES(entries))
+    url, sha = shd.KNOWN_FIRMWARE["51.6"]
+    assert stages == [[RemoteFile(url, shd._firmware_cache_path(sha))]]
+
+
+def test_prefetch_normalizes_update_like_the_schema(setup_core: Path) -> None:
+    """Quoted booleans behave as the schema will normalize them."""
+    url, sha = shd.KNOWN_FIRMWARE["51.6"]
+    off = [{"firmware": {"version": "51.6", "update": "false"}}]
+    assert list(shd.PREFETCH_FILES(off)) == [[]]
+    on = [{"firmware": {"version": "51.6", "update": "true"}}]
+    assert list(shd.PREFETCH_FILES(on)) == [
+        [RemoteFile(url, shd._firmware_cache_path(sha))]
+    ]
+
+
+def test_prefetch_rejects_malformed_sha256(setup_core: Path) -> None:
+    """A raw sha256 that is not a hash never becomes a path component."""
+    entries = [
+        {
+            "firmware": {
+                "url": "https://example.com/fw.bin",
+                "sha256": "/tmp/payload",
+                "update": True,
+            }
+        }
+    ]
+    assert list(shd.PREFETCH_FILES(entries)) == [[]]
+
+
+def test_prefetch_skips_content_addressed_blob_on_disk(setup_core: Path) -> None:
+    """A sha-keyed cache file needs no revalidation; get_firmware hashes it."""
+    url, sha = shd.KNOWN_FIRMWARE["51.6"]
+    shd._firmware_cache_path(sha).write_bytes(b"pinned firmware")
+    entries = [{"firmware": {"version": "51.6", "update": True}}]
+    assert list(shd.PREFETCH_FILES(entries)) == [[]]
+
+
+def test_prefetch_explicit_url_without_sha(setup_core: Path) -> None:
+    url = "https://example.com/fw.bin"
+    entries = [{"firmware": {"url": url, "update": True}}]
+    stages = list(shd.PREFETCH_FILES(entries))
+    key = external_files.url_cache_key(url)
+    # No sha means the bytes cannot be verified, so the prefetch itself
+    # must carry the validator's strict no-stale policy.
+    assert stages == [
+        [RemoteFile(url, shd._firmware_cache_path(key), allow_stale=False)]
+    ]
+
+
+def test_prefetch_skips_no_update(setup_core: Path) -> None:
+    entries = [
+        {"firmware": {"version": "51.6"}},
+        {"firmware": "51.6"},
+        {"firmware": {"version": "0.0", "update": True}},
+        {},
+    ]
+    assert list(shd.PREFETCH_FILES(entries)) == [[]]
+
+
+def test_get_firmware_rejects_corrupted_cache(setup_core: Path) -> None:
+    """A cached blob failing its hash check is discarded and re-downloaded."""
+    good = b"good firmware"
+    expected = _sha(good)
+    path = shd._firmware_cache_path(expected)
+    path.write_bytes(b"corrupted blob")
+
+    with patch(
+        "esphome.components.shelly_dimmer.light.external_files.download_content",
+        return_value=good,
+    ) as mock_download:
+        result = shd.get_firmware(
+            {
+                "update": True,
+                "url": "https://example.com/fw.bin",
+                "sha256": expected,
+            }
+        )
+
+    mock_download.assert_called_once()
+    assert result == [int(b) for b in good]
+
+
+def test_get_firmware_trusts_valid_cache(setup_core: Path) -> None:
+    """A cached blob passing its hash check is used with zero network."""
+    good = b"good firmware"
+    expected = _sha(good)
+    shd._firmware_cache_path(expected).write_bytes(good)
+
+    with patch(
+        "esphome.components.shelly_dimmer.light.external_files.download_content"
+    ) as mock_download:
+        result = shd.get_firmware(
+            {
+                "update": True,
+                "url": "https://example.com/fw.bin",
+                "sha256": expected,
+            }
+        )
+
+    mock_download.assert_not_called()
+    assert result == [int(b) for b in good]
+
+
+def test_get_firmware_hash_mismatch_raises_and_uncaches(setup_core: Path) -> None:
+    """A fresh download failing its hash check raises and is not cached."""
+    expected = _sha(b"expected firmware")
+    path = shd._firmware_cache_path(expected)
+
+    with (
+        patch(
+            "esphome.components.shelly_dimmer.light.external_files.download_content",
+            return_value=b"wrong firmware",
+        ),
+        pytest.raises(Invalid, match="Hash mismatch"),
+    ):
+        shd.get_firmware(
+            {"update": True, "url": "https://example.com/fw.bin", "sha256": expected}
+        )
+
+    assert not path.exists()
+
+
+def test_get_firmware_without_sha_rejects_stale(setup_core: Path) -> None:
+    """The unverifiable no-hash branch must not accept a stale copy."""
+    with patch(
+        "esphome.components.shelly_dimmer.light.external_files.download_content",
+        return_value=b"fw",
+    ) as mock_download:
+        shd.get_firmware({"update": True, "url": "https://example.com/fw.bin"})
+
+    assert mock_download.call_args.kwargs["allow_stale"] is False
diff --git a/tests/unit_tests/test_config_prefetch.py b/tests/unit_tests/test_config_prefetch.py
new file mode 100644
index 0000000000..afb93a09a0
--- /dev/null
+++ b/tests/unit_tests/test_config_prefetch.py
@@ -0,0 +1,355 @@
+"""Tests for the remote file prefetch validation step."""
+
+from __future__ import annotations
+
+from collections.abc import Iterable
+from pathlib import Path
+from types import SimpleNamespace
+from typing import Any
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from esphome import core
+from esphome.config import Config, PrefetchRemoteFilesValidationStep
+import esphome.config_validation as cv
+from esphome.external_files import RemoteFile
+
+
+def _component(prefetch: Any = None, is_platform: bool = False) -> SimpleNamespace:
+    return SimpleNamespace(
+        is_platform_component=is_platform,
+        prefetch_files=prefetch,
+    )
+
+
+def _run_step(
+    domains: dict[str, Any],
+    components: dict[str, Any],
+    platforms: dict[tuple[str, str], Any] | None = None,
+    download_side_effect: Any = None,
+) -> tuple[Config, MagicMock]:
+    result = Config()
+    for domain, conf in domains.items():
+        result[domain] = conf
+    with (
+        patch("esphome.config.get_component", side_effect=components.get),
+        patch(
+            "esphome.config.get_platform",
+            side_effect=lambda d, p: (platforms or {}).get((d, p)),
+        ),
+        patch(
+            "esphome.external_files.download_content_many",
+            side_effect=download_side_effect,
+        ) as mock_download,
+    ):
+        PrefetchRemoteFilesValidationStep().run(result)
+    return result, mock_download
+
+
+def _downloaded(mock_download: MagicMock, call: int = 0) -> list[RemoteFile]:
+    return list(mock_download.call_args_list[call][0][0])
+
+
+def test_component_hook_receives_normalized_entries() -> None:
+    """A bare dict conf is passed to the hook as a one-entry list."""
+    seen: list[Any] = []
+
+    def hook(entries: list[dict]) -> Iterable[list[RemoteFile]]:
+        seen.append(entries)
+        yield [RemoteFile("https://example.com/a", Path("/cache/a"))]
+
+    _, mock_download = _run_step(
+        {"my_comp": {"key": "value"}},
+        {"my_comp": _component(prefetch=hook)},
+    )
+
+    assert seen == [[{"key": "value"}]]
+    mock_download.assert_called_once()
+    assert _downloaded(mock_download) == [
+        RemoteFile("https://example.com/a", Path("/cache/a"))
+    ]
+
+
+def test_platform_entries_are_grouped_per_platform() -> None:
+    """Platform domains route entries to each platform module's hook."""
+    seen_a: list[Any] = []
+    seen_b: list[Any] = []
+
+    def hook_a(entries: list[dict]) -> Iterable[list[RemoteFile]]:
+        seen_a.extend(entries)
+        yield [RemoteFile("url-a", Path("/a"))]
+
+    def hook_b(entries: list[dict]) -> Iterable[list[RemoteFile]]:
+        seen_b.extend(entries)
+        yield [RemoteFile("url-b", Path("/b"))]
+
+    entries = [
+        {"platform": "a", "n": 1},
+        {"platform": "b", "n": 2},
+        {"platform": "a", "n": 3},
+    ]
+    _, mock_download = _run_step(
+        {"image": entries},
+        {"image": _component(is_platform=True)},
+        platforms={
+            ("image", "a"): _component(prefetch=hook_a),
+            ("image", "b"): _component(prefetch=hook_b),
+        },
+    )
+
+    assert seen_a == [entries[0], entries[2]]
+    assert seen_b == [entries[1]]
+    assert sorted(_downloaded(mock_download), key=lambda f: f.url) == [
+        RemoteFile("url-a", Path("/a")),
+        RemoteFile("url-b", Path("/b")),
+    ]
+
+
+def test_hook_failure_does_not_fail_validation(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """A raising hook is logged and other hooks still prefetch."""
+
+    def bad_hook(entries: list[dict]) -> list[RemoteFile]:
+        raise RuntimeError("garbage config")
+
+    def good_hook(entries: list[dict]) -> Iterable[list[RemoteFile]]:
+        yield [RemoteFile("url", Path("/g"))]
+
+    _, mock_download = _run_step(
+        {"bad": {"x": 1}, "good": {"y": 2}},
+        {
+            "bad": _component(prefetch=bad_hook),
+            "good": _component(prefetch=good_hook),
+        },
+    )
+
+    assert "Remote file prefetch for bad failed" in caplog.text
+    assert _downloaded(mock_download) == [RemoteFile("url", Path("/g"))]
+
+
+def test_stages_download_between_resumptions() -> None:
+    """Each yielded stage is downloaded before the generator resumes."""
+    order: list[str] = []
+
+    def hook(entries: list[dict]) -> Iterable[list[RemoteFile]]:
+        order.append("stage1")
+        yield [RemoteFile("css-url", Path("/css"))]
+        order.append("stage2")
+        yield [RemoteFile("ttf-url", Path("/ttf"))]
+
+    def record_download(items: Any, description: str) -> None:
+        order.append(f"download:{[file.url for file in items]}")
+
+    _, mock_download = _run_step(
+        {"font": {"f": 1}},
+        {"font": _component(prefetch=hook)},
+        download_side_effect=record_download,
+    )
+
+    assert order == [
+        "stage1",
+        "download:['css-url']",
+        "stage2",
+        "download:['ttf-url']",
+    ]
+    assert mock_download.call_count == 2
+
+
+def test_runaway_generator_is_capped(caplog: pytest.LogCaptureFixture) -> None:
+    """An endless generator stops after the stage backstop."""
+
+    def hook(entries: list[dict]) -> Iterable[list[RemoteFile]]:
+        n = 0
+        while True:
+            yield [RemoteFile(f"url-{n}", Path(f"/f{n}"))]
+            n += 1
+
+    _, mock_download = _run_step(
+        {"my_comp": {"x": 1}},
+        {"my_comp": _component(prefetch=hook)},
+    )
+
+    assert mock_download.call_count == 10
+    assert "stopped after" in caplog.text
+
+
+def test_mid_stage_failure_stops_only_that_hook(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """A generator raising on a later stage does not affect other hooks."""
+
+    def flaky_hook(entries: list[dict]) -> Iterable[list[RemoteFile]]:
+        yield [RemoteFile("first", Path("/first"))]
+        raise RuntimeError("stage two exploded")
+
+    def steady_hook(entries: list[dict]) -> Iterable[list[RemoteFile]]:
+        yield [RemoteFile("one", Path("/one"))]
+        yield [RemoteFile("two", Path("/two"))]
+
+    _, mock_download = _run_step(
+        {"flaky": {"x": 1}, "steady": {"y": 2}},
+        {
+            "flaky": _component(prefetch=flaky_hook),
+            "steady": _component(prefetch=steady_hook),
+        },
+    )
+
+    assert "Remote file prefetch for flaky failed" in caplog.text
+    assert mock_download.call_count == 2
+    assert _downloaded(mock_download, 1) == [RemoteFile("two", Path("/two"))]
+
+
+def test_download_failure_is_swallowed() -> None:
+    """cv.Invalid from the batch download never escapes the step."""
+
+    def hook(entries: list[dict]) -> Iterable[list[RemoteFile]]:
+        yield [RemoteFile("url", Path("/p"))]
+
+    result, mock_download = _run_step(
+        {"my_comp": {"x": 1}},
+        {"my_comp": _component(prefetch=hook)},
+        download_side_effect=cv.Invalid("download failed"),
+    )
+
+    mock_download.assert_called_once()
+    assert not result.errors
+
+
+def test_domains_without_hooks_do_not_download() -> None:
+    """Components without PREFETCH_FILES cause no download call."""
+    _, mock_download = _run_step(
+        {"plain": {"x": 1}, ".ignored": {"y": 2}, "unknown": {"z": 3}},
+        {"plain": _component()},
+    )
+    mock_download.assert_not_called()
+
+
+def test_none_and_autoload_confs_are_skipped() -> None:
+    """None and AutoLoad confs never reach a hook."""
+    hook = MagicMock()
+    _, mock_download = _run_step(
+        {"a": None, "b": core.AutoLoad()},
+        {"a": _component(prefetch=hook), "b": _component(prefetch=hook)},
+    )
+    hook.assert_not_called()
+    mock_download.assert_not_called()
+
+
+def test_non_dict_entries_are_ignored() -> None:
+    """Garbage entries never reach a component hook."""
+    hook = MagicMock()
+    _, mock_download = _run_step(
+        {"my_comp": ["just-a-string", 42]},
+        {"my_comp": _component(prefetch=hook)},
+    )
+    hook.assert_not_called()
+    mock_download.assert_not_called()
+
+
+def test_platform_entries_without_platform_key_are_ignored() -> None:
+    """Entries with a missing or unknown platform never reach a hook."""
+    _, mock_download = _run_step(
+        {"image": [{"n": 1}, "garbage", {"platform": "unknown"}]},
+        {"image": _component(is_platform=True)},
+    )
+    mock_download.assert_not_called()
+
+
+def test_generator_still_alive_at_the_cap_is_warned_and_closed(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """A generator with a stage left at the cap is warned about and closed."""
+    closed: list[bool] = []
+
+    def hook(entries: list[dict]) -> Iterable[list[RemoteFile]]:
+        try:
+            for n in range(10):
+                yield [RemoteFile(f"url-{n}", Path(f"/f{n}"))]
+        finally:
+            closed.append(True)
+
+    _, mock_download = _run_step(
+        {"my_comp": {"x": 1}},
+        {"my_comp": _component(prefetch=hook)},
+    )
+
+    assert mock_download.call_count == 10
+    assert "stopped after" in caplog.text
+    assert closed == [True]
+
+
+def test_plain_iterable_hook_survives_the_cap(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """A hook returning a plain list of batches cannot crash the backstop."""
+
+    def hook(entries: list[dict]) -> list[list[RemoteFile]]:
+        return [[RemoteFile(f"url-{n}", Path(f"/f{n}"))] for n in range(12)]
+
+    _, mock_download = _run_step(
+        {"my_comp": {"x": 1}},
+        {"my_comp": _component(prefetch=hook)},
+    )
+
+    assert mock_download.call_count == 10
+    assert "stopped after" in caplog.text
+
+
+def test_domain_level_hook_on_platform_component() -> None:
+    """A hook on the platform component's domain module sees all entries."""
+    seen: list[Any] = []
+
+    def domain_hook(entries: list[dict]) -> Iterable[list[RemoteFile]]:
+        seen.append(entries)
+        yield [RemoteFile("domain-url", Path("/domain"))]
+
+    entries = [{"platform": "a", "n": 1}, {"platform": "b", "n": 2}]
+    _, mock_download = _run_step(
+        {"image": entries},
+        {"image": _component(prefetch=domain_hook, is_platform=True)},
+    )
+
+    assert seen == [entries]
+    assert _downloaded(mock_download) == [RemoteFile("domain-url", Path("/domain"))]
+
+
+def test_generator_raising_on_close_is_contained(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """A generator whose close() raises at the cap is logged, not crashed on."""
+
+    def hook(entries: list[dict]) -> Iterable[list[RemoteFile]]:
+        try:
+            for n in range(10):
+                yield [RemoteFile(f"url-{n}", Path(f"/f{n}"))]
+        except GeneratorExit:
+            raise RuntimeError("close exploded") from None
+
+    _, mock_download = _run_step(
+        {"my_comp": {"x": 1}},
+        {"my_comp": _component(prefetch=hook)},
+    )
+
+    assert mock_download.call_count == 10
+    assert "stopped after" in caplog.text
+
+
+def test_unexpected_download_error_is_logged_visibly(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """A broken batch downloader warns instead of silently disabling prefetch."""
+
+    def hook(entries: list[dict]) -> Iterable[list[RemoteFile]]:
+        yield [RemoteFile("url", Path("/p"))]
+
+    result, mock_download = _run_step(
+        {"my_comp": {"x": 1}},
+        {"my_comp": _component(prefetch=hook)},
+        download_side_effect=TypeError("not a RemoteFile"),
+    )
+
+    mock_download.assert_called_once()
+    assert not result.errors
+    assert "Remote file prefetch failed" in caplog.text
diff --git a/tests/unit_tests/test_external_files.py b/tests/unit_tests/test_external_files.py
index 16cee9564f..4e993ff4f3 100644
--- a/tests/unit_tests/test_external_files.py
+++ b/tests/unit_tests/test_external_files.py
@@ -3,6 +3,7 @@
 import os
 from pathlib import Path
 import time
+from typing import Any
 from unittest.mock import MagicMock, patch
 
 import pytest
@@ -26,19 +27,21 @@ def _seed_etag(cache_file: Path, etag: str) -> Path:
 
 @pytest.fixture
 def mock_requests_head() -> MagicMock:
-    """Patch `external_files.requests.head` so the conditional HEAD-request
-    validator can be tested without doing real HTTP.
+    """Patch `requests.head` so the conditional HEAD-request validator can
+    be tested without doing real HTTP. Patched on the requests module
+    because external_files imports it lazily inside the function.
     """
-    with patch("esphome.external_files.requests.head") as m:
+    with patch("requests.head") as m:
         yield m
 
 
 @pytest.fixture
 def mock_requests_get() -> MagicMock:
-    """Patch `external_files.requests.get` so the download path can be
-    tested without doing real HTTP.
+    """Patch `requests.get` so the download path can be tested without
+    doing real HTTP. Patched on the requests module because
+    external_files imports it lazily inside the function.
     """
-    with patch("esphome.external_files.requests.get") as m:
+    with patch("requests.get") as m:
         yield m
 
 
@@ -549,6 +552,10 @@ def test_download_content_skip_external_update_uses_cache(
     assert result == cached_content
     mock_has_remote_file_changed.assert_not_called()
     mock_requests_get.assert_not_called()
+    # Deliberately unchecked is memoized for the run but never "fresh".
+    assert not external_files.is_fresh_this_run(test_file)
+    assert external_files.download_content(url, test_file) == cached_content
+    mock_has_remote_file_changed.assert_not_called()
 
 
 def test_download_content_skip_external_update_downloads_when_missing(
@@ -587,10 +594,16 @@ def test_download_content_many_single_item_avoids_pool(
     mock_download_content: MagicMock, setup_core: Path
 ) -> None:
     """A single item should be downloaded inline (no thread pool overhead)."""
-    item = ("https://example.com/file.txt", setup_core / "f.txt")
+    item = external_files.RemoteFile(
+        "https://example.com/file.txt", setup_core / "f.txt"
+    )
     external_files.download_content_many([item])
     mock_download_content.assert_called_once_with(
-        item[0], item[1], external_files.NETWORK_TIMEOUT
+        item.url,
+        item.path,
+        external_files.NETWORK_TIMEOUT,
+        allow_stale=True,
+        return_content=False,
     )
 
 
@@ -602,7 +615,12 @@ def test_download_content_many_runs_in_parallel(
 
     barrier = threading.Barrier(3)
 
-    def slow_download(url: str, path: Path, timeout: int) -> bytes:
+    def slow_download(
+        url: str,
+        path: Path,
+        *args: Any,
+        **kwargs: Any,
+    ) -> bytes:
         # If calls were serial this would deadlock (third caller never arrives
         # while the first is blocked at the barrier).
         barrier.wait(timeout=2.0)
@@ -610,9 +628,9 @@ def test_download_content_many_runs_in_parallel(
 
     mock_download_content.side_effect = slow_download
     items = [
-        ("https://example.com/a", setup_core / "a"),
-        ("https://example.com/b", setup_core / "b"),
-        ("https://example.com/c", setup_core / "c"),
+        external_files.RemoteFile("https://example.com/a", setup_core / "a"),
+        external_files.RemoteFile("https://example.com/b", setup_core / "b"),
+        external_files.RemoteFile("https://example.com/c", setup_core / "c"),
     ]
     external_files.download_content_many(items, max_workers=4)
     assert mock_download_content.call_count == 3
@@ -625,15 +643,20 @@ def test_download_content_many_propagates_single_error(
     it in a `MultipleInvalid` that the caller would have to unpack.
     """
 
-    def fake_download(url: str, path: Path, timeout: int) -> bytes:
+    def fake_download(
+        url: str,
+        path: Path,
+        *args: Any,
+        **kwargs: Any,
+    ) -> bytes:
         if url.endswith("bad"):
             raise Invalid(f"could not download {url}")
         return b""
 
     mock_download_content.side_effect = fake_download
     items = [
-        ("https://example.com/ok", setup_core / "ok"),
-        ("https://example.com/bad", setup_core / "bad"),
+        external_files.RemoteFile("https://example.com/ok", setup_core / "ok"),
+        external_files.RemoteFile("https://example.com/bad", setup_core / "bad"),
     ]
     with pytest.raises(Invalid, match="could not download") as exc_info:
         external_files.download_content_many(items)
@@ -648,16 +671,21 @@ def test_download_content_many_aggregates_multiple_errors(
     them one network round-trip at a time.
     """
 
-    def fake_download(url: str, path: Path, timeout: int) -> bytes:
+    def fake_download(
+        url: str,
+        path: Path,
+        *args: Any,
+        **kwargs: Any,
+    ) -> bytes:
         if url.endswith("ok"):
             return b""
         raise Invalid(f"could not download {url}")
 
     mock_download_content.side_effect = fake_download
     items = [
-        ("https://example.com/ok", setup_core / "ok"),
-        ("https://example.com/bad1", setup_core / "bad1"),
-        ("https://example.com/bad2", setup_core / "bad2"),
+        external_files.RemoteFile("https://example.com/ok", setup_core / "ok"),
+        external_files.RemoteFile("https://example.com/bad1", setup_core / "bad1"),
+        external_files.RemoteFile("https://example.com/bad2", setup_core / "bad2"),
     ]
     with pytest.raises(MultipleInvalid) as exc_info:
         external_files.download_content_many(items)
@@ -678,9 +706,9 @@ def test_download_content_many_dedupes_by_path(
     """
     path = setup_core / "shared"
     items = [
-        ("https://example.com/a", path),
-        ("https://example.com/b", path),
-        ("https://example.com/a", path),
+        external_files.RemoteFile("https://example.com/a", path),
+        external_files.RemoteFile("https://example.com/b", path),
+        external_files.RemoteFile("https://example.com/a", path),
     ]
     external_files.download_content_many(items)
     assert mock_download_content.call_count == 1
@@ -695,8 +723,8 @@ def test_download_content_many_clamps_invalid_max_workers(
     be clamped up to at least 1 worker.
     """
     items = [
-        ("https://example.com/a", setup_core / "a"),
-        ("https://example.com/b", setup_core / "b"),
+        external_files.RemoteFile("https://example.com/a", setup_core / "a"),
+        external_files.RemoteFile("https://example.com/b", setup_core / "b"),
     ]
     external_files.download_content_many(items, max_workers=0)
     assert mock_download_content.call_count == 2
@@ -724,8 +752,8 @@ def test_download_web_files_in_config_filters_and_dispatches(
     assert result is config
     mock_download_content_many.assert_called_once()
     assert list(mock_download_content_many.call_args[0][0]) == [
-        ("https://example.com/a", setup_core / "a"),
-        ("https://example.com/c", setup_core / "c"),
+        external_files.RemoteFile("https://example.com/a", setup_core / "a"),
+        external_files.RemoteFile("https://example.com/c", setup_core / "c"),
     ]
 
 
@@ -799,3 +827,264 @@ def test_download_content_atomic_write_no_partial_on_failure(
     # into the cache directory either way.
     leftover_tmps = list(setup_core.glob("tmp*"))
     assert leftover_tmps == []
+
+
+def test_download_content_memoizes_fresh_path(
+    mock_has_remote_file_changed: MagicMock,
+    mock_requests_get: MagicMock,
+    setup_core: Path,
+) -> None:
+    """A path downloaded once this run skips all network on later calls."""
+    test_file = setup_core / "memo.txt"
+
+    mock_has_remote_file_changed.return_value = True
+    mock_response = MagicMock()
+    mock_response.content = b"fresh content"
+    mock_response.headers = {}
+    mock_requests_get.return_value = mock_response
+
+    url = "https://example.com/file.txt"
+    assert external_files.download_content(url, test_file) == b"fresh content"
+    assert external_files.download_content(url, test_file) == b"fresh content"
+
+    mock_has_remote_file_changed.assert_called_once()
+    mock_requests_get.assert_called_once()
+
+
+def test_download_content_memo_revalidates_deleted_file(
+    mock_has_remote_file_changed: MagicMock,
+    mock_requests_get: MagicMock,
+    setup_core: Path,
+) -> None:
+    """A memoized path whose file vanished is downloaded again."""
+    test_file = setup_core / "memo.txt"
+
+    mock_has_remote_file_changed.return_value = True
+    mock_response = MagicMock()
+    mock_response.content = b"fresh content"
+    mock_response.headers = {}
+    mock_requests_get.return_value = mock_response
+
+    url = "https://example.com/file.txt"
+    external_files.download_content(url, test_file)
+    test_file.unlink()
+    external_files.download_content(url, test_file)
+
+    assert mock_requests_get.call_count == 2
+
+
+def test_download_content_failure_fails_fast_on_retry(
+    mock_has_remote_file_changed: MagicMock,
+    mock_requests_get: MagicMock,
+    setup_core: Path,
+) -> None:
+    """A failed download is remembered; a retry raises without network."""
+    test_file = setup_core / "memo.txt"
+
+    mock_has_remote_file_changed.return_value = True
+    mock_requests_get.side_effect = requests.exceptions.RequestException("boom")
+
+    url = "https://example.com/file.txt"
+    with pytest.raises(Invalid, match="boom"):
+        external_files.download_content(url, test_file)
+    with pytest.raises(Invalid, match="boom"):
+        external_files.download_content(url, test_file)
+
+    mock_requests_get.assert_called_once()
+
+
+def test_download_content_failed_path_revalidates_when_file_appears(
+    mock_has_remote_file_changed: MagicMock,
+    mock_requests_get: MagicMock,
+    setup_core: Path,
+) -> None:
+    """A recorded failure is dropped once the file exists on disk."""
+    test_file = setup_core / "memo.txt"
+
+    mock_has_remote_file_changed.return_value = True
+    mock_requests_get.side_effect = requests.exceptions.RequestException("boom")
+
+    url = "https://example.com/file.txt"
+    with pytest.raises(Invalid):
+        external_files.download_content(url, test_file)
+
+    # Another writer produced the file; the cached failure no longer applies
+    # and the network error now falls back to the on-disk copy.
+    test_file.write_bytes(b"appeared")
+    assert external_files.download_content(url, test_file) == b"appeared"
+
+
+def test_download_content_network_error_fallback_memoizes(
+    mock_has_remote_file_changed: MagicMock,
+    mock_requests_get: MagicMock,
+    setup_core: Path,
+) -> None:
+    """Falling back to a cached file memoizes, so a flaky host is hit once."""
+    test_file = setup_core / "memo.txt"
+    test_file.write_bytes(b"cached content")
+
+    mock_has_remote_file_changed.return_value = True
+    mock_requests_get.side_effect = requests.exceptions.RequestException("boom")
+
+    url = "https://example.com/file.txt"
+    assert external_files.download_content(url, test_file) == b"cached content"
+    assert external_files.download_content(url, test_file) == b"cached content"
+
+    mock_requests_get.assert_called_once()
+
+
+def test_download_content_not_changed_uses_cache(
+    mock_has_remote_file_changed: MagicMock,
+    mock_requests_get: MagicMock,
+    setup_core: Path,
+) -> None:
+    """A 304 not-changed check serves the cached file without a GET."""
+    test_file = setup_core / "cached.txt"
+    test_file.write_bytes(b"cached content")
+    mock_has_remote_file_changed.return_value = False
+
+    url = "https://example.com/file.txt"
+    assert external_files.download_content(url, test_file) == b"cached content"
+
+    mock_requests_get.assert_not_called()
+
+
+def test_head_failure_fallback_is_stale_not_fresh(
+    mock_requests_head: MagicMock,
+    mock_requests_get: MagicMock,
+    setup_core: Path,
+) -> None:
+    """A HEAD network failure serves the copy once and memoizes it as stale."""
+    test_file = setup_core / "cached.txt"
+    test_file.write_bytes(b"cached content")
+
+    mock_requests_head.side_effect = requests.exceptions.RequestException("boom")
+
+    url = "https://example.com/file.txt"
+    assert external_files.download_content(url, test_file) == b"cached content"
+    assert external_files.download_content(url, test_file) == b"cached content"
+
+    mock_requests_head.assert_called_once()
+    mock_requests_get.assert_not_called()
+
+
+def test_allow_stale_false_rejects_unverified_copy(
+    mock_has_remote_file_changed: MagicMock,
+    mock_requests_get: MagicMock,
+    setup_core: Path,
+) -> None:
+    """allow_stale=False raises instead of building from an unverified copy."""
+    test_file = setup_core / "cached.txt"
+    test_file.write_bytes(b"cached content")
+
+    mock_has_remote_file_changed.return_value = True
+    mock_requests_get.side_effect = requests.exceptions.RequestException("boom")
+
+    url = "https://example.com/file.txt"
+    with pytest.raises(Invalid, match="Could not download"):
+        external_files.download_content(url, test_file, allow_stale=False)
+
+    # A strict caller gets its own attempt at the network rather than
+    # inheriting the stale memo's verdict.
+    with pytest.raises(Invalid, match="Could not download"):
+        external_files.download_content(url, test_file, allow_stale=False)
+    assert mock_requests_get.call_count == 2
+
+    # A caller that tolerates stale copies still gets the cached bytes.
+    assert external_files.download_content(url, test_file) == b"cached content"
+
+
+def test_allow_stale_false_rejects_head_failure_fallback(
+    mock_requests_head: MagicMock,
+    mock_requests_get: MagicMock,
+    setup_core: Path,
+) -> None:
+    """allow_stale=False also rejects a copy the HEAD could not confirm."""
+    test_file = setup_core / "cached.txt"
+    test_file.write_bytes(b"cached content")
+
+    mock_requests_head.side_effect = requests.exceptions.RequestException("boom")
+
+    url = "https://example.com/file.txt"
+    with pytest.raises(Invalid, match="cannot be verified"):
+        external_files.download_content(url, test_file, allow_stale=False)
+    mock_requests_get.assert_not_called()
+
+
+def test_download_content_many_forwards_per_file_allow_stale(
+    mock_download_content: MagicMock, setup_core: Path
+) -> None:
+    """Each RemoteFile's own allow_stale reaches download_content."""
+    files = [
+        external_files.RemoteFile("https://example.com/a", setup_core / "a"),
+        external_files.RemoteFile(
+            "https://example.com/b", setup_core / "b", allow_stale=False
+        ),
+    ]
+    external_files.download_content_many(files)
+    forwarded = {
+        call.args[1]: call.kwargs["allow_stale"]
+        for call in mock_download_content.call_args_list
+    }
+    assert forwarded == {setup_core / "a": True, setup_core / "b": False}
+
+
+def test_download_content_many_dedupe_keeps_strictest(
+    mock_download_content: MagicMock, setup_core: Path
+) -> None:
+    """A strict duplicate wins over a permissive one for the same path."""
+    path = setup_core / "fw.bin"
+    files = [
+        external_files.RemoteFile("https://example.com/fw", path, allow_stale=False),
+        external_files.RemoteFile("https://example.com/fw", path),
+    ]
+    external_files.download_content_many(files)
+    mock_download_content.assert_called_once()
+    assert mock_download_content.call_args.kwargs["allow_stale"] is False
+
+
+def test_successful_head_revalidation_clears_stale(
+    mock_requests_head: MagicMock,
+    mock_requests_get: MagicMock,
+    setup_core: Path,
+) -> None:
+    """A confirmed 304 supersedes an earlier failed revalidation."""
+    test_file = setup_core / "cached.txt"
+    test_file.write_bytes(b"cached content")
+
+    ok_304 = MagicMock(status_code=304, headers={})
+    mock_requests_head.side_effect = [
+        requests.exceptions.RequestException("blip"),
+        ok_304,
+    ]
+
+    url = "https://example.com/file.txt"
+    assert external_files.download_content(url, test_file) == b"cached content"
+    # The stale memo short-circuits tolerant callers; a strict caller
+    # triggers a fresh HEAD, which now succeeds and clears the marker.
+    assert (
+        external_files.download_content(url, test_file, allow_stale=False)
+        == b"cached content"
+    )
+    # Verified now: served from the fresh memo with no more network.
+    assert external_files.download_content(url, test_file) == b"cached content"
+    assert mock_requests_head.call_count == 2
+    mock_requests_get.assert_not_called()
+
+
+def test_failed_path_replay_names_the_other_url(
+    mock_has_remote_file_changed: MagicMock,
+    mock_requests_get: MagicMock,
+    setup_core: Path,
+) -> None:
+    """A shared cache path replays the failure naming the original URL."""
+    test_file = setup_core / "shared.bin"
+
+    mock_has_remote_file_changed.return_value = True
+    mock_requests_get.side_effect = requests.exceptions.RequestException("boom")
+
+    with pytest.raises(Invalid, match="first-url"):
+        external_files.download_content("https://example.com/first-url", test_file)
+    with pytest.raises(Invalid, match="earlier download of.*first-url"):
+        external_files.download_content("https://example.com/second-url", test_file)
+    mock_requests_get.assert_called_once()