Revert "[core] Derive CORE state from cached config dict, drop apply_to_core"

This reverts commit cc6245b876.

esphome/compiled_config.py

@@ -13,7 +13,8 @@ paths) round-trip cleanly, and so the small ``StorageJSON`` sidecar
 isn't bloated by configs that can grow past a megabyte. Staleness
 is gated by an mtime check against the source YAML; the loader
 falls back to ``None`` (and the caller to ``read_config()``) whenever
-the cache is missing, stale, unparseable, or structurally incomplete.
+the cache is missing, stale, unparseable, or the companion
+``StorageJSON`` sidecar can't be loaded.
 """
 
 from __future__ import annotations
@@ -21,63 +22,14 @@ from __future__ import annotations
 import logging
 from pathlib import Path
 
-from esphome.const import (
-    CONF_BUILD_PATH,
-    CONF_ESPHOME,
-    CONF_FRAMEWORK,
-    CONF_FRIENDLY_NAME,
-    CONF_NAME,
-    CONF_TYPE,
-    KEY_CORE,
-    KEY_TARGET_FRAMEWORK,
-    KEY_TARGET_PLATFORM,
-)
 from esphome.core import CORE
 from esphome.helpers import write_file_if_changed
+from esphome.storage_json import StorageJSON, ext_storage_path
 from esphome.types import ConfigType
 
 _LOGGER = logging.getLogger(__name__)
 
 
-def _populate_core_from_validated_config(config: ConfigType) -> None:
-    """Set up ``CORE`` from an already-validated config dict.
-
-    Reads from the same config keys ``preload_core_config`` and the
-    target-platform component validator would write into during a full
-    ``read_config`` -- using the cached config dict as the canonical
-    source instead of duplicating the field list in a separate sidecar
-    schema. ``CORE.address`` and ``CORE.web_port`` aren't set here:
-    they're already properties that derive themselves from
-    ``CORE.config`` on access, which the dispatcher assigns next.
-    """
-    from esphome.core.config import _is_target_platform
-
-    esphome_block = config[CONF_ESPHOME]
-    CORE.name = esphome_block[CONF_NAME]
-    CORE.friendly_name = esphome_block.get(CONF_FRIENDLY_NAME)
-    if (build_path := esphome_block.get(CONF_BUILD_PATH)) is not None:
-        CORE.build_path = CORE.data_dir / build_path
-
-    CORE.data.setdefault(KEY_CORE, {})
-    for domain, sub in config.items():
-        if not isinstance(domain, str) or not _is_target_platform(domain):
-            continue
-        CORE.data[KEY_CORE][KEY_TARGET_PLATFORM] = domain
-        # Every platform component follows the
-        # `<platform>.framework.type` convention to express which
-        # framework the binary was built against (esp-idf, arduino,
-        # zephyr). A platform without a framework block (host) just
-        # leaves KEY_TARGET_FRAMEWORK unset, matching what
-        # `read_config` would do.
-        if (
-            isinstance(sub, dict)
-            and isinstance((framework := sub.get(CONF_FRAMEWORK)), dict)
-            and (framework_type := framework.get(CONF_TYPE)) is not None
-        ):
-            CORE.data[KEY_CORE][KEY_TARGET_FRAMEWORK] = framework_type
-        break
-
-
 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"
@@ -112,19 +64,17 @@ def save_compiled_config(config: ConfigType) -> None:
 
 
 def load_compiled_config(conf_path: Path) -> ConfigType | None:
-    """Load the cached validated config and set up ``CORE``.
+    """Load the cached validated config and apply storage metadata to CORE.
 
-    Single entry point for the ``upload`` / ``logs`` fast path. Returns
+    Single entry point for the ``upload`` / ``logs`` fast path: loads
+    the cache, applies the ``StorageJSON`` sidecar's platform / build
+    metadata to ``CORE``, and returns the config dict. Returns
     ``None`` (so the caller falls back to ``read_config``) when the
     cache is missing, older than the source YAML, unparseable, or
-    structurally incomplete. The mtime check catches the common "user
-    edited the YAML and forgot to recompile" case; deeper drift (an
-    edited ``!include`` whose parent YAML mtime didn't change) is the
-    user's responsibility.
-
-    CORE state is derived from the cached config dict directly --
-    same source ``read_config`` uses -- so there's no separate sidecar
-    schema to keep in sync.
+    when the sidecar can't be loaded. The mtime check catches the
+    common "user edited the YAML and forgot to recompile" case;
+    deeper drift (an edited ``!include`` whose parent YAML mtime
+    didn't change) is the user's responsibility.
     """
     cache_path = compiled_config_path(conf_path.name)
     if not _cache_is_fresh(cache_path, conf_path):
