Merge remote-tracking branch 'origin/dev' into binary-sensor-lazy-integration-imports

# Conflicts:
#	esphome/components/binary_sensor/__init__.py
#	esphome/components/web_server/__init__.py
#	esphome/components/zigbee/__init__.py
#	esphome/components/zigbee/const.py
#	script/ci-custom.py
#	tests/unit_tests/core/test_entity_helpers.py

script/analyze_component_buses.py

@@ -83,6 +83,7 @@ ISOLATED_COMPONENTS = {
     "animation": "Has display lambda in common.yaml that requires existing display platform - breaks when merged without display",
     "cdc_acm_uart": "Depends on tinyusb which conflicts with usb_host",
     "esphome": "Defines devices/areas in esphome: section that are referenced in other sections - breaks when merged",
+    "ecocomfort2": "Needs esp32_ble with io_capability: none and sc_bond, which a merged esp32_ble: block overrides",
     "ethernet": "Defines ethernet: which conflicts with wifi: used by most components",
     "ethernet_info": "Related to ethernet component which conflicts with wifi",
     "gps": "TinyGPSPlus library declares millis() function that creates ambiguity with ESPHome millis() macro when merged with components using millis() in lambdas",
@@ -97,6 +98,7 @@ ISOLATED_COMPONENTS = {
     "neopixelbus": "RMT type conflict with ESP32 Arduino/ESP-IDF headers (enum vs struct rmt_channel_t)",
     "packages": "cannot merge packages",
     "tinyusb": "Conflicts with usb_host component - cannot be used together",
+    "uart_mux": "Depends on tinyusb which conflicts with usb_host",
     "usb_cdc_acm": "Depends on tinyusb which conflicts with usb_host",
 }
 

script/build_helpers.py

@@ -22,8 +22,8 @@ from esphome.__main__ import command_compile, parse_args
 from esphome.config import validate_config
 from esphome.const import CONF_PLATFORM
 from esphome.core import CORE
+from esphome.host.toolchain import get_elf_path
 from esphome.loader import get_component, get_platform
-from esphome.platformio.toolchain import get_idedata
 from tests.testing_helpers import ComponentManifestOverride, set_testing_manifest
 
 # This must coincide with the version in /platformio.ini
@@ -46,6 +46,8 @@ EXIT_SKIPPED = 1
 EXIT_COMPILE_ERROR = 2
 EXIT_CONFIG_ERROR = 3
 EXIT_NO_EXECUTABLE = 4
+# A test folder with this name would be synced into src/esphome and swept away with the core tree
+CORE_TREE_DIR = "esphome"
 
 # Name of the per-component YAML config file in benchmark directories
 BENCHMARK_YAML_FILENAME = "benchmark.yaml"
@@ -220,7 +222,7 @@ def create_host_config(
     friendly_name: str,
     libraries: str | list[str],
     includes: list[str],
-    platformio_options: dict,
+    build_flags: list[str],
 ) -> dict:
     """Create an ESPHome host configuration for C++ builds.
 
@@ -229,7 +231,7 @@ def create_host_config(
         friendly_name: Human-readable name
         libraries: PlatformIO library specification(s)
         includes: List of include folders for the build
-        platformio_options: Dict of platformio_options to set
+        build_flags: Compiler/linker flags for the build
 
     Returns:
         Configuration dict for ESPHome
@@ -239,7 +241,7 @@ def create_host_config(
             "name": config_name,
             "friendly_name": friendly_name,
             "libraries": libraries,
-            "platformio_options": platformio_options,
+            "build_flags": build_flags,
             "includes": includes,
         },
         HOST_KEY: {},
@@ -403,13 +405,12 @@ def compile_and_get_binary(
         return EXIT_COMPILE_ERROR, None
 
     # After a successful compilation, locate the executable:
-    idedata = get_idedata(config)
-    if idedata is None:
+    program_path = get_elf_path()
+    if not program_path.is_file():
         print("Cannot find executable")
         return EXIT_NO_EXECUTABLE, None
 
-    program_path: str = idedata.raw["prog_path"]
-    return EXIT_OK, program_path
+    return EXIT_OK, str(program_path)
 
 
 def build_and_run(
@@ -419,7 +420,7 @@ def build_and_run(
     config_prefix: str,
     friendly_name: str,
     libraries: str | list[str],
-    platformio_options: dict,
+    build_flags: list[str],
     main_entry: str,
     label: str = "build",
     build_only: bool = False,
@@ -438,7 +439,7 @@ def build_and_run(
         config_prefix: Prefix for the config name (e.g. "cpptests", "cppbench")
         friendly_name: Human-readable name for the config
         libraries: PlatformIO library specification(s)
-        platformio_options: PlatformIO options dict
+        build_flags: Compiler/linker flags for the build
         main_entry: Name of the main entry file (e.g. "main.cpp")
         label: Label for log messages
         build_only: If True, print binary path and return without running
@@ -466,8 +467,19 @@ def build_and_run(
 
     components = sorted(components)
 
-    # Build include list: main entry point + component folders + extra dirs
-    includes: list[str] = [main_entry] + components
+    # Build include list: main entry point + component folders + extra dirs. The core tree
+    # folder is listed file by file, nested files included, since a folder include would
+    # land in src/esphome (see CORE_TREE_DIR)
+    includes: list[str] = [main_entry]
+    for component in components:
+        if component != CORE_TREE_DIR:
+            includes.append(component)
+            continue
+        includes.extend(
+            str(path.relative_to(tests_dir))
+            for path in sorted((tests_dir / component).rglob("*"))
+            if path.suffix in (".cpp", ".h")
+        )
     if extra_include_dirs:
         for d in extra_include_dirs:
             if d.is_dir() and (any(d.glob("*.cpp")) or any(d.glob("*.h"))):
@@ -488,7 +500,7 @@ def build_and_run(
     config_name: str = f"{config_prefix}-" + hash_components(components)
 
     config = create_host_config(
-        config_name, friendly_name, libraries, includes, platformio_options
+        config_name, friendly_name, libraries, includes, build_flags
     )
 
     exit_code, program_path = compile_and_get_binary(

script/build_language_schema.py

@@ -456,7 +456,10 @@ def fix_lvgl_widgets():
 
 
 def get_logger_tags():
-    pattern = re.compile(r'^static const char \*const TAG = "(\w.*)";', re.MULTILINE)
+    pattern = re.compile(
+        r'^(?:static const char \*const TAG = "|ESPHOME_LOG_TAG\(TAG, ")(\w[^"]*)"',
+        re.MULTILINE,
+    )
     # tags not in components dir
     tags = [
         "app",

script/check_idf_py_equivalence.py

@@ -0,0 +1,308 @@
+#!/usr/bin/env python3
+"""Check that a native ESP-IDF build tree is what idf.py itself would produce.
+
+ESPHome runs cmake and ninja directly with the arguments idf.py uses. This
+runs the real ``idf.py reconfigure`` and ``idf.py build`` on a finished tree
+(in place: CMake rejects a moved cache) and fails if either one changes the
+cache, the generated build files or the firmware, or recompiles anything.
+It catches drift when the pinned ESP-IDF version changes what idf.py does.
+The color and ``size`` environment only change what is printed, so those
+parts of the contract are pinned by the unit tests instead.
+"""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import os
+from pathlib import Path
+import re
+import subprocess
+import sys
+
+REPO_ROOT = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(REPO_ROOT))
+
+WATCHED = (
+    "build/CMakeCache.txt",
+    "build/build.ninja",
+    "build/compile_commands.json",
+    "build/project_description.json",
+    "build/config/sdkconfig.h",
+)
+# Only a tree that builds the bootloader has one to watch.
+BOOTLOADER_BIN = "build/bootloader/bootloader.bin"
+OVERRIDE_INEFFECTIVE = (
+    "skip-mode tree built a bootloader; the IDF_BOOTLOADER_OVERRIDE macro in "
+    "esphome/build_gen/espidf.py is not taking effect"
+)
+# Ninja logs whose outputs mean real work when their recorded mtime changes;
+# the bootloader is judged by its own sub-build log when one exists.
+TOP_NINJA_LOG = "build/.ninja_log"
+
+
+def _ninja_logs(skip_bootloader: bool) -> list[str]:
+    """A missing sub-build log stays an error in the mode that requires one."""
+    logs = [TOP_NINJA_LOG]
+    if not skip_bootloader:
+        logs.append("build/bootloader/.ninja_log")
+    return logs
+
+
+BOOTLOADER_BYPRODUCT = re.compile(r"(^|/build/)bootloader/")
+MACRO_CHANGED = (
+    "IDF changed __build_process_project_includes; update "
+    "IDF_BOOTLOADER_OVERRIDE in esphome/build_gen/espidf.py"
+)
+VERSION_DRIFT = (
+    "ESPHome reads ESP-IDF version {ours!r} from {source} but idf_tools reports "
+    "{theirs!r}; update read_idf_version_{source} in esphome/espidf/framework.py"
+)
+LWIP_NOT_EMPTY = (
+    "lwip source {source} compiles to a non-empty object with {option} off; "
+    "drop it from LWIP_EMPTY_SOURCES in esphome/build_gen/espidf.py"
+)
+LWIP_NOTHING_MATCHED = (
+    "no lwip object matched {regex!r} for {option}; the lwip layout or the "
+    "pattern in esphome/build_gen/espidf.py changed"
+)
+LWIP_NM_FAILED = "nm failed on lwip object {source}: {error}"
+WORK_SUFFIXES = (".obj", ".o", ".a", ".elf", ".map", ".bin", ".ld")
+DEFAULT_GLOB = "tests/test_build_components/build/.esphome/build/*"
+
+
+def _digest(path: Path) -> str | None:
+    return hashlib.sha256(path.read_bytes()).hexdigest() if path.is_file() else None
+
+
+def watched(name: str, skip_bootloader: bool) -> list[str]:
+    """The files idf.py must leave untouched for this tree's mode."""
+    files = [*WATCHED, f"build/{name}.elf", f"build/{name}.bin"]
+    if not skip_bootloader:
+        files.append(BOOTLOADER_BIN)
+    return files
+
+
+def _snapshot(
+    build_path: Path, name: str, skip_bootloader: bool
+) -> dict[str, str | None]:
+    return {f: _digest(build_path / f) for f in watched(name, skip_bootloader)}
+
+
+def _ninja_mtimes(build_path: Path, logs: list[str]) -> dict[tuple[str, str], str]:
+    """(log, output) -> recorded mtime; compaction-safe, unlike a line count."""
+    mtimes = {}
+    for name in logs:
+        log = build_path / name
+        lines = log.read_text(encoding="utf-8").splitlines() if log.is_file() else []
+        for fields in (line.split("\t") for line in lines if not line.startswith("#")):
+            if len(fields) >= 4 and (
+                name != TOP_NINJA_LOG or not BOOTLOADER_BYPRODUCT.search(fields[3])
+            ):
+                mtimes[name, fields[3]] = fields[2]
+    return mtimes
+
+
+def _log_problems(
+    build_path: Path, mtimes: dict[tuple[str, str], str], logs: list[str]
+) -> list[str]:
+    """A missing or unparsable ninja log would otherwise compare as unchanged."""
+    problems = []
+    for log in logs:
+        if not (build_path / log).is_file():
+            problems.append(f"missing {log}")
+        elif not any(k[0] == log and k[1].endswith(WORK_SUFFIXES) for k in mtimes):
+            problems.append(f"no build entries parsed from {log}")
+    return problems
+
+
+def _lwip_empty_source_problems(build_path: Path) -> list[str]:
+    """Compile the lwip sources the generated CMakeLists drops; any with
+    symbols is a problem. Leaves the tree configured with every source."""
+    # pylint: disable=protected-access
+    from esphome.build_gen.espidf import LWIP_EMPTY_SOURCES, LWIP_FULL_SOURCES_ENV
+    from esphome.espidf import toolchain
+
+    if (rc := toolchain.run_reconfigure(extra_env={LWIP_FULL_SOURCES_ENV: "1"})) != 0:
+        return [f"CMake configure with every lwip source failed with exit code {rc}"]
+    if rc := toolchain._run_ninja("esp-idf/lwip/liblwip.a", verbose=False, jobs=None):
+        return [f"building every lwip source failed with exit code {rc}"]
+    build = build_path / "build"
+    config = json.loads(
+        (build / "config" / "sdkconfig.json").read_text(encoding="utf-8")
+    )
+    objects = [
+        obj.as_posix().removesuffix(".obj")
+        for obj in (build / "esp-idf" / "lwip").rglob("*.obj")
+    ]
+    nm = toolchain._parse_cmakecache(build / "CMakeCache.txt")["CMAKE_NM"]
+    problems = []
+    for option, regex in LWIP_EMPTY_SOURCES:
+        # Absent means the option is invisible here; the filter keeps those.
+        if config.get(option.removeprefix("CONFIG_"), True):
+            continue
+        matched = [source for source in objects if re.search(regex, source)]
+        if not matched:
+            problems.append(LWIP_NOTHING_MATCHED.format(regex=regex, option=option))
+        for source in matched:
+            name = Path(source).name
+            result = subprocess.run(
+                [nm, "--defined-only", f"{source}.obj"],
+                capture_output=True,
+                text=True,
+                check=False,
+            )
+            if result.returncode:
+                problems.append(
+                    LWIP_NM_FAILED.format(source=name, error=result.stderr.strip())
+                )
+            elif result.stdout.strip():
+                problems.append(LWIP_NOT_EMPTY.format(source=name, option=option))
+    return problems
+
+
+def _setup_core(build_path: Path, description: dict) -> tuple[str, str]:
+    """Point CORE at the tree so ESPHome resolves the same IDF env as the build."""
+    from esphome.components.esp32.const import KEY_ESP32, KEY_IDF_VERSION, KEY_VARIANT
+    import esphome.config_validation as cv
+    from esphome.core import CORE
+
+    name = description["project_name"]
+    version = Path(description["idf_path"]).name
+    CORE.config_path = build_path.parents[2] / f"{name}.yaml"
+    CORE.build_path = build_path
+    CORE.name = name
+    CORE.data[KEY_ESP32] = {
+        KEY_IDF_VERSION: cv.Version.parse(version),
+        KEY_VARIANT: description["target"].upper(),
+    }
+    return name, version
+
+
+def check(build_path: Path) -> list[str]:
+    """Return the problems found in one build tree."""
+    # pylint: disable=protected-access
+    from esphome.build_gen.espidf import idf_macro_matches
+    from esphome.core import CORE
+    from esphome.espidf import framework, toolchain
+
+    description = json.loads(
+        (build_path / "build" / "project_description.json").read_text(encoding="utf-8")
+    )
+    name, version = _setup_core(build_path, description)
+    # Reconfiguring must not flip the tree's bootloader mode: the check
+    # validates the shape the build produced, not this process's flags.
+    skip_bootloader = toolchain.tree_skips_bootloader(build_path / "build")
+    CORE.skip_bootloader = skip_bootloader
+    # A prior tree's memoized decision must not leak into this one.
+    toolchain._cache().skip_bootloader = None
+    idf_path = toolchain._get_idf_path(version)
+    if not idf_macro_matches(idf_path):
+        return [MACRO_CHANGED]
+    # A managed tree always has version.txt, so the header branch is
+    # compared on its own or it would never be exercised here.
+    theirs = framework.idf_tools_version(idf_path)
+    for source, read in (
+        ("txt", framework.read_idf_version_txt),
+        ("header", framework.read_idf_version_header),
+    ):
+        if (ours := read(idf_path)) != theirs:
+            return [VERSION_DRIFT.format(ours=ours, source=source, theirs=theirs)]
+    # ESP-IDF's openthread stamps the configure time into its compile flags;
+    # pin it before the env is cached so both configures get the same value.
+    os.environ["SOURCE_DATE_EPOCH"] = "0"
+    env = toolchain._get_idf_env(version)
+    python = toolchain._get_idf_tool("python")
+    idf_py = toolchain._get_idf_path(version) / "tools" / "idf.py"
+    sdkconfig = build_path / f"sdkconfig.{name}"
+    sdkconfig_args = ["-D", f"SDKCONFIG={sdkconfig}"] if sdkconfig.is_file() else []
+
+    # CMake writes a different build.ninja on a tree's first configure than on
+    # a reconfigure, so the baseline is ESPHome's own reconfigure and build.
+    if (rc := toolchain.run_reconfigure()) != 0:
+        return [f"ESPHome's CMake configure failed with exit code {rc}"]
+    if (rc := toolchain._run_ninja("all", verbose=False, jobs=None)) != 0:
+        return [f"ESPHome's ninja build failed with exit code {rc}"]
+    before = _snapshot(build_path, name, skip_bootloader)
+    logs = _ninja_logs(skip_bootloader)
+    mtimes_before = _ninja_mtimes(build_path, logs)
+    # A moved or renamed output would otherwise compare as "unchanged".
+    problems = [f"missing {f}" for f, digest in before.items() if digest is None]
+    if skip_bootloader and (build_path / BOOTLOADER_BIN).is_file():
+        problems.append(OVERRIDE_INEFFECTIVE)
+    if problems := problems + _log_problems(build_path, mtimes_before, logs):
+        return problems
+    for action in ("reconfigure", "build"):
+        result = subprocess.run(
+            [python, str(idf_py), *sdkconfig_args, action],
+            cwd=build_path,
+            env=env,
+            capture_output=True,
+            text=True,
+            check=False,
+        )
+        if result.returncode != 0:
+            return [f"idf.py {action} failed:\n{result.stdout}{result.stderr}"]
+    after = _snapshot(build_path, name, skip_bootloader)
+    mtimes_after = _ninja_mtimes(build_path, logs)
+    problems = [f"idf.py changed {f}" for f in before if before[f] != after[f]]
+    problems += _log_problems(build_path, mtimes_after, logs)
+    for key in sorted(mtimes_before.keys() | mtimes_after.keys()):
+        log, out = key
+        if not out.endswith(WORK_SUFFIXES):
+            continue
+        if key not in mtimes_after:
+            problems.append(f"idf.py dropped {out} from {log}")
+        elif mtimes_before.get(key) != mtimes_after[key]:
+            problems.append(f"idf.py rebuilt {out}")
+    # Last: it reconfigures the tree, which would otherwise relink above.
+    return problems or _lwip_empty_source_problems(build_path)
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__.split("\n", 1)[0])
+    parser.add_argument(
+        "build_paths",
+        nargs="*",
+        type=Path,
+        help=f"ESPHome build dirs (default: the first native ESP-IDF tree in {DEFAULT_GLOB})",
+    )
+    args = parser.parse_args()
+    paths = args.build_paths or sorted(REPO_ROOT.glob(DEFAULT_GLOB))
+    # Not resolved: SDKCONFIG must be spelled as the build spelled it.
+    trees = [
+        p
+        for p in paths
+        if (p / "build" / "project_description.json").is_file()
+        and (p / "build" / "CMakeCache.txt").is_file()
+    ]
+    if rejected := [p for p in args.build_paths if p not in trees]:
+        for path in rejected:
+            print(f"{path}: not a configured native ESP-IDF build tree")
+        return 1
+    if not trees:
+        print("No native ESP-IDF build tree found")
+        return 1
+    if not args.build_paths:
+        # The contract does not depend on the target, so one tree is enough.
+        trees = trees[:1]
+
+    failed = False
+    for tree in trees:
+        problems = check(tree)
+        print(f"{tree}: {'OK' if not problems else 'DIFFERS'}")
+        for problem in problems:
+            print(f"  {problem}")
+        failed |= bool(problems)
+    if failed:
+        print(
+            "The direct cmake/ninja build no longer matches idf.py. Compare "
+            "_IdfPyContract and its users in esphome/espidf/toolchain.py with the "
+            "pinned ESP-IDF tools/idf_py_actions."
+        )
+    return 1 if failed else 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

