[core] Cache validated config to skip re-validation on upload/logs (#16381)

esphome/__main__.py

@@ -2434,10 +2434,28 @@ def run_esphome(argv):
     # Commands that don't need fresh external components: logs just connects
     # to the device, and clean is about to delete the build directory.
     skip_external = args.command in ("logs", "clean")
-    config = read_config(
+    command_line_substitutions = dict(args.substitution) if args.substitution else {}
-        dict(args.substitution) if args.substitution else {},
+
-        skip_external_update=skip_external,
+    # Fast path for upload/logs: reuse the validated-config cache the
-    )
+    # last compile wrote. Falls back to read_config when missing/stale.
+    # Skipped when -s overrides are passed, since the cache was written
+    # against the previous substitution set.
+    config: ConfigType | None = None
+    if args.command in ("upload", "logs") and not command_line_substitutions:
+        from esphome.compiled_config import load_compiled_config
+
+        config = load_compiled_config(conf_path)
+        if config is not None:
+            _LOGGER.info(
+                "Loaded validated config cache for %s, skipping validation.",
+                conf_path.name,
+            )
+
+    if config is None:
+        config = read_config(
+            command_line_substitutions,
+            skip_external_update=skip_external,
+        )
     if config is None:
         return 2
     CORE.config = config

esphome/compiled_config.py

@@ -0,0 +1,76 @@
+"""Validated-config cache for the upload/logs fast path.
+
+compile dumps the validated config to <data_dir>/storage/<file>.validated.yaml;
+the next upload/logs for that YAML reuses it instead of running the full
+read_config pipeline. YAML round-trip (yaml_util.dump/load_yaml) keeps
+!lambda/!include/IDs/paths intact; mtime gates staleness.
+"""
+
+from __future__ import annotations
+
+import logging
+from pathlib import Path
+
+from esphome.core import CORE
+from esphome.helpers import write_file
+from esphome.storage_json import StorageJSON, ext_storage_path
+from esphome.types import ConfigType
+
+_LOGGER = logging.getLogger(__name__)
+
+
+def compiled_config_path(config_filename: str) -> Path:
+    """Path to the cached validated config alongside the storage sidecar."""
+    return CORE.data_dir / "storage" / f"{config_filename}.validated.yaml"
+
+
+def _cache_is_fresh(cache_path: Path, source_path: Path) -> bool:
+    """True iff the cache file exists and isn't older than the source."""
+    try:
+        return cache_path.stat().st_mtime >= source_path.stat().st_mtime
+    except OSError:
+        return False
+
+
+def save_compiled_config(config: ConfigType) -> None:
+    """Write the validated-config cache. Always-write so mtime stays fresh.
+
+    Mode 0600 because show_secrets=True resolves !secret inline.
+    Failures are non-fatal: the fast path falls back to read_config.
+    """
+    from esphome import yaml_util
+
+    try:
+        rendered = yaml_util.dump(config, show_secrets=True)
+        write_file(compiled_config_path(CORE.config_filename), rendered, private=True)
+    except Exception as err:  # pylint: disable=broad-except
+        _LOGGER.debug("Skipping compiled config cache write: %s", err)
+
+
+def load_compiled_config(conf_path: Path) -> ConfigType | None:
+    """Load the cached validated config and apply storage metadata to CORE.
+
+    Returns None (caller falls back to read_config) when the cache is
+    missing, older than the source YAML, unparseable, or the sidecar
+    is incomplete.
+    """
+    cache_path = compiled_config_path(conf_path.name)
+    if not _cache_is_fresh(cache_path, conf_path):
+        return None
+
+    from esphome import yaml_util
+
+    try:
+        config = yaml_util.load_yaml(cache_path, clear_secrets=False)
+    except Exception:  # pylint: disable=broad-except
+        return None
+
+    storage = StorageJSON.load(ext_storage_path(conf_path.name))
+    if storage is None:
+        return None
+    # apply_to_core assumes a real compile wrote the sidecar; wizard-only
+    # sidecars leave both of these unset and can't drive upload/logs.
+    if not storage.core_platform and not storage.target_platform:
+        return None
+    storage.apply_to_core()
+    return config

esphome/storage_json.py

@@ -8,7 +8,13 @@ import os
 from pathlib import Path
 
 from esphome import const
-from esphome.const import CONF_DISABLED, CONF_MDNS
+from esphome.const import (
+    CONF_DISABLED,
+    CONF_MDNS,
+    KEY_CORE,
+    KEY_TARGET_FRAMEWORK,
+    KEY_TARGET_PLATFORM,
+)
 from esphome.core import CORE
 from esphome.helpers import write_file_if_changed
 from esphome.types import CoreType
@@ -256,6 +262,22 @@ class StorageJSON:
         except Exception:  # pylint: disable=broad-except
             return None
 
+    def apply_to_core(self) -> None:
+        """Populate CORE with the metadata upload/logs read.
+
+        Inverse of :meth:`from_esphome_core`. Keep paired -- a new
+        attribute upload/logs needs has to be captured there too.
+        Validator-only fields (loaded_integrations/platforms,
+        friendly_name) are skipped; the fast path doesn't run
+        validation and CORE.__init__ defaults them.
+        """
+        CORE.name = self.name
+        CORE.build_path = self.build_path
+        CORE.data[KEY_CORE] = {
+            KEY_TARGET_PLATFORM: self.core_platform or self.target_platform.lower(),
+            KEY_TARGET_FRAMEWORK: self.framework,
+        }
+
     def __eq__(self, o) -> bool:
         return isinstance(o, StorageJSON) and self.as_dict() == o.as_dict()
 

esphome/writer.py

@@ -7,6 +7,7 @@ import re
 import time
 
 from esphome import loader
+from esphome.compiled_config import save_compiled_config
 from esphome.config import iter_component_configs, iter_components
 from esphome.const import (
     HEADER_FILE_EXTENSIONS,
@@ -109,6 +110,11 @@ def update_storage_json() -> None:
     path = storage_path()
     old = StorageJSON.load(path)
     new = StorageJSON.from_esphome_core(CORE, old)
+
+    # Refresh the cache upload/logs read on the next call.
+    if CORE.config is not None:
+        save_compiled_config(CORE.config)
+
     if old == new:
         return
 

tests/unit_tests/test_compiled_config.py

@@ -0,0 +1,282 @@
+"""Tests for the validated-config cache used by upload/logs."""
+
+from __future__ import annotations
+
+import json
+import os
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+
+from esphome.__main__ import run_esphome
+from esphome.compiled_config import (
+    compiled_config_path,
+    load_compiled_config,
+    save_compiled_config,
+)
+from esphome.const import (
+    CONF_API,
+    CONF_ESPHOME,
+    CONF_NAME,
+    KEY_CORE,
+    KEY_TARGET_FRAMEWORK,
+    KEY_TARGET_PLATFORM,
+)
+from esphome.core import CORE
+
+_VALIDATED_CONFIG_YAML = """\
+esphome:
+  name: lite_test
+  friendly_name: Lite Test Device
+esp32:
+  board: nodemcu-32s
+logger:
+  baud_rate: 115200
+api:
+  port: 6053
+  encryption:
+    key: 6dGhpcyBpcyBhIHRlc3Q=
+ota:
+  - platform: esphome
+    port: 3232
+    password: secret
+wifi:
+  ssid: ssid
+  use_address: 192.168.1.42
+"""
+
+
+def _write_storage(storage_path: Path) -> None:
+    """Write a vanilla StorageJSON sidecar for the cache tests."""
+    storage_path.parent.mkdir(parents=True, exist_ok=True)
+    data = {
+        "storage_version": 1,
+        "name": "lite_test",
+        "friendly_name": "Lite Test Device",
+        "comment": None,
+        "esphome_version": "2026.1.0",
+        "src_version": 1,
+        "address": "192.168.1.42",
+        "web_port": None,
+        "esp_platform": "ESP32",
+        "build_path": "/build/lite_test",
+        "firmware_bin_path": "/build/lite_test/firmware.bin",
+        "loaded_integrations": ["api", "logger", "ota", "wifi"],
+        "loaded_platforms": [],
+        "no_mdns": False,
+        "framework": "arduino",
+        "core_platform": "esp32",
+    }
+    storage_path.write_text(json.dumps(data))
+
+
+def _write_cache(cache_path: Path, body: str = _VALIDATED_CONFIG_YAML) -> Path:
+    """Write the cache file and return it."""
+    cache_path.parent.mkdir(parents=True, exist_ok=True)
+    cache_path.write_text(body)
+    return cache_path
+
+
+def _set_cache_mtime(cache_path: Path, yaml_path: Path, *, offset: int) -> None:
+    """Force the cache file's mtime relative to the source YAML.
+
+    Positive offset → cache is fresh. Negative → cache is stale.
+    """
+    yaml_stat = yaml_path.stat()
+    os.utime(cache_path, (yaml_stat.st_atime, yaml_stat.st_mtime + offset))
+
+
+@pytest.fixture
+def fresh_cache_files(tmp_path: Path) -> Path:
+    """YAML + StorageJSON + cache, all consistent and fresh."""
+    yaml_path = tmp_path / "lite_test.yaml"
+    yaml_path.write_text("esphome:\n  name: lite_test\n")
+    CORE.config_path = yaml_path
+
+    storage_dir = tmp_path / ".esphome" / "storage"
+    _write_storage(storage_dir / "lite_test.yaml.json")
+    cache = _write_cache(storage_dir / "lite_test.yaml.validated.yaml")
+    _set_cache_mtime(cache, yaml_path, offset=5)
+
+    return yaml_path
+
+
+def test_compiled_config_path_lives_alongside_sidecar(setup_core: Path) -> None:
+    """The cache file shape is predictable from the YAML filename."""
+    path = compiled_config_path("device.yaml")
+    assert path.name == "device.yaml.validated.yaml"
+    assert path.parent.name == "storage"
+
+
+def test_load_compiled_config_happy_path(fresh_cache_files: Path) -> None:
+    """Fresh cache + sidecar → returns config and populates CORE."""
+    config = load_compiled_config(fresh_cache_files)
+
+    assert config is not None
+    assert config[CONF_ESPHOME][CONF_NAME] == "lite_test"
+    assert config[CONF_API]["encryption"]["key"] == "6dGhpcyBpcyBhIHRlc3Q="
+    assert config["ota"][0]["password"] == "secret"
+
+    # apply_to_core populated exactly what upload/logs read off CORE.
+    assert CORE.name == "lite_test"
+    assert CORE.build_path == Path("/build/lite_test")
+    assert CORE.data[KEY_CORE][KEY_TARGET_PLATFORM] == "esp32"
+    assert CORE.data[KEY_CORE][KEY_TARGET_FRAMEWORK] == "arduino"
+
+
+@pytest.mark.parametrize(
+    "scenario",
+    ["missing_cache", "stale_cache", "corrupt_cache", "missing_sidecar"],
+)
+def test_load_compiled_config_falls_back(tmp_path: Path, scenario: str) -> None:
+    """All non-happy cases return None so the caller falls back."""
+    yaml_path = tmp_path / "lite_test.yaml"
+    yaml_path.write_text("esphome:\n  name: lite_test\n")
+    CORE.config_path = yaml_path
+    storage_dir = tmp_path / ".esphome" / "storage"
+    cache_path = storage_dir / "lite_test.yaml.validated.yaml"
+    sidecar_path = storage_dir / "lite_test.yaml.json"
+
+    if scenario == "missing_cache":
+        pass  # no cache, no sidecar
+    elif scenario == "stale_cache":
+        _write_storage(sidecar_path)
+        _set_cache_mtime(_write_cache(cache_path), yaml_path, offset=-60)
+    elif scenario == "corrupt_cache":
+        _write_storage(sidecar_path)
+        _set_cache_mtime(
+            _write_cache(cache_path, "not: valid: yaml: ["), yaml_path, offset=5
+        )
+    elif scenario == "missing_sidecar":
+        # Cache fresh + parseable, but no StorageJSON → can't populate CORE.
+        _set_cache_mtime(_write_cache(cache_path), yaml_path, offset=5)
+
+    assert load_compiled_config(yaml_path) is None
+
+
+@pytest.mark.parametrize("command", ["upload", "logs"])
+def test_run_esphome_upload_and_logs_use_cache_when_fresh(
+    command: str,
+    fresh_cache_files: Path,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """upload/logs skip read_config() when the cache is fresh."""
+    captured: dict = {}
+
+    def _stub(_args, config):
+        captured["config"] = config
+        return 0
+
+    with (
+        caplog.at_level("INFO", logger="esphome.__main__"),
+        patch("esphome.__main__.read_config") as mock_read,
+        patch.dict("esphome.__main__.POST_CONFIG_ACTIONS", {command: _stub}),
+    ):
+        assert run_esphome(["esphome", command, str(fresh_cache_files)]) == 0
+
+    mock_read.assert_not_called()
+    assert captured["config"][CONF_ESPHOME][CONF_NAME] == "lite_test"
+    assert captured["config"][CONF_API]["encryption"]["key"] == "6dGhpcyBpcyBhIHRlc3Q="
+    # The success-branch log line is part of the patch; assert on it so
+    # branch coverage stays unambiguous in CI.
+    assert "Loaded validated config cache" in caplog.text
+
+
+@pytest.mark.parametrize("command", ["upload", "logs"])
+def test_run_esphome_upload_and_logs_fall_back_when_no_cache(
+    tmp_path: Path, command: str
+) -> None:
+    """Without a cache, the dispatcher falls back to read_config()."""
+    yaml_path = tmp_path / "lite_test.yaml"
+    yaml_path.write_text("esphome:\n  name: lite_test\n")
+
+    with (
+        patch("esphome.__main__.read_config", return_value=None) as mock_read,
+        patch.dict(
+            "esphome.__main__.POST_CONFIG_ACTIONS",
+            {command: lambda args, config: 0},
+        ),
+    ):
+        assert run_esphome(["esphome", command, str(yaml_path)]) == 2
+
+    mock_read.assert_called_once()
+
+
+def test_run_esphome_upload_with_substitution_skips_cache(
+    fresh_cache_files: Path,
+) -> None:
+    """`-s key value` forces a fresh validation -- the cache was written
+    against the prior substitution set, so reusing it would silently
+    ignore the override."""
+    with (
+        patch("esphome.__main__.read_config", return_value=None) as mock_read,
+        patch.dict(
+            "esphome.__main__.POST_CONFIG_ACTIONS",
+            {"upload": lambda args, config: 0},
+        ),
+    ):
+        run_esphome(["esphome", "-s", "var", "val", "upload", str(fresh_cache_files)])
+
+    mock_read.assert_called_once()
+
+
+def test_run_esphome_compile_does_not_use_cache(fresh_cache_files: Path) -> None:
+    """The compile subcommand always re-validates -- it's what writes the cache."""
+    with (
+        patch("esphome.__main__.read_config", return_value=None) as mock_read,
+        patch.dict(
+            "esphome.__main__.POST_CONFIG_ACTIONS",
+            {"compile": lambda args, config: 0},
+        ),
+    ):
+        run_esphome(["esphome", "compile", str(fresh_cache_files)])
+
+    mock_read.assert_called_once()
+
+
+def test_save_compiled_config_writes_cache(tmp_path: Path) -> None:
+    """`save_compiled_config` writes the dumped YAML next to the sidecar."""
+    CORE.config_path = tmp_path / "lite_test.yaml"
+    save_compiled_config({"esphome": {"name": "lite_test"}, "logger": {}})
+
+    cache_path = compiled_config_path("lite_test.yaml")
+    assert cache_path.is_file()
+    body = cache_path.read_text()
+    assert "name: lite_test" in body
+    assert "logger:" in body
+
+
+def test_save_compiled_config_swallows_dump_errors(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """Failures during the dump are non-fatal -- a bad cache just means
+    the next fast path falls back to read_config()."""
+    CORE.config_path = tmp_path / "lite_test.yaml"
+    with patch("esphome.yaml_util.dump", side_effect=RuntimeError("boom")):
+        save_compiled_config({"esphome": {"name": "lite_test"}})
+    assert not compiled_config_path("lite_test.yaml").exists()
+
+
+def test_load_compiled_config_rejects_wizard_only_sidecar(tmp_path: Path) -> None:
+    """A wizard-only sidecar (no compile -- no core_platform / target_platform)
+    can't drive upload/logs, so the fast path falls back."""
+    yaml_path = tmp_path / "lite_test.yaml"
+    yaml_path.write_text("esphome:\n  name: lite_test\n")
+    CORE.config_path = yaml_path
+
+    storage_dir = tmp_path / ".esphome" / "storage"
+    storage_dir.mkdir(parents=True, exist_ok=True)
+    # StorageJSON with both core_platform and target_platform unset.
+    (storage_dir / "lite_test.yaml.json").write_text(
+        '{"storage_version": 1, "name": "lite_test", "friendly_name": null, '
+        '"comment": null, "esphome_version": null, "src_version": 1, '
+        '"address": null, "web_port": null, "esp_platform": null, '
+        '"build_path": null, "firmware_bin_path": null, '
+        '"loaded_integrations": [], "loaded_platforms": [], "no_mdns": false, '
+        '"framework": null, "core_platform": null}'
+    )
+    cache_path = _write_cache(storage_dir / "lite_test.yaml.validated.yaml")
+    _set_cache_mtime(cache_path, yaml_path, offset=5)
+
+    assert load_compiled_config(yaml_path) is None