@@ -139,11 +89,8 @@ def load_compiled_config(conf_path: Path) -> ConfigType | None:
     except Exception:  # pylint: disable=broad-except
         return None
 
-    if not isinstance(config, dict) or CONF_ESPHOME not in config:
-        return None
-
-    try:
-        _populate_core_from_validated_config(config)
-    except (KeyError, TypeError):
+    storage = StorageJSON.load(ext_storage_path(conf_path.name))
+    if storage is None:
         return None
+    storage.apply_to_core()
     return config

esphome/storage_json.py

@@ -256,6 +256,33 @@ class StorageJSON:
         except Exception:  # pylint: disable=broad-except
             return None
 
+    def apply_to_core(self) -> None:
+        """Populate ``CORE`` from this sidecar.
+
+        Used by the ``--from-storage-json`` fast path in
+        ``esphome upload`` / ``esphome logs``: those subcommands read
+        a handful of ``CORE`` attributes (``target_platform``,
+        ``build_path``, ``name``, ``loaded_integrations``) that the
+        normal flow populates during ``read_config``. Lifting them off
+        the sidecar lets us skip the validation pass entirely.
+        """
+        from esphome.const import KEY_CORE, KEY_TARGET_FRAMEWORK, KEY_TARGET_PLATFORM
+
+        CORE.name = self.name
+        CORE.friendly_name = self.friendly_name
+        CORE.build_path = self.build_path
+        CORE.loaded_integrations = set(self.loaded_integrations)
+        CORE.loaded_platforms = set(self.loaded_platforms)
+
+        core_platform = self.core_platform or (
+            self.target_platform.lower() if self.target_platform else None
+        )
+        CORE.data.setdefault(KEY_CORE, {})
+        if core_platform is not None:
+            CORE.data[KEY_CORE][KEY_TARGET_PLATFORM] = core_platform
+        if self.framework is not None:
+            CORE.data[KEY_CORE][KEY_TARGET_FRAMEWORK] = self.framework
+
     def __eq__(self, o) -> bool:
         return isinstance(o, StorageJSON) and self.as_dict() == o.as_dict()
 

tests/unit_tests/test_compiled_config.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import json
 import os
 from pathlib import Path
 from unittest.mock import patch
@@ -24,11 +25,8 @@ _VALIDATED_CONFIG_YAML = """\
 esphome:
   name: lite_test
   friendly_name: Lite Test Device
-  build_path: build/lite_test
 esp32:
   board: nodemcu-32s
-  framework:
-    type: arduino
 logger:
   baud_rate: 115200
 api:
@@ -45,6 +43,30 @@ wifi:
 """
 
 
+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)
@@ -63,14 +85,14 @@ def _set_cache_mtime(cache_path: Path, yaml_path: Path, *, offset: int) -> None:
 
 @pytest.fixture
 def fresh_cache_files(tmp_path: Path) -> Path:
-    """YAML + cache, both consistent and fresh."""
+    """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
 
-    cache = _write_cache(
-        tmp_path / ".esphome" / "storage" / "lite_test.yaml.validated.yaml"
-    )
+    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
@@ -84,7 +106,7 @@ def test_compiled_config_path_lives_alongside_sidecar(setup_core: Path) -> None:
 
 
 def test_load_compiled_config_happy_path(fresh_cache_files: Path) -> None:
-    """Fresh cache → returns the config and populates CORE from it."""
+    """Fresh cache + sidecar → returns config and populates CORE."""
     config = load_compiled_config(fresh_cache_files)
 
     assert config is not None
@@ -92,37 +114,40 @@ def test_load_compiled_config_happy_path(fresh_cache_files: Path) -> None:
     assert config[CONF_API]["encryption"]["key"] == "6dGhpcyBpcyBhIHRlc3Q="
     assert config["ota"][0]["password"] == "secret"
 
-    # CORE state derives from the cached config dict -- same source
-    # `read_config` uses, no separate sidecar schema.
+    # apply_to_core ran as part of the orchestration.
     assert CORE.name == "lite_test"
-    assert CORE.friendly_name == "Lite Test Device"
-    assert CORE.build_path == CORE.data_dir / "build/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"
+    assert "api" in CORE.loaded_integrations
 
 
 @pytest.mark.parametrize(
     "scenario",
-    ["missing_cache", "stale_cache", "corrupt_cache", "missing_esphome_block"],
+    ["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
-    cache_path = tmp_path / ".esphome" / "storage" / "lite_test.yaml.validated.yaml"
+    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
+        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_esphome_block":
-        # Parseable YAML but no esphome: block -- can't populate CORE.
-        _set_cache_mtime(_write_cache(cache_path, "logger:\n"), 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