script/ci-custom.py

@@ -1,8 +1,10 @@
 #!/usr/bin/env python3
 
 import argparse
+import ast
 import codecs
 import collections
+from collections.abc import Iterator
 import fnmatch
 import functools
 import os.path
@@ -163,7 +165,21 @@ def lint_post_check(func):
     return func
 
 
-def lint_re_check(regex, **kwargs):
+def _nolint_in_match(content, haystack, match, mask):
+    """With masking, only a trailing comment counts: the raw span still holds string contents, and
+    the masked text is blank exactly where comments and strings were, so NOLINT must sit after the
+    last real code character of the span."""
+    raw = content[match.start() : match.end()]
+    if not mask:
+        return "NOLINT" in raw
+    masked = haystack[match.start() : match.end()].rstrip()
+    return "NOLINT" in raw[len(masked) :]
+
+
+def lint_re_check(regex, mask=False, prefilter=None, **kwargs):
+    """mask=True blanks comments and string literals first so prose about the pattern is not reported;
+    the masked text keeps its length, so match offsets still index the original content.
+    prefilter is a literal every match must contain, checked before the costlier masking."""
     flags = kwargs.pop("flags", re.MULTILINE)
     prog = re.compile(regex, flags)
     decor = lint_content_check(**kwargs)
@@ -172,8 +188,11 @@ def lint_re_check(regex, **kwargs):
         @functools.wraps(func)
         def new_func(fname, content):
             errs = []
-            for match in prog.finditer(content):
-                if "NOLINT" in match.group(0):
+            if prefilter is not None and prefilter not in content:
+                return errs
+            haystack = _mask_cpp_comments_strings(content) if mask else content
+            for match in prog.finditer(haystack):
+                if _nolint_in_match(content, haystack, match, mask):
                     continue
                 lineno = content.count("\n", 0, match.start()) + 1
                 substr = content[: match.start()]
@@ -247,6 +266,9 @@ def lint_ext_check(fname):
         "CLAUDE.md",
         "GEMINI.md",
         ".github/copilot-instructions.md",
+        # Symlinks to the shared .agents/skills directory
+        ".claude/skills",
+        ".github/skills",
         # Symlink to the real wifi scan_list.h so the test stub cannot drift
         "tests/integration/fixtures/external_components/wifi/scan_list.h",
     ]
@@ -525,21 +547,67 @@ def lint_conf_matches(fname, match):
 CONF_RE = r'^(CONF_[a-zA-Z0-9_]+)\s*=\s*[\'"].*?[\'"]\s*?$'
 with codecs.open("esphome/const.py", "r", encoding="utf-8") as const_f_handle:
     constants_content = const_f_handle.read()
+with codecs.open(
+    "esphome/components/const/__init__.py", "r", encoding="utf-8"
+) as component_const_f_handle:
+    component_constants_content = component_const_f_handle.read()
+
+# The two canonical homes for shared constants: esphome/const.py (core, frozen) and
+# esphome/components/const/__init__.py (shared by components). A constant defined in
+# either must be imported from there rather than redefined in a component.
+CONST_HOMES = ["esphome/const.py", "esphome/components/const/__init__.py"]
+
 CONSTANTS = [m.group(1) for m in re.finditer(CONF_RE, constants_content, re.MULTILINE)]
+COMPONENT_CONSTANTS = [
+    m.group(1) for m in re.finditer(CONF_RE, component_constants_content, re.MULTILINE)
+]
 
 CONSTANTS_USES = collections.defaultdict(list)
 
 
-@lint_re_check(CONF_RE, include=["*.py"], exclude=["esphome/const.py"])
+def _const_home_error(name, core_constants, component_constants):
+    """Return an error if the constant already lives in one of the canonical homes."""
+    if name in core_constants:
+        return (
+            f"Constant {highlight(name)} has already been defined in const.py - "
+            "please import the constant from const.py directly."
+        )
+    if name in component_constants:
+        return (
+            f"Constant {highlight(name)} has already been defined in "
+            "esphome/components/const/__init__.py - please import the constant from "
+            "esphome.components.const directly."
+        )
+    return None
+
+
+@lint_re_check(CONF_RE, include=["*.py"], exclude=CONST_HOMES)
 def lint_conf_from_const_py(fname, match):
     name = match.group(1)
-    if name not in CONSTANTS:
+    err = _const_home_error(name, CONSTANTS, COMPONENT_CONSTANTS)
+    if err is None:
         CONSTANTS_USES[name].append(fname)
-        return None
-    return (
-        f"Constant {highlight(name)} has already been defined in const.py - "
-        "please import the constant from const.py directly."
-    )
+    return err
+
+
+UNIT_RE = r'^(UNIT_[a-zA-Z0-9_]+)\s*=\s*[\'"].*?[\'"]\s*?$'
+UNIT_CONSTANTS = [
+    m.group(1) for m in re.finditer(UNIT_RE, constants_content, re.MULTILINE)
+]
+COMPONENT_UNIT_CONSTANTS = [
+    m.group(1) for m in re.finditer(UNIT_RE, component_constants_content, re.MULTILINE)
+]
+
+UNIT_CONSTANTS_USES = collections.defaultdict(list)
+
+
+@lint_re_check(UNIT_RE, include=["*.py"], exclude=CONST_HOMES)
+def lint_unit_from_const_py(fname, match):
+    name = match.group(1)
+    err = _const_home_error(name, UNIT_CONSTANTS, COMPONENT_UNIT_CONSTANTS)
+    if err is None:
+        UNIT_CONSTANTS_USES[name].append(fname)
+    return err
 
 
 RAW_PIN_ACCESS_RE = (
@@ -707,6 +775,20 @@ def lint_constants_usage():
     return errs
 
 
+@lint_post_check
+def lint_unit_constants_usage():
+    errs = []
+    for constant, uses in UNIT_CONSTANTS_USES.items():
+        if len(uses) < 3:
+            continue
+        errs.append(
+            f"Constant {highlight(constant)} is defined in {len(uses)} files. Please move all definitions of the "
+            f"constant to esphome/components/const/__init__.py (Uses: {', '.join(str(u) for u in uses)}) in a separate PR. "
+            "See https://developers.esphome.io/contributing/code/#python"
+        )
+    return errs
+
+
 # Maximum allowed CONF_ constants in esphome/const.py.
 # This file is frozen — new constants go in esphome/components/const/__init__.py.
 # Decrease this number when constants are moved out of const.py.
@@ -730,6 +812,40 @@ def lint_const_py_frozen(fname, content):
     return None
 
 
+@lint_content_check(include=["esphome/components/*/__init__.py"])
+def lint_component_domain(fname: Path, content: str) -> str | None:
+    """Require every component's __init__.py to define DOMAIN as its own name."""
+    if len(fname.parts) != 4:
+        # Platform packages such as esphome/components/<name>/sensor/__init__.py
+        return None
+    domain = fname.parts[2]
+    expected = f'DOMAIN = "{domain}"'
+    try:
+        tree = ast.parse(content, filename=str(fname))
+    except SyntaxError:
+        # Reported by the Python linters; nothing useful to add here.
+        return None
+    for node in tree.body:
+        if isinstance(node, ast.Assign):
+            targets = node.targets
+        elif isinstance(node, ast.AnnAssign):
+            targets = [node.target]
+        else:
+            continue
+        if (
+            any(isinstance(t, ast.Name) and t.id == "DOMAIN" for t in targets)
+            and isinstance(node.value, ast.Constant)
+            and node.value.value == domain
+        ):
+            return None
+    return (
+        f"Component is missing the {highlight(expected)} constant. "
+        "Add it in alphabetical order with the other component metadata such as "
+        "CODEOWNERS and DEPENDENCIES, so other code can refer to the component by "
+        "name, e.g. CORE.data[DOMAIN]."
+    )
+
+
 def relative_cpp_search_text(fname: Path, content) -> str:
     parts = fname.parts
     integration = parts[2]
@@ -823,6 +939,8 @@ def lint_relative_py_import(fname: Path, line, col, content):
         # neither can live in a C++ namespace.
         "esphome/components/esp32_hosted/esp_now_hosted.cpp",
         "esphome/components/esp32_hosted/esp_now_hosted_rpc.h",
+        # C header shared with the vendored decoder
+        "esphome/components/esphome/ota/ota_esphome_inflate.h",
     ],
 )
 def lint_namespace(fname: Path, content: str) -> str | None:
@@ -947,15 +1065,13 @@ HEAP_ALLOCATING_HELPERS = {
     r"str_sprintf|"
     r"str_snprintf|"
     r"value_accuracy_to_string"
-    r")\s*\(" + CPP_RE_EOL,
+    # Explicit template arguments at any nesting depth, e.g. format_hex<std::array<uint8_t, 4>>(
+    r")\s*(?:<[^;{}()\n]*>\s*)?\(" + CPP_RE_EOL,
     include=cpp_include,
     exclude=[
         # The definitions themselves
         "esphome/core/alloc_helpers.h",
         "esphome/core/alloc_helpers.cpp",
-        # Backward compatibility re-exports (remove before 2026.11.0)
-        "esphome/core/helpers.h",
-        "esphome/core/helpers.cpp",
         # Vendored third-party library
         "esphome/components/http_request/httplib.h",
     ],
@@ -1008,7 +1124,6 @@ def lint_no_sprintf(fname, match):
         # Vendored library
         "esphome/components/http_request/httplib.h",
         # Deprecated helpers that return std::string
-        "esphome/core/helpers.cpp",
         "esphome/core/alloc_helpers.cpp",
         # The using declaration itself
         "esphome/core/helpers.h",
@@ -1135,7 +1250,108 @@ def lint_no_std_bind(fname, match):
     )
 
 
-LOG_MULTILINE_RE = re.compile(r"ESP_LOG\w+\s*\(.*?;", re.DOTALL)
+@lint_re_check(
+    r"[^\w]std\s*::\s*nothrow\b" + CPP_RE_EOL,
+    mask=True,
+    prefilter="nothrow",
+    include=cpp_include,
+)
+def lint_no_std_nothrow(fname, match):
+    return (
+        f"{highlight('new (std::nothrow)')} aborts on ESP-IDF when the allocation fails, exceptions are disabled "
+        f"there, so it never returns nullptr.\n"
+        f"Please use {highlight('RAMAllocator')} from esphome/core/helpers.h, which does.\n"
+        f"  Before: {highlight('auto *buf = new (std::nothrow) uint8_t[n];')}\n"
+        f"  After:  {highlight('auto buf = RAMAllocator<uint8_t>().make_unique_array_for_overwrite(n);')}\n"
+        f"For one object use {highlight('RAMAllocator<T>().make_unique(args...)')}; both return empty on failure.\n"
+        f"Default flags prefer PSRAM; pass RAMAllocator<T>::PREFER_INTERNAL to keep it where new put it.\n"
+        f"(If strictly necessary, add `// NOLINT` to the end of the line)"
+    )
+
+
+@lint_re_check(
+    r"^\s*(?:(?:static|constexpr|inline)\s+)*(?:const\s+)?char\s*(?:\*\s*(?:const\s+)?TAG|\s(?:const\s+)?TAG\s*\[\w*\])"
+    r"\s*(?:=\s*\{?|[{(])\s*\"",
+    prefilter="TAG",
+    include=["esphome/components/*.cpp", "esphome/components/**/*.cpp"],
+)
+def lint_log_tag_macro(fname, match):
+    return (
+        f"Declare log tags with {highlight('ESPHOME_LOG_TAG(TAG, "name");')}, which keeps the tag in flash on "
+        f"ESP8266.\n"
+        f"  Before: {highlight('static const char *const TAG = "name";')}\n"
+        f"  After:  {highlight('ESPHOME_LOG_TAG(TAG, "name");')}\n"
+        f"Never pass such a TAG to set_timeout/set_interval names or string functions.\n"
+        f"(If strictly necessary, add `// NOLINT` to the end of the line)"
+    )
+
+
+@lint_re_check(
+    r"(?:\b(?:set_timeout|set_interval|set_retry|cancel_timeout|cancel_interval|cancel_retry|defer)"
+    r"|\b(?:strcmp|strncmp|strcasecmp|strlen|strcpy|strncpy)|std::string)\s*\((?:(?:[^();]|\([^()]*\))*,)?\s*TAG\b"
+    r"|\bstd::string\s+\w+\s*[={(]\s*TAG\b",
+    mask=True,
+    prefilter="TAG",
+    include=["esphome/components/*.cpp", "esphome/components/**/*.cpp"],
+)
+def lint_log_tag_as_string(fname, match):
+    return (
+        f"{highlight('TAG')} is in flash on ESP8266 (ESPHOME_LOG_TAG), so it can only be used for logging.\n"
+        f"Use a separate name or a numeric id for scheduler calls, and do not pass it to string functions.\n"
+        f"(If strictly necessary, add `// NOLINT` to the end of the line)"
+    )
+
+
+LOG_CALL_START_RE = re.compile(r"ESP_LOG\w+\s*\(")
+# Comments, raw/plain string literals and single char literals are consumed whole so ; ( ) ? :
+# inside them are never seen. A char literal is exactly one (escaped) char so a digit separator
+# like 1'000'000 cannot open one.
+CPP_COMMENT_RE = r"//[^\n]*|/\*.*?\*/"
+CPP_SKIP_RE = (
+    CPP_COMMENT_RE
+    + r'|R"(?P<raw_delim>[^(\s]*)\(.*?\)(?P=raw_delim)"|"(?:[^"\\]|\\.)*"|\'(?:[^\'\\\n]|\\.)\''
+)
+LOG_CALL_TOKEN_RE = re.compile(CPP_SKIP_RE + r"|[()]", re.DOTALL)
+# The last alternative matches a ? or : followed (after spaces or comments) by an opening quote,
+# i.e. a string literal used as a ternary branch.
+LOG_TERNARY_LITERAL_RE = re.compile(
+    CPP_SKIP_RE + r"|[?:](?:\s|" + CPP_COMMENT_RE + r')*(?=")', re.DOTALL
+)
+# A bare NOLINT; a clang-tidy NOLINT(check-name) is aimed at a different tool.
+NOLINT_RE = re.compile(r"\bNOLINT\b(?!\()")
+
+
+def _line_col(content: str, pos: int) -> tuple[int, int]:
+    """1-based line and column of an offset in content."""
+    return content.count("\n", 0, pos) + 1, pos - content.rfind("\n", 0, pos)
+
+
+def _iter_log_calls(content: str) -> Iterator[tuple[int, str | None]]:
+    """Yield (start, text) for every ESP_LOG*(...) call, text running to the matching close paren.
+    text is None when no matching paren exists so callers can report the call instead of skipping it."""
+    for head in LOG_CALL_START_RE.finditer(content):
+        depth = 1
+        for tok in LOG_CALL_TOKEN_RE.finditer(content, head.end()):
+            if tok.group(0) == "(":
+                depth += 1
+            elif tok.group(0) == ")":
+                depth -= 1
+                if depth == 0:
+                    yield head.start(), content[head.start() : tok.end()]
+                    break
+        else:
+            yield head.start(), None
+
+
+def _unbalanced_log_call_error(content: str, pos: int) -> tuple[int, int, str]:
+    lineno, col = _line_col(content, pos)
+    return (
+        lineno,
+        col,
+        "ESP_LOG call has no matching closing parenthesis, so it cannot be checked.",
+    )
+
+
 LOG_BAD_CONTINUATION_RE = re.compile(r'\\n(?:[^ \\"\r\n\t]|"\s*\n\s*"[^ \\])')
 LOG_PERCENT_S_CONTINUATION_RE = re.compile(r'\\n(?:%s|"\s*\n\s*"%s)')
 
@@ -1143,16 +1359,16 @@ LOG_PERCENT_S_CONTINUATION_RE = re.compile(r'\\n(?:%s|"\s*\n\s*"%s)')
 @lint_content_check(include=cpp_include)
 def lint_log_multiline_continuation(fname, content):
     errs = []
-    for log_match in LOG_MULTILINE_RE.finditer(content):
-        log_text = log_match.group(0)
+    for log_start, log_text in _iter_log_calls(content):
+        if log_text is None:
+            errs.append(_unbalanced_log_call_error(content, log_start))
+            continue
         for bad_match in LOG_BAD_CONTINUATION_RE.finditer(log_text):
             # %s may expand to a whitespace prefix at runtime, skip those
             if LOG_PERCENT_S_CONTINUATION_RE.match(log_text, bad_match.start()):
                 continue
             # Calculate line number from position in full content
-            abs_pos = log_match.start() + bad_match.start()
-            lineno = content.count("\n", 0, abs_pos) + 1
-            col = abs_pos - content.rfind("\n", 0, abs_pos)
+            lineno, col = _line_col(content, log_start + bad_match.start())
             errs.append(
                 (
                     lineno,
@@ -1170,6 +1386,90 @@ def lint_log_multiline_continuation(fname, content):
     return errs
 
 
+def _find_ternary_literals(text: str) -> Iterator[tuple[int, str]]:
+    """Yield (offset, literal) for every string literal used as a ternary branch."""
+    branch = False
+    for m in LOG_TERNARY_LITERAL_RE.finditer(text):
+        tok = m.group(0)
+        # An empty literal is merged with every other string's terminator, so it costs no RAM,
+        # while a PSTR("") would add its own flash array; leave it alone.
+        if branch and tok[0] == '"' and tok != '""':
+            yield m.start(), tok
+        branch = tok[0] in "?:"
+
+
+# LOG_STR_LITERAL is a no op everywhere except ESP8266, so code that never builds there is skipped
+# to avoid churn: platform specific sources and components for ESP32, LibreTiny, RP2 and Zephyr only.
+# A component belongs here only if it has no tests/components/<name>/test.esp8266-ard.yaml.
+LOG_LITERAL_LINT_EXCLUDE = [
+    "*_esp32.cpp",
+    "*_esp32_*.cpp",
+    "*_esp_idf.cpp",
+    "*_rmt.cpp",
+    "*_zephyr.cpp",
+    "*_bk72xx.cpp",
+    "*_libretiny.cpp",
+    "*_pico_w.cpp",
+    "*_host.cpp",
+    "esphome/components/esp32*/*",
+    "esphome/components/bk72xx*/*",
+    "esphome/components/ln882h*/*",
+    "esphome/components/ln882x*/*",
+    "esphome/components/rp2*/*",
+    "esphome/components/zephyr*/*",
+    "esphome/components/host/*",
+    "esphome/components/libretiny*/*",
+    "esphome/components/bluetooth_proxy/*",
+    "esphome/components/bluetooth_connection/*",
+    "esphome/components/ble_client/*",
+    "esphome/components/bedjet/*",
+    "esphome/components/anova/*",
+    "esphome/components/xiaomi_ble/*",
+    "esphome/components/bthome_mithermometer/*",
+    "esphome/components/usb_host/*",
+    "esphome/components/zigbee/*",
+    "esphome/components/lvgl/*",
+    # Test fixtures and host only unit tests - not production embedded code
+    "tests/integration/fixtures/*",
+    "tests/components/*",
+]
+
+
+@lint_content_check(include=cpp_include, exclude=LOG_LITERAL_LINT_EXCLUDE)
+def lint_log_no_bare_literal_ternary(
+    fname: Path, content: str
+) -> list[tuple[int, int, str]]:
+    errs = []
+    for log_start, log_text in _iter_log_calls(content):
+        if log_text is None:
+            continue  # reported by lint_log_multiline_continuation, which sees every file
+        # A NOLINT anywhere on the lines the call spans silences every branch in it
+        first_line = content.rfind("\n", 0, log_start) + 1
+        last_line = content.find("\n", log_start + len(log_text))
+        if NOLINT_RE.search(
+            content[first_line : last_line if last_line != -1 else None]
+        ):
+            continue
+        for offset, literal in _find_ternary_literals(log_text):
+            lineno, col = _line_col(content, log_start + offset)
+            errs.append(
+                (
+                    lineno,
+                    col,
+                    (
+                        "String literal used as a ternary branch in a log call. On ESP8266 the "
+                        "log macro moves the format string to flash, but bare literal arguments "
+                        "stay in RAM. Wrap each branch passed straight to the log call in "
+                        f"{highlight('LOG_STR_LITERAL(...)')}:\n"
+                        f"  Before: {highlight(literal)}\n"
+                        f"  After:  {highlight(f'LOG_STR_LITERAL({literal})')}\n"
+                        f"(If strictly necessary, add `{highlight('// NOLINT')}` to the end of the line)"
+                    ),
+                )
+            )
+    return errs
+
+
 @lint_content_find_check(
     "ESP_LOG",
     include=["*.h", "*.tcc"],

script/clang-tidy

@@ -1,6 +1,7 @@
 #!/usr/bin/env python3
 
 import argparse
+import heapq
 import os
 from pathlib import Path
 import queue
@@ -11,6 +12,7 @@ import subprocess
 import sys
 import tempfile
 import threading
+import time
 from typing import Any
 
 import click
@@ -277,10 +279,21 @@ def clang_options(idedata, environment):
 
 pids = set()
 
+# Generated files the static analyzer is skipped on. The analyzer inlines the
+# small dump helpers into every generated dump function and takes
+# over 200 s on this file in CI, 99% of its clang-tidy time; any finding there
+# would be fixed in script/api_protobuf anyway. The other checks still run.
+ANALYZER_SKIPPED_FILES: frozenset[str] = frozenset(
+    {"esphome/components/api/api_pb2_dump.cpp"}
+)
 
-def run_tidy(executable, args, options, tmpdir, path_queue, lock, failed_files):
+
+def run_tidy(
+    executable, args, options, tmpdir, path_queue, lock, failed_files, timings
+):
     while True:
         path = path_queue.get()
+        started = time.monotonic()
         invocation = [executable]
 
         if tmpdir is not None:
@@ -297,16 +310,22 @@ def run_tidy(executable, args, options, tmpdir, path_queue, lock, failed_files):
         if sys.stdout.isatty():
             invocation.append("--use-color")
 
+        disabled_checks = []
         if args.environment.startswith(("rp2", "bk72xx", "ln882h", "rtl87xx")):
             # MMIO peripheral access on these bare-metal platforms is all
             # fixed-address.
             # bugprone-pointer-arithmetic-on-polymorphic-object (and its
             # cert-ctr56-cpp alias) crashes clang-tidy 22 with infinite matcher
             # recursion on lvgl_esphome.h under the RP2 defines.
-            invocation.append(
-                "--checks=-clang-analyzer-core.FixedAddressDereference,"
-                "-bugprone-pointer-arithmetic-on-polymorphic-object,-cert-ctr56-cpp"
-            )
+            disabled_checks += [
+                "-clang-analyzer-core.FixedAddressDereference",
+                "-bugprone-pointer-arithmetic-on-polymorphic-object",
+                "-cert-ctr56-cpp",
+            ]
+        if os.path.normpath(path) in ANALYZER_SKIPPED_FILES:
+            disabled_checks.append("-clang-analyzer-*")
+        if disabled_checks:
+            invocation.append("--checks=" + ",".join(disabled_checks))
         invocation.append(f"--header-filter={Path(basepath).resolve()}/.*")
         invocation.append(str(Path(path).resolve()))
         invocation.append("--")
@@ -319,8 +338,9 @@ def run_tidy(executable, args, options, tmpdir, path_queue, lock, failed_files):
             check=False,
             close_fds=False,
         )
-        if proc.returncode != 0:
-            with lock:
+        with lock:
+            timings.append((time.monotonic() - started, path))
+            if proc.returncode != 0:
                 print_error_for_file(path, proc.stdout)
                 failed_files.append(path)
         path_queue.task_done()
@@ -333,8 +353,27 @@ def progress_bar_show(value):
 
 
 def split_list(a, n):
-    k, m = divmod(len(a), n)
-    return [a[i * k + min(i, m) : (i + 1) * k + min(i + 1, m)] for i in range(n)]
+    """Deal the sorted files out to n splits in turn.
+
+    Contiguous slices put whole directories on one split; expensive
+    neighbours such as esphome/core then land together and that split takes
+    several times longer than the others. Dealing spreads them out.
+    """
+    return [a[i::n] for i in range(n)]
+
+
+SLOWEST_FILES_SHOWN = 15
+
+
+def print_slowest_files(timings):
+    """Print the files clang-tidy spent the longest on, so the splits can be
+    rebalanced from CI logs."""
+    if not timings:
+        return
+    total = sum(seconds for seconds, _ in timings)
+    print(f"\nclang-tidy spent {total:.0f} s on {len(timings)} files; slowest:")
+    for seconds, path in heapq.nlargest(SLOWEST_FILES_SHOWN, timings):
+        print(f"  {seconds:6.1f} s  {path}")
 
 
 def main():
@@ -458,6 +497,7 @@ def main():
         tmpdir = tempfile.mkdtemp()
 
     failed_files = []
+    timings = []
     try:
         executable = get_binary("clang-tidy", 22)
         task_queue = queue.Queue(args.jobs)
@@ -473,6 +513,7 @@ def main():
                     task_queue,
                     lock,
                     failed_files,
+                    timings,
                 ),
             )
             t.daemon = True
@@ -487,6 +528,7 @@ def main():
 
         # Wait for all threads to be done.
         task_queue.join()
+        print_slowest_files(timings)
 
     except FileNotFoundError:
         return 1

script/cpp_benchmark.py

@@ -25,19 +25,14 @@ CORE_BENCHMARKS_DIR: Path = Path(root_path) / "tests" / "benchmarks" / "core"
 # allow benchmarks to compile on the host platform.
 STUBS_DIR: Path = Path(root_path) / "tests" / "benchmarks" / "stubs"
 
-PLATFORMIO_OPTIONS = {
-    "build_flags": [
-        "-Os",  # match firmware optimization level (detects inlining regressions)
-        "-g",  # debug symbols for profiling
-        "-ffunction-sections",  # required for dead-code stripping with -Os
-        "-fdata-sections",  # required for dead-code stripping with -Os
-        "-DUSE_BENCHMARK",  # disable WarnIfComponentBlockingGuard in finish()
-        f"-I{STUBS_DIR}",  # stub headers for ESP32-only components
-    ],
-    # Use deep+ LDF mode to ensure PlatformIO detects the benchmark
-    # library dependency from nested includes.
-    "lib_ldf_mode": "deep+",
-}
+BUILD_FLAGS = [
+    "-Os",  # match firmware optimization level (detects inlining regressions)
+    "-g",  # debug symbols for profiling
+    "-ffunction-sections",  # required for dead-code stripping with -Os
+    "-fdata-sections",  # required for dead-code stripping with -Os
+    "-DUSE_BENCHMARK",  # disable WarnIfComponentBlockingGuard in finish()
+    f"-I{STUBS_DIR}",  # stub headers for ESP32-only components
+]
 
 
 def run_benchmarks(selected_components: list[str], build_only: bool = False) -> int:
@@ -46,7 +41,7 @@ def run_benchmarks(selected_components: list[str], build_only: bool = False) ->
     # containing {"lib_path": "/path/to/google_benchmark"}.
     lib_config_json = os.environ.get("BENCHMARK_LIB_CONFIG")
 
-    pio_options = PLATFORMIO_OPTIONS
+    build_flags = BUILD_FLAGS
     if lib_config_json:
         lib_config = json.loads(lib_config_json)
         benchmark_lib = f"benchmark=symlink://{lib_config['lib_path']}"
@@ -61,10 +56,7 @@ def run_benchmarks(selected_components: list[str], build_only: bool = False) ->
             "-DCODSPEED_ANALYSIS",
             f'-DCODSPEED_ROOT_DIR=\\"{project_root}\\"',
         ]
-        pio_options = {
-            **PLATFORMIO_OPTIONS,
-            "build_flags": PLATFORMIO_OPTIONS["build_flags"] + codspeed_flags,
-        }
+        build_flags = BUILD_FLAGS + codspeed_flags
     else:
         benchmark_lib = PLATFORMIO_GOOGLE_BENCHMARK_LIB
 
@@ -77,7 +69,7 @@ def run_benchmarks(selected_components: list[str], build_only: bool = False) ->
         config_prefix="cppbench",
         friendly_name="CPP Benchmarks",
         libraries=benchmark_lib,
-        platformio_options=pio_options,
+        build_flags=build_flags,
         main_entry="main.cpp",
         label="benchmarks",
         build_only=build_only,

script/cpp_unit_test.py

@@ -15,24 +15,16 @@ from helpers import get_all_components, root_path
 # Path to /tests/components
 COMPONENTS_TESTS_DIR: Path = Path(root_path) / "tests" / "components"
 
-PLATFORMIO_OPTIONS = {
-    "build_type": "debug",
-    "build_unflags": [
-        "-Os",  # remove size-opt flag
-    ],
-    "build_flags": [
-        "-Og",  # optimize for debug
-        "-DESPHOME_DEBUG",  # enable debug assertions
-        # Enable the address and undefined behavior sanitizers
-        "-fsanitize=address",
-        "-fsanitize=undefined",
-        "-fno-omit-frame-pointer",
-    ],
-    "debug_build_flags": [  # only for debug builds
-        "-g3",  # max debug info
-        "-ggdb3",
-    ],
-}
+BUILD_FLAGS = [
+    "-Og",  # optimize for debug
+    "-g3",  # max debug info
+    "-ggdb3",
+    "-DESPHOME_DEBUG",  # enable debug assertions
+    # Enable the address and undefined behavior sanitizers
+    "-fsanitize=address",
+    "-fsanitize=undefined",
+    "-fno-omit-frame-pointer",
+]
 
 
 def run_tests(selected_components: list[str]) -> int:
@@ -47,7 +39,7 @@ def run_tests(selected_components: list[str]) -> int:
         config_prefix="cpptests",
         friendly_name="CPP Unit Tests",
         libraries=PLATFORMIO_GOOGLE_TEST_LIB,
-        platformio_options=PLATFORMIO_OPTIONS,
+        build_flags=BUILD_FLAGS,
         main_entry="main.cpp",
         label="unit tests",
     )

script/determine-jobs.py

@@ -31,7 +31,7 @@ The CI workflow uses this information to:
 - Skip or run integration tests
 - Skip or run clang-tidy (and whether to do a full scan)
 - Skip or run clang-format
-- Skip or run Python linters (ruff, flake8, pylint, pyupgrade)
+- Skip or run pylint
 - Skip or run downstream esphome/device-builder tests against the PR's Python code
 - Determine which components to test individually
 - Decide how to split component tests (if there are many)
@@ -50,6 +50,7 @@ from __future__ import annotations
 
 import argparse
 from collections import Counter
+from collections.abc import Callable
 from enum import StrEnum
 from functools import cache
 import json
@@ -98,10 +99,6 @@ from split_components_for_ci import create_intelligent_batches
 # For large PRs (>= 65 files), use split for better parallelization
 CLANG_TIDY_SPLIT_THRESHOLD = 65
 
-# Component test batch size (weighted)
-# Isolated components count as 10x, groupable components count as 1x
-COMPONENT_TEST_BATCH_SIZE = 40
-
 # Above the threshold, fan out across up to this many jobs, balanced by the
 # recorded per-file durations. The target is serial junit-time weight per
 # bucket, not wall time (calibrated with the conftest compile cap); it
@@ -110,14 +107,25 @@ INTEGRATION_TESTS_SPLIT_THRESHOLD = 10
 INTEGRATION_TESTS_SPLIT_BUCKETS = 5
 INTEGRATION_TESTS_TARGET_BUCKET_WEIGHT = 360.0
 
-# platformio and aioesphomeapi (requirements.txt), the pytest stack
-# (requirements_test.txt) and the fixture every session compiles; a change
-# to any runs the full matrix
+# aioesphomeapi (requirements.txt), the pytest stack (requirements_test.txt)
+# and the native host build backend every test compiles with; a change to
+# any runs the full matrix
 INTEGRATION_TESTS_TRIGGER_FILES = frozenset(
     {
         "requirements.txt",
         "requirements_test.txt",
-        "tests/integration/fixtures/cache_init.yaml",
+        "esphome/arduino/library.py",
+        "esphome/build_gen/build_tool.py",
+        "esphome/build_gen/host.py",
+        "esphome/build_helpers/ccache.py",
+        "esphome/build_helpers/idedata.py",
+        "esphome/build_helpers/native.py",
+        "esphome/build_helpers/ninja.py",
+        "esphome/build_helpers/ninja_gen.py",
+        "esphome/build_helpers/tools_cache.py",
+        "esphome/framework_helpers.py",
+        "esphome/host/toolchain.py",
+        "esphome/platformio/library.py",
     }
 )
 
@@ -239,7 +247,7 @@ def determine_integration_tests(branch: str | None = None) -> tuple[bool, list[s
        - conftest.py, types.py, const.py, entity_utils.py, state_utils.py, etc.
 
     4. A file in INTEGRATION_TESTS_TRIGGER_FILES changed
-       - The dependency pins and the session init fixture affect every test
+       - The dependency pins and the host build backend affect every test
 
     Returns (run_all=False, [test_files...]) when:
 
@@ -395,21 +403,33 @@ def should_run_clang_format(branch: str | None = None) -> bool:
     return _any_changed_file_endswith(branch, CPP_FILE_EXTENSIONS)
 
 
+# Linter pins, dependencies and linter config can change findings without a .py change.
+PYTHON_LINTERS_TRIGGER_FILES = frozenset(
+    {
+        "requirements.txt",
+        "requirements_test.txt",
+        "pyproject.toml",
+    }
+)
+
+
 def should_run_python_linters(branch: str | None = None) -> bool:
-    """Determine if Python linters (ruff, flake8, pylint, pyupgrade) should run based on changed files.
+    """Determine if the pylint job should run based on changed files.
 
-    This function is used by the CI workflow to skip Python linting checks when no Python files
-    have changed, saving CI time and resources.
-
-    Python linters will run when any Python source files have changed.
+    The CI workflow skips pylint unless a Python file or a file in
+    PYTHON_LINTERS_TRIGGER_FILES has changed. ruff, flake8 and pyupgrade run in
+    lint-format, which this flag does not gate.
 
     Args:
         branch: Branch to compare against. If None, uses default.
 
     Returns:
-        True if Python linters should run, False otherwise.
+        True if pylint should run, False otherwise.
     """
-    return _any_changed_file_endswith(branch, PYTHON_FILE_EXTENSIONS)
+    return any(
+        file.endswith(PYTHON_FILE_EXTENSIONS) or file in PYTHON_LINTERS_TRIGGER_FILES
+        for file in changed_files(branch)
+    )
 
 
 # Files outside esphome/**/*.py whose changes can affect `import esphome.__main__`
@@ -520,48 +540,78 @@ ESP32_PLATFORMIO_TEST_COMPONENTS = frozenset(
     }
 )
 
+# Shared by every toolchain smoke-test job: the base config and the bus
+# packages each generated build includes
+_SMOKE_HARNESS_TRIGGER_PATH_PREFIXES = ("tests/test_build_components/",)
+
 # Path prefixes whose changes always trigger the PlatformIO compile test:
 # anything under esphome/platformio/ (the PlatformIO runner / toolchain that
 # drives every PlatformIO build). The esp32 platform component is already in
 # ESP32_PLATFORMIO_TEST_COMPONENTS, so its changes are covered by the normal
 # component-narrowing path.
-ESP32_PLATFORMIO_TRIGGER_PATH_PREFIXES = ("esphome/platformio/",)
+ESP32_PLATFORMIO_TRIGGER_PATH_PREFIXES = (
+    "esphome/platformio/",
+    *_SMOKE_HARNESS_TRIGGER_PATH_PREFIXES,
+)
 
-# Standalone files that, when changed, trigger the PlatformIO compile test:
-#   - esphome/build_gen/platformio.py -- the PlatformIO build generator
-#   - script/test_build_components.py -- the harness the job invokes
-#   - .github/workflows/ci.yml -- the job's own definition
-ESP32_PLATFORMIO_TRIGGER_FILES = frozenset(
+# Shared by every toolchain smoke-test job: the harness it invokes and the
+# workflow that defines it
+_SMOKE_HARNESS_TRIGGER_FILES = frozenset(
     {
-        "esphome/build_gen/platformio.py",
         "script/test_build_components.py",
         ".github/workflows/ci.yml",
     }
 )
 
+# Standalone files that, when changed, trigger the PlatformIO compile test
+ESP32_PLATFORMIO_TRIGGER_FILES = _SMOKE_HARNESS_TRIGGER_FILES | {
+    # The PlatformIO build generator
+    "esphome/build_gen/platformio.py",
+    # Decides which platforms take the pch script in esphome/platformio/,
+    # and the module that parses its switch and the ccache settings
+    "esphome/build_helpers/pch.py",
+    "esphome/build_helpers/ccache.py",
+}
+
+
+def _path_or_file_trigger(
+    files: list[str],
+    trigger_files: frozenset[str],
+    trigger_prefixes: tuple[str, ...],
+) -> bool:
+    """Whether any changed file matches the given infrastructure triggers."""
+    return any(
+        file in trigger_files or file.startswith(trigger_prefixes) for file in files
+    )
+
+
+@cache
+def _cached_components_closure(files: tuple[str, ...]) -> frozenset[str]:
+    """Dependency closure of the changed components; cached because the
+    walk is expensive and every smoke-test job asks for the same list."""
+    component_files = [f for f in files if filter_component_and_test_files(f)]
+    return frozenset(get_components_with_dependencies(component_files, True))
+
 
 def _esp32_platformio_path_or_file_trigger(files: list[str]) -> bool:
     """Whether any changed file is a PlatformIO infrastructure / harness trigger."""
-    for file in files:
-        if file in ESP32_PLATFORMIO_TRIGGER_FILES:
-            return True
-        if any(
-            file.startswith(prefix) for prefix in ESP32_PLATFORMIO_TRIGGER_PATH_PREFIXES
-        ):
-            return True
-    return False
+    return _path_or_file_trigger(
+        files, ESP32_PLATFORMIO_TRIGGER_FILES, ESP32_PLATFORMIO_TRIGGER_PATH_PREFIXES
+    )
+
+
+# Checks the native ESP-IDF build in CI but does not shape it, so it is kept
+# out of ESP_IDF_INFRA_TRIGGER_FILES (hashed into the clang-tidy cache key).
+_ESP_IDF_CHECK_TRIGGER_FILES = frozenset({"script/check_idf_py_equivalence.py"})
 
 
 def _esp_idf_infra_changed(files: list[str]) -> bool:
     """Whether any changed file is ESP-IDF build/runner infrastructure."""
-    for file in files:
-        if file in ESP_IDF_INFRA_TRIGGER_FILES:
-            return True
-        if any(
-            file.startswith(prefix) for prefix in ESP_IDF_INFRA_TRIGGER_PATH_PREFIXES
-        ):
-            return True
-    return False
+    return _path_or_file_trigger(
+        files,
+        ESP_IDF_INFRA_TRIGGER_FILES | _ESP_IDF_CHECK_TRIGGER_FILES,
+        ESP_IDF_INFRA_TRIGGER_PATH_PREFIXES,
+    )
 
 
 def esp32_platformio_components_to_test(branch: str | None = None) -> list[str]:
@@ -599,15 +649,23 @@ def esp32_platformio_components_to_test(branch: str | None = None) -> list[str]:
     Returns:
         Sorted list of component names to compile.
     """
+    return _toolchain_components_to_test(
+        branch, ESP32_PLATFORMIO_TEST_COMPONENTS, _esp32_platformio_path_or_file_trigger
+    )
+
+
+def _toolchain_components_to_test(
+    branch: str | None,
+    test_set: frozenset[str],
+    infra_trigger: Callable[[list[str]], bool],
+) -> list[str]:
+    """The shared narrowing rule for the per-toolchain smoke-test jobs."""
     files = changed_files(branch)
 
-    if core_changed(files) or _esp32_platformio_path_or_file_trigger(files):
-        return sorted(ESP32_PLATFORMIO_TEST_COMPONENTS)
+    if core_changed(files) or infra_trigger(files):
+        return sorted(test_set)
 
-    component_files = [f for f in files if filter_component_and_test_files(f)]
-    changed = get_components_with_dependencies(component_files, True)
-
-    return sorted(ESP32_PLATFORMIO_TEST_COMPONENTS & set(changed))
+    return sorted(test_set & _cached_components_closure(tuple(files)))
 
 
 def should_run_esp32_platformio(branch: str | None = None) -> bool:
@@ -628,6 +686,85 @@ def should_run_esp32_platformio(branch: str | None = None) -> bool:
     return bool(esp32_platformio_components_to_test(branch))
 
 
+# The `--toolchain arduino` smoke-test set: covers the core, the bundled and
+# converted registry libraries, and the waveform path.
+ESP8266_NATIVE_TEST_COMPONENTS = frozenset(
+    {
+        "esp8266",
+        "api",
+        "web_server",
+        "captive_portal",
+        "mqtt",
+        "esp8266_pwm",
+        "neopixelbus",
+        "bme280_i2c",
+        "uart",
+    }
+)
+
+# Infrastructure whose changes always trigger the native ESP8266
+# compile test
+ESP8266_NATIVE_TRIGGER_PATH_PREFIXES = (
+    "esphome/arduino8266/",
+    "esphome/arduino/",
+    "esphome/build_helpers/",
+    *_SMOKE_HARNESS_TRIGGER_PATH_PREFIXES,
+)
+# Shared library-conversion modules every native build imports; espidf-only
+# infra (build_gen/espidf.py) deliberately stays out of the esp8266 set.
+_NATIVE_SHARED_TRIGGER_FILES = frozenset(
+    {
+        "esphome/framework_helpers.py",
+        "esphome/platformio/library.py",
+        "esphome/platformio/extra_script.py",
+    }
+)
+# Tripwire: the shared modules must stay in the ESP-IDF trigger set too
+# (now defined in clang_tidy_hash), or its smoke test silently skips them
+assert _NATIVE_SHARED_TRIGGER_FILES <= ESP_IDF_INFRA_TRIGGER_FILES
+ESP8266_NATIVE_TRIGGER_FILES = (
+    _NATIVE_SHARED_TRIGGER_FILES
+    | _SMOKE_HARNESS_TRIGGER_FILES
+    | {
+        "esphome/build_gen/arduino8266.py",
+        "esphome/build_gen/build_tool.py",
+        "esphome/components/esp8266/build_surgery.py",
+        "esphome/components/esp8266/boards.py",
+        "esphome/platformio/registry.py",
+        # esp8266/__init__.py imports copy_ccache_script from it
+        "esphome/platformio/toolchain.py",
+        ".github/actions/cache-arduino8266/action.yml",
+    }
+)
+
+
+def _esp8266_native_path_or_file_trigger(files: list[str]) -> bool:
+    """Whether any changed file is native-ESP8266 infrastructure / harness."""
+    # base_python_changed covers the top-level esphome/*.py modules the
+    # native backend imports directly (framework_helpers, helpers, writer,
+    # __main__); without it a change there would silently skip this job.
+    # base_python_changed is deliberately broad (any top-level esphome/*.py)
+    # as belt-and-braces while the backend is new; narrow it to the modules
+    # the backend imports once the toolchain has soaked a few releases
+    return base_python_changed(files) or _path_or_file_trigger(
+        files, ESP8266_NATIVE_TRIGGER_FILES, ESP8266_NATIVE_TRIGGER_PATH_PREFIXES
+    )
+
+
+def esp8266_native_components_to_test(branch: str | None = None) -> list[str]:
+    """The smoke set on a native-build change, nothing otherwise.
+
+    Unlike the esp32 PlatformIO job, this one does not narrow to the changed
+    components: the component matrix already compiles every esp8266 fixture
+    with this toolchain, so the only gap left is a change to the native build
+    itself that brings no component along.
+    """
+    files = changed_files(branch)
+    if core_changed(files) or _esp8266_native_path_or_file_trigger(files):
+        return sorted(ESP8266_NATIVE_TEST_COMPONENTS)
+    return []
+
+
 def determine_cpp_unit_tests(
     branch: str | None = None,
 ) -> tuple[bool, list[str]]:
@@ -1183,6 +1320,7 @@ def detect_memory_impact_config(
         "components": compatible_components,
         "platform": platform,
         "use_merged_config": "true",
+        "needs_arduino8266": platform.startswith("esp8266"),
     }
 
 
@@ -1226,6 +1364,8 @@ def main() -> None:
         run_device_builder = True
         esp32_platformio_components = sorted(ESP32_PLATFORMIO_TEST_COMPONENTS)
         run_esp32_platformio = True
+        esp8266_native_components = sorted(ESP8266_NATIVE_TEST_COMPONENTS)
+        run_esp8266_native = True
     else:
         integration_run_all, integration_test_files = determine_integration_tests(
             args.branch
@@ -1237,6 +1377,8 @@ def main() -> None:
         run_device_builder = should_run_device_builder(args.branch)
         esp32_platformio_components = esp32_platformio_components_to_test(args.branch)
         run_esp32_platformio = bool(esp32_platformio_components)
+        esp8266_native_components = esp8266_native_components_to_test(args.branch)
+        run_esp8266_native = bool(esp8266_native_components)
     run_integration, integration_test_buckets = _compute_integration_test_buckets(
         integration_run_all, integration_test_files
     )
@@ -1400,21 +1542,43 @@ def main() -> None:
         batches, _ = create_intelligent_batches(
             components=changed_components_with_tests,
             tests_dir=tests_dir,
-            batch_size=COMPONENT_TEST_BATCH_SIZE,
             directly_changed=batch_directly_changed,
         )
         # Convert batches to CI matrix entries: the component list plus which
         # native toolchain installs the batch's test platforms need, so the
         # workflow only restores the matching multi-GB toolchain caches.
+        # The idf.py check does not depend on the components, so it runs once
+        # per workflow, in the first batch that compiles an esp32 test (a
+        # validate-only component is never compiled).
+        idf_py_check_assigned = False
+        skip_compile = set(validate_only_components)
         for batch in batches:
             platforms: set[str] = set()
+            compiled_platforms: set[str] = set()
             for component in batch:
-                platforms.update(get_component_test_platforms(component))
+                # Variants included: the compile stage builds them, so a
+                # component tested only by test-<variant>.<platform>.yaml
+                # still needs that platform's toolchain
+                component_platforms = get_component_test_platforms(
+                    component, base_only=False
+                )
+                platforms.update(component_platforms)
+                if component not in skip_compile:
+                    compiled_platforms.update(component_platforms)
+            needs_idf = any(p.startswith("esp32") for p in platforms)
+            check_idf_py = not idf_py_check_assigned and any(
+                p.startswith("esp32") for p in compiled_platforms
+            )
+            idf_py_check_assigned |= check_idf_py
             component_test_batches.append(
                 {
                     "components": " ".join(batch),
-                    "needs_idf": any(p.startswith("esp32") for p in platforms),
+                    "needs_idf": needs_idf,
+                    "check_idf_py": check_idf_py,
                     "needs_nrf": any(p.startswith("nrf52") for p in platforms),
+                    "needs_arduino8266": any(
+                        p.startswith("esp8266") for p in platforms
+                    ),
                 }
             )
 
@@ -1432,6 +1596,8 @@ def main() -> None:
         "device_builder": run_device_builder,
         "esp32_platformio": run_esp32_platformio,
         "esp32_platformio_components": ",".join(esp32_platformio_components),
+        "esp8266_native": run_esp8266_native,
+        "esp8266_native_components": ",".join(esp8266_native_components),
         "changed_components": changed_components,
         "changed_components_with_tests": changed_components_with_tests,
         "directly_changed_components_with_tests": list(directly_changed_with_tests),

script/helpers.py

@@ -429,11 +429,9 @@ def build_all_include(header_files: list[str] | None = None) -> None:
             if line
         ]
 
-    from esphome.writer import ENTITY_TYPES_H_TARGET
+    from esphome.writer import ESPHOME_H_EXCLUDE
 
-    # X-macro files are included multiple times with different macro definitions
-    # and must not be included bare in the all-include header
-    exclude = {ENTITY_TYPES_H_TARGET}
+    exclude = {path.as_posix() for path in ESPHOME_H_EXCLUDE}
     headers = [f'#include "{h}"' for h in header_files if h not in exclude]
     headers.sort()
     headers.append("")
@@ -1618,7 +1616,7 @@ def lpt_partition(
 
     Heaviest item first into the lightest group. Ties keep input order, so
     pass pre-sorted items for deterministic output. script/clang-tidy's
-    split_list is the unweighted contiguous sibling.
+    split_list is the unweighted sibling that deals files out in turn.
     """
     groups: list[list[str]] = [[] for _ in range(count)]
     group_weights = [0.0] * count

script/platformio_install_deps.py

@@ -1,377 +0,0 @@
-#!/usr/bin/env python3
-# This script is used to preinstall
-# all platformio libraries in the global storage
-
-import argparse
-from concurrent.futures import ThreadPoolExecutor
-import configparser
-from contextlib import suppress
-import os
-from pathlib import Path
-import queue
-import subprocess
-import threading
-import traceback
-
-# esphome is not installed at this docker layer; pio's fs.rmtree is the
-# same chmod-on-readonly shape its own installer uses
-try:
-    from platformio import fs
-    from platformio.cache import ContentCache
-    from platformio.package.manager.base import BasePackageManager
-    from platformio.package.manager.library import LibraryPackageManager
-    from platformio.package.manager.tool import ToolPackageManager
-    from platformio.package.meta import PackageCompatibility
-
-    PARALLEL_AVAILABLE = True
-except ImportError as err:  # pragma: no cover
-    # A moved pio module must degrade to the serial pass, not kill the
-    # image build; the tripwire test makes the drift loud in CI
-    PARALLEL_AVAILABLE = False
-    IMPORT_ERROR = repr(err)
-
-# Network-bound downloads release the GIL, so the pool oversubscribes
-# the cores. This bypasses pio's 500ms registry throttle and races its
-# self-unlinking cache LockFiles; both are cache-only and self-healing.
-MAX_WORKERS = 16
-
-
-class CleanupError(RuntimeError):
-    """A torn destination could not be removed; the serial pass would
-    trust it, so the build must fail rather than bake a corrupt image."""
-
-
-class LockReleaseError(RuntimeError):
-    """The manager lock could not be released; the serial pass would
-    block on it, so the build must fail with the cause named."""
-
-
-def parse_specs(path: str, args: argparse.Namespace) -> tuple[list, list, list]:
-    """Extract lib/platform/tool specs from every section of a platformio.ini."""
-    config = configparser.ConfigParser(inline_comment_prefixes=(";",))
-    if not config.read(path):
-        # ConfigParser silently ignores unreadable files; an empty spec
-        # list would build an image with no dependencies at all
-        raise SystemExit(f"Could not read {path}")
-    libs = []
-    tools = []
-    platforms = []
-    for section in config.sections():
-        conf = config[section]
-        if "lib_deps" in conf and args.libraries:
-            for lib_dep in conf["lib_deps"].splitlines():
-                if not lib_dep:
-                    # Empty line or comment
-                    continue
-                if lib_dep.startswith("${"):
-                    # Extending from another section
-                    continue
-                if "@" not in lib_dep:
-                    # No version pinned, this is an internal lib
-                    continue
-                libs.append(lib_dep)
-        if "platform" in conf and args.platforms:
-            platforms.append(conf["platform"])
-        if "platform_packages" in conf and args.tools:
-            for tool in conf["platform_packages"].splitlines():
-                if not tool:
-                    # Empty line or comment
-                    continue
-                if tool.startswith("${"):
-                    # Extending from another section
-                    continue
-                if tool.find("https://github.com") != -1:
-                    split = tool.find("@")
-                    tool = tool[split + 1 :]
-                tools.append(tool)
-    # Exact-string dedupe only: name-level dedupe would change which
-    # version conflicts the pkg install pass reconciles
-    return (
-        list(dict.fromkeys(libs)),
-        list(dict.fromkeys(platforms)),
-        list(dict.fromkeys(tools)),
-    )
-
-
-def piopm_matches(package_dir: str, spec) -> list[Path]:
-    """Dirs whose .piopm metadata names this spec; a positive match beats
-    guessing the manifest-derived dirname from the registry name."""
-    want = (BasePackageManager.ensure_spec(spec).name or "").lower()
-    matches: list[Path] = []
-    if not want:
-        return matches
-    try:
-        entries = list(Path(package_dir).iterdir())
-    except FileNotFoundError:
-        return matches
-    for d in entries:
-        if not d.is_dir():
-            continue  # pio's get_installed skips files and *.pio-link too
-        try:
-            meta = fs.load_json(str(d / ".piopm"))
-        except FileNotFoundError:
-            continue  # no metadata means pio does not trust it either
-        except (OSError, ValueError):
-            if d.name.lower() == want:
-                # A corrupt .piopm under this spec's own name would crash
-                # pio's whole storage scan; remove it
-                matches.append(d)
-            continue
-        mspec = meta.get("spec") or {}
-        if (mspec.get("name") or meta.get("name") or "").lower() == want:
-            matches.append(d)
-    return matches
-
-
-def remove_dir(spec, dest: Path) -> None:
-    # fs.rmtree never raises (errors go to a printing onexc handler);
-    # only the destination's absence proves the cleanup worked
-    fs.rmtree(str(dest))
-    if dest.exists():
-        # Failing the build beats baking a corrupt image
-        raise CleanupError(
-            f"could not remove the failed pre-install of {spec} at {dest}"
-        )
-    print(f"Removed torn destination {dest}", flush=True)
-
-
-def cleanup_or_die(mgr, spec) -> None:
-    """Cleanup that did not demonstrably succeed must fail the build."""
-    try:
-        clean_torn(mgr, spec)
-    except CleanupError:
-        raise
-    except Exception as err:  # noqa: BLE001
-        raise CleanupError(f"cleanup failed for {spec}: {err!r}") from err
-
-
-def clean_torn(mgr, spec) -> None:
-    """Remove a torn destination so the serial pass cannot trust it."""
-    pkg = None
-    with suppress(Exception):
-        # get_package memoizes a pre-install snapshot; reset to see the
-        # torn dir. It also recognizes manifest-only legacy dirs pio's
-        # storage scan would trust, which the .piopm fallback cannot see.
-        mgr.memcache_reset()
-        pkg = mgr.get_package(spec)
-    if pkg is not None:
-        remove_dir(spec, Path(pkg.path))
-    elif dests := piopm_matches(mgr.package_dir, spec):
-        # A .piopm naming this spec is the exact shape the serial pass
-        # trusts; a dir without one is overwritten by pio's own install
-        for dest in dests:
-            remove_dir(spec, dest)
-    else:
-        print(f"No resolvable destination to clean for {spec}", flush=True)
-
-
-def spec_key(spec) -> str | None:
-    """The destination identity of a spec: PlatformIO installs by package
-    name, so two specs sharing a name share a directory. ``None`` means
-    the name could not be derived; such a spec must stay out of the wave
-    (a raw-string key would break the one-per-destination guarantee)."""
-    name = BasePackageManager.ensure_spec(spec).name
-    return name.lower() if name else None
-
-
-def dependency_specs(manager, specs: list) -> list:
-    """``(spec, compatibility)`` registry dependencies of installed
-    packages, from local manifest reads. Name-only dependencies
-    (platform-bundled libs like SPI) stay with the ``pkg install`` pass;
-    the compatibility qualifiers mirror pio's install_dependency, so a
-    qualified dep resolves to the same package the serial pass picks."""
-    return [
-        (manager.dependency_to_spec(dep), PackageCompatibility.from_dependency(dep))
-        for spec in specs
-        if (pkg := manager.get_package(spec)) is not None
-        for dep in manager.get_pkg_dependencies(pkg) or []
-        if dep.get("owner") or dep.get("version")
-    ]
-
-
-def parallel_install(manager_cls, specs: list, prior_names: set | None = None) -> None:
-    """Best-effort parallel top-level install.
-
-    PlatformIO's own installer downloads and unpacks one package at a time
-    on one core. Dependencies are skipped (two packages sharing one must
-    not extract into the same directory from two threads) and failures are
-    only reported: the stock ``pkg install`` pass afterwards installs
-    whatever is missing and is the authority on the final state.
-    """
-    if not specs:
-        return
-    manager = manager_cls(None)
-    # One spec per destination: two threads must not extract into the
-    # same directory. Second versions of a name and URL specs (their dir
-    # comes from the archive manifest) stay with the pkg install pass.
-    seen_names: set = prior_names if prior_names is not None else set()
-    # Wave-1 items are strings; dependency waves carry (spec, compatibility)
-    pairs = [item if isinstance(item, tuple) else (item, None) for item in specs]
-    unique = {}
-    for spec, compat in pairs:
-        # Normalize once: a dependency's URL version surfaces as spec.uri
-        parsed = BasePackageManager.ensure_spec(spec)
-        if parsed.uri:
-            continue
-        if (key := spec_key(parsed)) is None:
-            # No name, no destination identity; leave it to the serial pass
-            print(f"Skipping unresolvable spec {spec!r} in the wave", flush=True)
-            continue
-        unique.setdefault(key, (spec, compat))  # first-wins, like pio's walk
-    pending = [
-        (spec, compat)
-        for spec, compat in unique.values()
-        if not manager.get_package(spec)
-    ]
-    if not pending:
-        # Nothing to install, but a warm store's dependencies must still
-        # feed the next wave (a transitive dep may be missing)
-        _next_wave(manager_cls, manager, unique, seen_names)
-        return
-    workers = min(len(pending), MAX_WORKERS)
-    # One manager per worker (_install mutates instance state); built
-    # serially because construction rewires the shared manager logger
-    managers: queue.SimpleQueue = queue.SimpleQueue()
-    for _ in range(workers):
-        managers.put(manager_cls(None))
-    local = threading.local()
-
-    def install_one(item) -> bool:
-        spec, compat = item
-        if (mgr := getattr(local, "mgr", None)) is None:
-            mgr = local.mgr = managers.get_nowait()
-        try:
-            mgr._install(  # noqa: SLF001
-                spec, skip_dependencies=True, compatibility=compat
-            )
-            return True
-        except Exception as err:  # noqa: BLE001
-            print(f"Pre-install of {spec} failed ({err!r})", flush=True)
-            cleanup_or_die(mgr, spec)
-            return False
-        except BaseException:
-            # A worker SystemExit (main() guards against it) must not skip
-            # the cleanup and leave a torn dir the serial pass trusts
-            cleanup_or_die(mgr, spec)
-            raise
-
-    print(f"Preinstalling {len(pending)} package(s) with {workers} workers", flush=True)
-    # The serial getter calls create pio's lazy dirs (made without
-    # exist_ok) before cold-cache workers can race the creation
-    manager.get_download_dir()
-    manager.get_tmp_dir()
-    ContentCache("http")
-    cwd = Path.cwd()
-    manager.lock()
-    try:
-        with ThreadPoolExecutor(max_workers=workers) as ex:
-            futures = [ex.submit(install_one, item) for item in pending]
-        # The with-block joined every future; drain them all so a
-        # concurrent CleanupError is never dropped
-        errors = [err for f in futures if (err := f.exception()) is not None]
-        for err in errors:
-            # Every failure is on the record; the raised one is a summary
-            print(f"Wave failure: {err!r}", flush=True)
-        if errors:
-            raise next((e for e in errors if isinstance(e, CleanupError)), errors[0])
-        results = [f.result() for f in futures]
-    finally:
-        try:
-            manager.unlock()
-        except Exception as unlock_err:  # noqa: BLE001
-            # A held flock would hang the serial pass in another process;
-            # failing loudly beats an unexplained stuck docker build. Any
-            # in-flight error stays attached as the context.
-            raise LockReleaseError(
-                f"could not release the manager lock: {unlock_err!r}"
-            ) from unlock_err
-        # Worker postinstall scripts chdir process-wide (pio's fs.cd);
-        # restore between waves. The serial pass pins its own cwd.
-        with suppress(OSError):
-            os.chdir(cwd)
-    if failures := len(results) - sum(results):
-        # The stock pass retries CLI specs and re-walks installed
-        # packages' dependencies, so failed deps retry too
-        print(
-            f"Pre-install failed for {failures} of {len(results)} package(s); "
-            "pkg install retries them serially",
-            flush=True,
-        )
-
-    # Waves skip dependencies (a shared one must not extract from two
-    # threads); the installed manifests feed the next wave
-    _next_wave(manager_cls, manager, unique, seen_names)
-
-
-def _next_wave(manager_cls, manager, unique: dict, seen_names: set) -> None:
-    """Queue the dependency wave for every requested spec, installed or
-    freshly waved; a warm store can still be missing a transitive dep.
-    Terminates without a cap: each wave admits only never-seen names."""
-    seen_names.update(unique)
-    # The pre-wave get_package calls memoized an empty storage snapshot
-    manager.memcache_reset()
-    next_specs = [
-        item
-        for item in dependency_specs(manager, [spec for spec, _ in unique.values()])
-        if spec_key(item[0]) not in seen_names
-    ]
-    if next_specs:
-        parallel_install(manager_cls, next_specs, seen_names)
-
-
-def build_cli_args(libs: list, platforms: list, tools: list) -> list:
-    return [
-        arg
-        for flag, specs in (("-l", libs), ("-p", platforms), ("-t", tools))
-        for spec in specs
-        for arg in (flag, spec)
-    ]
-
-
-def main() -> None:
-    parser = argparse.ArgumentParser(description="")
-    parser.add_argument("file", help="Path to platformio.ini", nargs=1)
-    parser.add_argument(
-        "-l", "--libraries", help="Install libraries", action="store_true"
-    )
-    parser.add_argument(
-        "-p", "--platforms", help="Install platforms", action="store_true"
-    )
-    parser.add_argument("-t", "--tools", help="Install tools", action="store_true")
-    args = parser.parse_args()
-    start_cwd = Path.cwd()
-    libs, platforms, tools = parse_specs(args.file[0], args)
-
-    # Platforms stay serial: PlatformPackageManager.install runs an
-    # on_installed hook the private _install path would skip
-    if PARALLEL_AVAILABLE:
-        wave_groups = [(ToolPackageManager, tools), (LibraryPackageManager, libs)]
-    else:  # pragma: no cover
-        wave_groups = []
-        print(
-            f"PlatformIO layout changed ({IMPORT_ERROR}); serial install only",
-            flush=True,
-        )
-    for manager_cls, specs in wave_groups:
-        try:
-            parallel_install(manager_cls, specs)
-        except (CleanupError, LockReleaseError, KeyboardInterrupt):
-            # A torn package or a held lock must fail the build
-            raise
-        except BaseException:  # noqa: BLE001
-            # BaseException: a worker postinstall's SystemExit must not
-            # skip the authoritative serial pass (partial deps, exit 0)
-            print("Parallel preinstall failed, falling back to serial", flush=True)
-            traceback.print_exc()
-
-    # Postinstall scripts chdir process-wide (pio's fs.cd captures its
-    # restore path at construction); pin the authoritative pass's cwd
-    subprocess.check_call(
-        ["platformio", "pkg", "install", "-g", *build_cli_args(libs, platforms, tools)],
-        close_fds=False,
-        cwd=start_cwd,
-    )
-
-
-if __name__ == "__main__":
-    main()

script/split_components_for_ci.py

@@ -13,13 +13,20 @@ from __future__ import annotations
 
 import argparse
 from collections import defaultdict
+from collections.abc import Set as AbstractSet
+import contextlib
+from dataclasses import dataclass, field
+import functools
 import json
+import math
 from pathlib import Path
 import sys
+from typing import Any, NamedTuple
 
 # Add esphome to path
 sys.path.insert(0, str(Path(__file__).parent.parent))
 
+from esphome import yaml_util
 from script.analyze_component_buses import (
     ISOLATED_COMPONENTS,
     ISOLATED_SIGNATURE_PREFIX,
@@ -28,13 +35,57 @@ from script.analyze_component_buses import (
     create_grouping_signature,
     merge_compatible_bus_groups,
 )
-from script.helpers import get_component_test_files, split_conflicting_groups
+from script.helpers import (
+    _extract_components_from_yaml,
+    get_component_test_files,
+    parse_test_filename,
+    split_conflicting_groups,
+)
 
-# Weighting for batch creation
-# Isolated components can't be grouped/merged, so they count as 10x
-# Groupable components can be merged into single builds, so they count as 1x
-ISOLATED_WEIGHT = 10
-GROUPABLE_WEIGHT = 1
+# Estimated CI seconds for one build per test platform, fitted to the batch
+# job logs of a full run (every component, PR #20389). Absolute accuracy does
+# not matter, only the ratios between platforms. A full run takes over an
+# hour, so recording real timings is not practical.
+PLATFORM_BUILD_SECONDS = {
+    "host": 10,
+    "esp8266-ard": 30,
+    "rtl87xx-ard": 40,
+    "esp32-c3-idf": 40,
+    "esp32-s2-idf": 40,
+    "rp2350-ard": 40,
+    "rp2040-ard": 45,
+    "esp32-idf": 55,
+    "esp32-p4-idf": 55,
+    "ln882x-ard": 55,
+    "esp32-s3-idf": 60,
+    "bk72xx-ard": 70,
+    "esp32-c6-idf": 70,
+}
+NRF52_BUILD_SECONDS = 50
+# Arduino on any ESP32 chip builds the IDF plus the Arduino core on top
+ESP32_ARDUINO_BUILD_SECONDS = 90
+# The remaining ESP32 chips on ESP-IDF
+DEFAULT_BUILD_SECONDS = 50
+# Each extra component merged into a grouped build makes it larger
+GROUPED_COMPONENT_SECONDS = 3
+# Components that pull in a large library the build compiles from source
+# (TensorFlow Lite Micro, FastLED, LVGL, the HTTP server), added to every
+# build whose test config enables them. Fitted the same way as the platforms.
+HEAVY_COMPONENT_SECONDS: dict[str, int] = {
+    "micro_wake_word": 200,
+    "fastled_clockless": 90,
+    "fastled_spi": 90,
+    "wled": 60,
+    "lvgl": 50,
+    "web_server": 50,
+    "online_image": 45,
+    "prometheus": 35,
+    "bluetooth_proxy": 30,
+}
+# Estimated build seconds per CI runner; sets how many runners are used.
+# Approximate: the runner count charges each grouped build once, but a large
+# group spreads over several runners, which each pay for that build.
+TARGET_BATCH_SECONDS = 600
 
 # Platform used for batching (platform-agnostic batching)
 # Batches are split across CI runners and each runner tests all platforms
@@ -61,10 +112,193 @@ def has_test_files(component_name: str, tests_dir: Path) -> bool:
     )
 
 
+def build_seconds(platform: str) -> int:
+    """Return the estimated CI seconds for one build on a test platform."""
+    if (seconds := PLATFORM_BUILD_SECONDS.get(platform)) is not None:
+        return seconds
+    if platform.startswith("nrf52"):
+        return NRF52_BUILD_SECONDS
+    if platform.startswith("esp32") and platform.endswith("-ard"):
+        return ESP32_ARDUINO_BUILD_SECONDS
+    return DEFAULT_BUILD_SECONDS
+
+
+def _config_components(data: Any) -> set[str]:
+    """Return the components a loaded test config and its ``packages:`` use."""
+    if not isinstance(data, dict):
+        return set()
+    found = _extract_components_from_yaml(data)
+    if isinstance(packages := data.get("packages"), dict):
+        for package in packages.values():
+            if isinstance(package, yaml_util.IncludeFile):
+                found |= _file_components(
+                    (package.parent_file.parent / package.file).resolve()
+                )
+            else:
+                found |= _config_components(package)
+    return found
+
+
+@functools.cache
+def _file_components(path: Path) -> frozenset[str]:
+    """Return the components a config file uses; shared packages load once."""
+    return frozenset(
+        _config_components(yaml_util.load_yaml(path, track_document_range=False))
+    )
+
+
+def heavy_components(component: str, test_file: Path) -> frozenset[str]:
+    """Return the heavy components a test compiles.
+
+    A test counts as heavy when it belongs to a heavy component or when its
+    config enables one, such as a display test that draws with lvgl.
+    """
+    heavy = {component} & HEAVY_COMPONENT_SECONDS.keys()
+    # A config that does not load is left to the build to report
+    with contextlib.suppress(Exception):
+        heavy |= _file_components(test_file.resolve()) & HEAVY_COMPONENT_SECONDS.keys()
+    return frozenset(heavy)
+
+
+def heavy_seconds(heavy: AbstractSet[str]) -> int:
+    """Return the library seconds a set of heavy components adds to a build."""
+    return sum(HEAVY_COMPONENT_SECONDS[name] for name in heavy)
+
+
+class _GroupedShare(NamedTuple):
+    """A component's part in one (signature, platform) shared build."""
+
+    # Seconds of its files on that platform when no other member shares it
+    solo: int
+    # Heavy components its base test brings into the shared build
+    heavy: frozenset[str]
+
+
+@dataclass
+class _BatchItem:
+    """A component and the builds it adds to the batch it lands in."""
+
+    component: str
+    # Builds that always run on their own (isolated tests, variants on a
+    # platform without a base test)
+    own_seconds: int
+    grouped_builds: dict[tuple[str, str], _GroupedShare] = field(default_factory=dict)
+
+    def standalone_seconds(self) -> int:
+        return self.own_seconds + sum(
+            share.solo for share in self.grouped_builds.values()
+        )
+
+
+def _grouped_build_seconds(platform: str, members: list[_GroupedShare]) -> int:
+    """Return the seconds of one (signature, platform) build in a batch.
+
+    A library two members both enable compiles once, so the union is charged.
+    test_build_components only groups when two or more members share the
+    build; a grouped member then skips its variants on that platform.
+    """
+    if len(members) <= 1:
+        return sum(share.solo for share in members)
+    return (
+        build_seconds(platform)
+        + GROUPED_COMPONENT_SECONDS * (len(members) - 1)
+        + heavy_seconds(set().union(*(share.heavy for share in members)))
+    )
+
+
+def _make_item(
+    tests_dir: Path, component: str, signature: str, is_isolated: bool
+) -> _BatchItem:
+    # Per platform: the seconds of its test files, and the base test's heavy
+    # components; a platform without a base test cannot join a shared build
+    seconds_by_platform: dict[str, int] = defaultdict(int)
+    base_heavy: dict[str, frozenset[str]] = {}
+    for test_file in (tests_dir / component).glob("test[.-]*.yaml"):
+        test_name, platform = parse_test_filename(test_file)
+        heavy = heavy_components(component, test_file)
+        seconds_by_platform[platform] += build_seconds(platform) + heavy_seconds(heavy)
+        if test_name == "test":
+            base_heavy[platform] = heavy
+    own_seconds = 0
+    grouped: dict[tuple[str, str], _GroupedShare] = {}
+    for platform, seconds in seconds_by_platform.items():
+        if is_isolated or platform not in base_heavy:
+            own_seconds += seconds
+        else:
+            grouped[(signature, platform)] = _GroupedShare(
+                seconds, base_heavy[platform]
+            )
+    return _BatchItem(component, own_seconds, grouped)
+
+
+class _Batch:
+    """A CI runner's components and its estimated build seconds."""
+
+    def __init__(self) -> None:
+        self.components: list[str] = []
+        self.seconds = 0
+        self.grouped_builds: dict[tuple[str, str], list[_GroupedShare]] = defaultdict(
+            list
+        )
+
+    def added_seconds(self, item: _BatchItem) -> int:
+        """Return the seconds item would add; joining an existing build is cheap."""
+        added = item.own_seconds
+        for (signature, platform), share in item.grouped_builds.items():
+            members = self.grouped_builds.get((signature, platform), [])
+            added += _grouped_build_seconds(
+                platform, [*members, share]
+            ) - _grouped_build_seconds(platform, members)
+        return added
+
+    def add(self, item: _BatchItem) -> None:
+        self.seconds += self.added_seconds(item)
+        for build, share in item.grouped_builds.items():
+            self.grouped_builds[build].append(share)
+        self.components.append(item.component)
+
+
+def balance_batches(items: list[_BatchItem], target_seconds: int) -> list[list[str]]:
+    """Spread items over enough runners to stay near target_seconds each.
+
+    The runner count comes from the total estimate with every grouped build
+    counted once. Heaviest items go first, each to the runner it adds the
+    fewest seconds to among those ending near the lightest, so grouped
+    components follow the builds they can join.
+    """
+    if not items:
+        return []
+    # One batch holding every item counts each grouped build once
+    whole = _Batch()
+    for item in items:
+        whole.add(item)
+    total = whole.seconds
+    count = min(len(items), max(1, math.ceil(total / target_seconds)))
+    batches = [_Batch() for _ in range(count)]
+    # Among the runners ending within slack of the lightest, the one the item
+    # adds the fewest seconds to wins, so a component lands with the bus
+    # group partners it shares builds with; placed away from them it builds
+    # alone on every platform they only share there.
+    slack = target_seconds // 4
+    for item in sorted(items, key=lambda i: (-i.standalone_seconds(), i.component)):
+        costs = [(batch, batch.added_seconds(item)) for batch in batches]
+        ceiling = min(batch.seconds + added for batch, added in costs) + slack
+        batch, _ = min(
+            (
+                (batch, added)
+                for batch, added in costs
+                if batch.seconds + added <= ceiling
+            ),
+            key=lambda cost: (cost[1], cost[0].seconds + cost[1]),
+        )
+        batch.add(item)
+    return [batch.components for batch in batches if batch.components]
+
+
 def create_intelligent_batches(
     components: list[str],
     tests_dir: Path,
-    batch_size: int = 40,
+    target_seconds: int = TARGET_BATCH_SECONDS,
     directly_changed: set[str] | None = None,
 ) -> tuple[list[list[str]], dict[tuple[str, str], list[str]]]:
     """Create batches optimized for component grouping.
@@ -76,7 +310,7 @@ def create_intelligent_batches(
     Args:
         components: List of component names to batch
         tests_dir: Path to tests/components directory
-        batch_size: Target size for each batch
+        target_seconds: Estimated build seconds per batch
         directly_changed: Set of directly changed components (for logging only)
 
     Returns:
@@ -157,51 +391,17 @@ def create_intelligent_batches(
     # actually be split into two at build time -- throwing off CI distribution.
     signature_groups = split_conflicting_groups(signature_groups)
 
-    # Create batches by keeping signature groups together
-    # Components with the same signature stay in the same batches
-    batches = []
-
-    # Sort signature groups to prioritize groupable components
-    # 1. Put "isolated_*" signatures last (can't be grouped with others)
-    # 2. Sort groupable signatures by size (largest first)
-    # 3. "no_buses" components CAN be grouped together
-    def sort_key(item):
-        (_platform, signature), components = item
-        is_isolated = signature.startswith(ISOLATED_SIGNATURE_PREFIX)
-        # Put "isolated_*" last (1), groupable first (0)
-        # Within each category, sort by size (largest first)
-        return (is_isolated, -len(components))
-
-    sorted_groups = sorted(signature_groups.items(), key=sort_key)
-
-    # Strategy: Create batches using weighted sizes
-    # - Isolated components count as 10x (since they can't be grouped/merged)
-    # - Groupable components count as 1x (can be merged into single builds)
-    # - This distributes isolated components across more runners
-    # - Ensures each runner has a good mix of groupable vs isolated components
-
-    current_batch = []
-    current_weight = 0
-
-    for (_platform, signature), group_components in sorted_groups:
-        is_isolated = signature.startswith(ISOLATED_SIGNATURE_PREFIX)
-        weight_per_component = ISOLATED_WEIGHT if is_isolated else GROUPABLE_WEIGHT
-
-        for component in group_components:
-            # Check if adding this component would exceed the batch size
-            if current_weight + weight_per_component > batch_size and current_batch:
-                # Start a new batch
-                batches.append(current_batch)
-                current_batch = []
-                current_weight = 0
-
-            # Add component to current batch
-            current_batch.append(component)
-            current_weight += weight_per_component
-
-    # Don't forget the last batch
-    if current_batch:
-        batches.append(current_batch)
+    items = [
+        _make_item(
+            tests_dir,
+            component,
+            signature,
+            signature.startswith(ISOLATED_SIGNATURE_PREFIX),
+        )
+        for (_platform, signature), group_components in sorted(signature_groups.items())
+        for component in group_components
+    ]
+    batches = balance_batches(items, target_seconds)
 
     return batches, signature_groups
 
@@ -218,11 +418,11 @@ def main() -> int:
         help="JSON array of component names",
     )
     parser.add_argument(
-        "--batch-size",
-        "-b",
+        "--target-seconds",
+        "-t",
         type=int,
-        default=40,
-        help="Target batch size (default: 40, weighted)",
+        default=TARGET_BATCH_SECONDS,
+        help=f"Estimated build seconds per batch (default: {TARGET_BATCH_SECONDS})",
     )
     parser.add_argument(
         "--tests-dir",
@@ -268,7 +468,7 @@ def main() -> int:
     batches, signature_groups = create_intelligent_batches(
         components=components,
         tests_dir=args.tests_dir,
-        batch_size=args.batch_size,
+        target_seconds=args.target_seconds,
         directly_changed=directly_changed,
     )
 
@@ -378,15 +578,15 @@ def main() -> int:
             file=sys.stderr,
         )
 
-    print(f"  - Groupable (weight=1): {groupable_count}", file=sys.stderr)
-    print(f"  - Isolated (weight=10): {isolated_count}", file=sys.stderr)
+    print(f"  - Groupable: {groupable_count}", file=sys.stderr)
+    print(f"  - Isolated: {isolated_count}", file=sys.stderr)
     if actual_components < len(components):
         print(
             f"Components skipped (no test files): {len(components) - actual_components}",
             file=sys.stderr,
         )
     print(f"Number of batches: {len(batches)}", file=sys.stderr)
-    print(f"Batch size target (weighted): {args.batch_size}", file=sys.stderr)
+    print(f"Target build seconds per batch: {args.target_seconds}", file=sys.stderr)
     if len(batches) > 0:
         print(
             f"Average components per batch: {actual_components / len(batches):.1f}",

script/test_build_components.py

@@ -21,6 +21,7 @@ from dataclasses import dataclass
 import hashlib
 import os
 from pathlib import Path
+import shutil
 import subprocess
 import sys
 import time
@@ -83,7 +84,9 @@ def show_disk_space_if_ci(esphome_command: str) -> None:
     print("=" * 80)
     # Use sys.stdout.flush() to ensure output appears immediately
     sys.stdout.flush()
-    subprocess.run(["df", "-h"], check=False, stdout=sys.stdout, stderr=sys.stderr)
+    # Windows has no df
+    if df := shutil.which("df"):
+        subprocess.run([df, "-h"], check=False, stdout=sys.stdout, stderr=sys.stderr)
     print("=" * 80 + "\n")
     sys.stdout.flush()
 
@@ -333,6 +336,12 @@ def extract_platform_with_version(base_file: Path) -> str:
     return base_file.stem.replace("build_components_base.", "")
 
 
+def _wants_skip_bootloader(skip: bool, command: str, platform: str) -> bool:
+    """Every esp32 target: idf variants, and esp32-ard, whose Arduino
+    core builds as an ESP-IDF component under the native toolchain."""
+    return skip and command == "compile" and platform.startswith("esp32")
+
+
 def run_esphome_test(
     component: str,
     test_file: Path,
@@ -344,6 +353,7 @@ def run_esphome_test(
     continue_on_fail: bool,
     use_testing_mode: bool = False,
     toolchain: str | None = None,
+    skip_bootloader: bool = False,
 ) -> TestResult:
     """Run esphome test for a single component.
 
@@ -408,6 +418,8 @@ def run_esphome_test(
 
     # Add command
     cmd.append(esphome_command)
+    if _wants_skip_bootloader(skip_bootloader, esphome_command, platform):
+        cmd.append("--skip-bootloader")
 
     # Add config file
     cmd.append(str(output_file))
@@ -469,6 +481,7 @@ def run_grouped_test(
     esphome_command: str,
     continue_on_fail: bool,
     toolchain: str | None = None,
+    skip_bootloader: bool = False,
 ) -> TestResult:
     """Run esphome test for a group of components with shared bus configs.
 
@@ -554,6 +567,8 @@ def run_grouped_test(
 
     # Add command
     cmd.append(esphome_command)
+    if _wants_skip_bootloader(skip_bootloader, esphome_command, platform):
+        cmd.append("--skip-bootloader")
 
     cmd.append(str(output_file))
 
@@ -614,6 +629,7 @@ def run_grouped_component_tests(
     continue_on_fail: bool,
     additional_isolated: set[str] | None = None,
     toolchain: str | None = None,
+    skip_bootloader: bool = False,
 ) -> tuple[set[tuple[str, str]], list[TestResult]]:
     """Run grouped component tests.
 
@@ -956,6 +972,7 @@ def run_grouped_component_tests(
                 esphome_command=esphome_command,
                 continue_on_fail=continue_on_fail,
                 toolchain=toolchain,
+                skip_bootloader=skip_bootloader,
             )
 
             # Mark all components as tested
@@ -980,6 +997,7 @@ def run_individual_component_test(
     tested_components: set[tuple[str, str]],
     test_results: list[TestResult],
     toolchain: str | None = None,
+    skip_bootloader: bool = False,
 ) -> None:
     """Run an individual component test if not already tested in a group.
 
@@ -1014,6 +1032,7 @@ def run_individual_component_test(
         esphome_command=esphome_command,
         continue_on_fail=continue_on_fail,
         toolchain=toolchain,
+        skip_bootloader=skip_bootloader,
     )
     test_results.append(test_result)
 
@@ -1027,6 +1046,8 @@ def test_components(
     isolated_components: set[str] | None = None,
     base_only: bool = False,
     toolchain: str | None = None,
+    skip_bootloader: bool = False,
+    fail_on_no_tests: bool = False,
 ) -> int:
     """Test components with optional intelligent grouping.
 
@@ -1061,20 +1082,32 @@ def test_components(
     # toolchain build.
     include_validate = esphome_command != "compile"
 
-    # Find all component tests
+    # A blank pattern list would slide into the reference-baseline
+    # fallback and exit green while building nothing
+    if fail_on_no_tests and not any(component_patterns):
+        print("No components requested (blank component list)")
+        return 1
+
+    # Find all component tests; remember which components each pattern
+    # (wildcards included) matched, for the deferred no-tests accounting
     all_tests = {}
+    pattern_components: dict[str, set[str]] = {}
     for pattern in component_patterns:
         # Skip empty patterns (happens when components list is empty string)
         if not pattern:
             continue
-        all_tests.update(
-            find_component_tests(
-                tests_dir, pattern, base_only, include_validate=include_validate
-            )
+        found = find_component_tests(
+            tests_dir, pattern, base_only, include_validate=include_validate
         )
+        pattern_components[pattern] = set(found)
+        all_tests.update(found)
+
+    if fail_on_no_tests and not all_tests:
+        # Nothing matched: fail before the synthetic baseline spends a
+        # compile reporting success on nothing
+        print(f"No components found matching: {component_patterns}")
+        return 1
 
-    # If no components found, build a reference configuration for baseline comparison
-    # Create a synthetic "empty" component test that will build just the base config
     if not all_tests:
         print(f"No components found matching: {component_patterns}")
         print(
@@ -1114,6 +1147,7 @@ def test_components(
             continue_on_fail=continue_on_fail,
             additional_isolated=isolated_components,
             toolchain=toolchain,
+            skip_bootloader=skip_bootloader,
         )
         test_results.extend(grouped_results)
 
@@ -1143,6 +1177,7 @@ def test_components(
                             tested_components=tested_components,
                             test_results=test_results,
                             toolchain=toolchain,
+                            skip_bootloader=skip_bootloader,
                         )
             else:
                 # Platform-specific test
@@ -1176,8 +1211,26 @@ def test_components(
                         tested_components=tested_components,
                         test_results=test_results,
                         toolchain=toolchain,
+                        skip_bootloader=skip_bootloader,
                     )
 
+    silent: list[str] = []
+    if fail_on_no_tests:
+        # A green run that built nothing for a requested pattern must not
+        # pass CI. Per pattern so one silent pattern cannot hide behind
+        # the others; opt-in because some legs legitimately match nothing;
+        # deferred past the summary so reproduce commands still print.
+        built = {c for r in test_results for c in r.components}
+        # A pattern is silent when it matched no fixture, or when none of
+        # its matched components produced a build (wildcards included)
+        silent = [
+            p
+            for p in component_patterns
+            if p and not (pattern_components.get(p, set()) & built)
+        ]
+        if silent:
+            print(f"No tests ran for requested pattern(s): {', '.join(silent)}")
+
     # Separate results into passed and failed
     passed_results = [r for r in test_results if r.success]
     failed_results = [r for r in test_results if not r.success]
@@ -1209,7 +1262,7 @@ def test_components(
     if os.environ.get("GITHUB_STEP_SUMMARY"):
         write_github_summary(test_results, toolchain=toolchain)
 
-    if failed_results:
+    if failed_results or silent:
         return 1
 
     return 0
@@ -1264,6 +1317,18 @@ def main() -> int:
         "--toolchain",
         help="Select toolchain for compiling.",
     )
+    parser.add_argument(
+        "--skip-bootloader",
+        action="store_true",
+        help="Pass --skip-bootloader to esphome compile; component builds "
+        "never flash, and the bootloader is covered by the toolchain jobs",
+    )
+    parser.add_argument(
+        "--fail-on-no-tests",
+        action="store_true",
+        help="Exit non-zero when no test matched (for CI legs whose "
+        "components must all have fixtures)",
+    )
 
     args = parser.parse_args()
 
@@ -1282,8 +1347,10 @@ def main() -> int:
         continue_on_fail=args.continue_on_fail,
         enable_grouping=not args.no_grouping,
         isolated_components=isolated_components,
+        fail_on_no_tests=args.fail_on_no_tests,
         base_only=args.base_only,
         toolchain=args.toolchain,
+        skip_bootloader=args.skip_bootloader,
     )