Merge remote-tracking branch 'upstream/dev' into noise-session-resume

script/ci-custom.py

@@ -247,6 +247,8 @@ def lint_ext_check(fname):
         "CLAUDE.md",
         "GEMINI.md",
         ".github/copilot-instructions.md",
+        # Symlink to the real wifi scan_list.h so the test stub cannot drift
+        "tests/integration/fixtures/external_components/wifi/scan_list.h",
     ]
 )
 def lint_executable_bit(fname: Path) -> str | None:
@@ -317,6 +319,154 @@ def lint_no_long_delays(fname, match):
     )
 
 
+# An if/else/for/while whose only body is an unbraced ESP_LOG*() call. When the build's compile-time
+# log level drops that macro, the body expands to nothing and the compiler warns (-Wempty-body).
+# clang-tidy's brace check does not catch these (ShortStatementLines allows short unbraced bodies), so
+# this fills that gap. Matched against comment/string-masked content, so commented-out or quoted code
+# is ignored. Both spellings are covered: core/log.h defines the uppercase ESP_LOG*() macros and
+# the lowercase esph_log_*() ones, and both expand to nothing below their log level.
+# 'for' allows ';' inside its parentheses (the classic C-style header); 'if'/'while' do not, so their
+# condition cannot run past the statement it guards. The 'for' header permits one level of nested
+# parens so it stays bounded to its own statement: without that, it can run past the loop body and
+# latch onto a later ')', mis-reporting the line and skipping the '#' preprocessor check below.
+ESP_LOG_NEEDS_BRACES_RE = re.compile(
+    r"(?:\bif\s*\([^{};]*\)|\bwhile\s*\([^{};]*\)|\bfor\s*\((?:[^{}()]|\([^{}()]*\))*\)|\belse\b)"
+    r"[ \t]*\n?[ \t]*(?:ESP_LOG[A-Z]*|esph_log_[a-z]+)\s*\(",
+    re.MULTILINE,
+)
+
+
+def _mask_cpp_comments_strings(s):
+    """Return s with // and /* */ comments and string/char/raw-string literals blanked to spaces
+    (length and newlines preserved) so a regex only matches real code. Parentheses in real code are
+    kept, so callers can still balance them on the masked text."""
+    out = list(s)
+    i = 0
+    n = len(s)
+    while i < n:
+        c = s[i]
+        # Raw string literal: an optional encoding prefix, then R"delim( ... )delim". The body may
+        # contain quotes, //, /* and unbalanced parens, so it must be consumed as one unit.
+        if c == "R" and i + 1 < n and s[i + 1] == '"':
+            j = i + 2
+            delim = ""
+            while j < n and s[j] not in "( \t\r\n\\" and len(delim) < 16:
+                delim += s[j]
+                j += 1
+            if j < n and s[j] == "(":
+                closing = ")" + delim + '"'
+                end = s.find(closing, j + 1)
+                end = n if end == -1 else end + len(closing)
+                for k in range(i, end):
+                    if s[k] != "\n":
+                        out[k] = " "
+                i = end
+                continue
+            i += 1
+        elif c == "/" and i + 1 < n and s[i + 1] == "/":
+            while i < n and s[i] != "\n":
+                out[i] = " "
+                i += 1
+        elif c == "/" and i + 1 < n and s[i + 1] == "*":
+            out[i] = out[i + 1] = " "
+            i += 2
+            while i < n and not (s[i] == "*" and i + 1 < n and s[i + 1] == "/"):
+                if s[i] != "\n":
+                    out[i] = " "
+                i += 1
+            if i < n:
+                out[i] = " "
+            if i + 1 < n:
+                out[i + 1] = " "
+            i += 2
+        # A "'" after an alphanumeric or '_' is a C++ digit separator (1'000), not a literal opener.
+        elif c == '"' or (
+            c == "'" and not (i and (s[i - 1].isalnum() or s[i - 1] == "_"))
+        ):
+            quote = c
+            out[i] = " "
+            i += 1
+            while i < n:
+                if s[i] == "\\":
+                    out[i] = " "
+                    if i + 1 < n:
+                        out[i + 1] = " "
+                    i += 2
+                    continue
+                if s[i] == quote:
+                    out[i] = " "
+                    i += 1
+                    break
+                if s[i] != "\n":
+                    out[i] = " "
+                i += 1
+        else:
+            i += 1
+    return "".join(out)
+
+
+def _log_statement_end(masked, open_paren):
+    """Index of the ';' ending the ESP_LOG call whose '(' is at open_paren, or None. Balanced on the
+    masked text so quotes/comments inside the arguments do not confuse the paren count."""
+    depth = 0
+    i = open_paren
+    n = len(masked)
+    while i < n:
+        ch = masked[i]
+        if ch == "(":
+            depth += 1
+        elif ch == ")":
+            depth -= 1
+            if depth == 0:
+                j = i + 1
+                while j < n and masked[j] != ";":
+                    if not masked[j].isspace():
+                        return None
+                    j += 1
+                return j if j < n else None
+        i += 1
+    return None
+
+
+@lint_content_check(include=cpp_include)
+def lint_esp_log_needs_braces(fname, content):
+    # Cheap bailout: no log call means nothing to flag, and skips masking the file entirely.
+    if "ESP_LOG" not in content and "esph_log_" not in content:
+        return []
+    masked = _mask_cpp_comments_strings(content)
+    errors = []
+    for match in ESP_LOG_NEEDS_BRACES_RE.finditer(masked):
+        pos = match.start()
+        line_start = content.rfind("\n", 0, pos) + 1
+        # Skip preprocessor conditionals (#if/#else/#elif): not C++ control statements.
+        if content[line_start:pos].lstrip().startswith("#"):
+            continue
+        # A '// NOLINT' may sit at the end of the log line (where the message says to put it) or on the
+        # control-statement line, so scan the whole statement rather than only up to the ESP_LOG token.
+        stmt_end = _log_statement_end(masked, match.end() - 1)
+        nolint_end = (
+            content.find("\n", stmt_end) if stmt_end is not None else match.end()
+        )
+        if nolint_end == -1:
+            nolint_end = len(content)
+        if "NOLINT" in content[pos:nolint_end]:
+            continue
+        snippet = content[pos : match.end()].replace("\n", " ").strip()
+        errors.append(
+            (
+                content.count("\n", 0, pos) + 1,
+                pos - line_start + 1,
+                (
+                    f"{highlight(snippet)} - an if/else/for/while body that is a single log "
+                    "call must be wrapped in braces. When the log level compiles the macro out, the "
+                    "body becomes empty and the compiler warns (-Wempty-body). Add { } around the "
+                    "log call (or a '// NOLINT' comment if this is genuinely intended)."
+                ),
+            )
+        )
+    return errors
+
+
 @lint_content_check(
     include=[
         "esphome/const.py",

script/clang_tidy_hash.py

@@ -1,13 +1,10 @@
-"""Files that affect clang-tidy results, and a content hash over them.
+"""Files that affect clang-tidy results and the idedata built from them.
 
-``CLANG_TIDY_GLOBAL_FILES`` (plus ``SDKCONFIG_DEFAULTS_PREFIX``) is the single
-source of truth for which files influence clang-tidy output. A change to any of
-them can surface warnings in source files a PR didn't touch, so:
-
-* ``script/determine-jobs.py`` runs a full clang-tidy scan when one changes, and
-* ``calculate_clang_tidy_hash()`` folds them into the idedata cache key used by
-  ``script/helpers.py`` (a content hash, unlike an mtime check, stays correct
-  across git checkouts).
+``CLANG_TIDY_GLOBAL_FILES`` (plus ``SDKCONFIG_DEFAULTS_PREFIX``) lists the files
+that influence clang-tidy output; ``script/determine-jobs.py`` runs a full scan
+when one changes. ``ESP_IDF_INFRA_TRIGGER_*`` lists the native ESP-IDF build
+code. ``idedata_cache_hash()`` folds the right set into the idedata cache key
+used by ``script/helpers.py`` and the CI cache action.
 """
 
 from __future__ import annotations
@@ -31,6 +28,18 @@ CLANG_TIDY_GLOBAL_FILES = (
 # this prefix at the repo root.
 SDKCONFIG_DEFAULTS_PREFIX = "sdkconfig.defaults"
 
+# Native ESP-IDF build infra: determine-jobs forces an esp32 compile when these
+# change, and they feed the clang-tidy idedata cache key.
+ESP_IDF_INFRA_TRIGGER_PATH_PREFIXES = ("esphome/espidf/", "esphome/build_helpers/")
+ESP_IDF_INFRA_TRIGGER_FILES = frozenset(
+    {
+        "esphome/build_gen/espidf.py",
+        "esphome/framework_helpers.py",
+        "esphome/platformio/library.py",
+        "esphome/platformio/extra_script.py",
+    }
+)
+
 
 def read_file_bytes(path: Path) -> bytes:
     """Read bytes from a file."""
@@ -66,3 +75,33 @@ def calculate_clang_tidy_hash(repo_root: Path | None = None) -> str:
         hasher.update(read_file_bytes(path))
 
     return hasher.hexdigest()
+
+
+def calculate_idedata_cache_hash(repo_root: Path | None = None) -> str:
+    """Clang-tidy hash plus the Python that generates the idedata."""
+    repo_root = _ensure_repo_root(repo_root)
+
+    hasher = hashlib.sha256()
+    hasher.update(calculate_clang_tidy_hash(repo_root).encode())
+
+    paths = {repo_root / name for name in ESP_IDF_INFRA_TRIGGER_FILES}
+    for prefix in ESP_IDF_INFRA_TRIGGER_PATH_PREFIXES:
+        # .pyc files appear between the CI key computation and load_idedata's.
+        paths.update(
+            path
+            for path in (repo_root / prefix).rglob("*")
+            if "__pycache__" not in path.parts
+        )
+    for path in sorted(paths):
+        if path.is_file():
+            hasher.update(str(path.relative_to(repo_root)).encode())
+            hasher.update(read_file_bytes(path))
+
+    return hasher.hexdigest()
+
+
+def idedata_cache_hash(environment: str, repo_root: Path | None = None) -> str:
+    """Hash gating the cached idedata of one clang-tidy environment."""
+    if "esp32" in environment:
+        return calculate_idedata_cache_hash(repo_root)
+    return calculate_clang_tidy_hash(repo_root)

script/determine-jobs.py

@@ -53,16 +53,25 @@ from collections import Counter
 from enum import StrEnum
 from functools import cache
 import json
+import math
 import os
 from pathlib import Path
+import statistics
 import sys
 from typing import Any
 
-from clang_tidy_hash import CLANG_TIDY_GLOBAL_FILES, SDKCONFIG_DEFAULTS_PREFIX
+from clang_tidy_hash import (
+    CLANG_TIDY_GLOBAL_FILES,
+    ESP_IDF_INFRA_TRIGGER_FILES,
+    ESP_IDF_INFRA_TRIGGER_PATH_PREFIXES,
+    SDKCONFIG_DEFAULTS_PREFIX,
+)
 from helpers import (
     CPP_FILE_EXTENSIONS,
     ESPHOME_TESTS_COMPONENTS_PATH,
+    INTEGRATION_TESTS_PATH,
     PYTHON_FILE_EXTENSIONS,
+    all_integration_test_files,
     base_python_changed,
     changed_files,
     core_changed,
@@ -78,6 +87,8 @@ from helpers import (
     get_target_branch,
     git_ls_files,
     is_validate_only_file,
+    load_integration_durations,
+    lpt_partition,
     root_path,
 )
 from split_components_for_ci import create_intelligent_batches
@@ -91,24 +102,24 @@ CLANG_TIDY_SPLIT_THRESHOLD = 65
 # Isolated components count as 10x, groupable components count as 1x
 COMPONENT_TEST_BATCH_SIZE = 40
 
-# Integration test bucketing: when more than the threshold tests are scheduled,
-# fan out across this many parallel jobs. Below the threshold, a single job runs.
+# 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
+# sizes the bucket count for small subsets.
 INTEGRATION_TESTS_SPLIT_THRESHOLD = 10
-INTEGRATION_TESTS_SPLIT_BUCKETS = 3
+INTEGRATION_TESTS_SPLIT_BUCKETS = 5
+INTEGRATION_TESTS_TARGET_BUCKET_WEIGHT = 360.0
 
-
-def _split_list(items: list[str], n: int) -> list[list[str]]:
-    """Split a list into n roughly-equal contiguous parts (matches script/clang-tidy)."""
-    k, m = divmod(len(items), n)
-    return [items[i * k + min(i, m) : (i + 1) * k + min(i + 1, m)] for i in range(n)]
-
-
-def _all_integration_test_files() -> list[str]:
-    """Return all integration test file paths, sorted, relative to repo root."""
-    return sorted(
-        str(p.relative_to(root_path))
-        for p in (Path(root_path) / "tests" / "integration").glob("test_*.py")
-    )
+# 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
+INTEGRATION_TESTS_TRIGGER_FILES = frozenset(
+    {
+        "requirements.txt",
+        "requirements_test.txt",
+        "tests/integration/fixtures/cache_init.yaml",
+    }
+)
 
 
 def _compute_integration_test_buckets(
@@ -117,7 +128,7 @@ def _compute_integration_test_buckets(
 ) -> tuple[bool, list[dict[str, Any]]]:
     """Compute (run_integration, buckets) from the determine_integration_tests result.
 
-    Pure function for unit testing — no I/O beyond `_all_integration_test_files`
+    Pure function for unit testing — no I/O beyond `all_integration_test_files`
     when `integration_run_all` is set.
 
     `buckets` is a list of `{name, tests}` dicts where `tests` is a JSON-friendly
@@ -125,7 +136,7 @@ def _compute_integration_test_buckets(
     shell word-splitting / glob hazards.
     """
     if integration_run_all:
-        files = _all_integration_test_files()
+        files = all_integration_test_files()
     else:
         files = sorted(integration_test_files)
 
@@ -136,12 +147,23 @@ def _compute_integration_test_buckets(
         return False, []
 
     if len(files) > INTEGRATION_TESTS_SPLIT_THRESHOLD:
-        parts = [
-            part for part in _split_list(files, INTEGRATION_TESTS_SPLIT_BUCKETS) if part
-        ]
+        durations = load_integration_durations()
+        # Unrecorded files weigh the recording's median; with no recording a
+        # file weighs a whole bucket, which keeps the full fan-out
+        default = (
+            statistics.median(durations.values())
+            if durations
+            else INTEGRATION_TESTS_TARGET_BUCKET_WEIGHT
+        )
+        weights = {f: durations.get(f, default) for f in files}
+        count = min(
+            INTEGRATION_TESTS_SPLIT_BUCKETS,
+            math.ceil(sum(weights.values()) / INTEGRATION_TESTS_TARGET_BUCKET_WEIGHT),
+        )
+        # count <= SPLIT_BUCKETS < threshold < len(files): no group is empty
+        parts = [sorted(part) for part in lpt_partition(files, weights, count)]
         buckets = [
-            {"name": f"{i + 1}/{len(parts)}", "tests": part}
-            for i, part in enumerate(parts)
+            {"name": f"{i + 1}/{count}", "tests": part} for i, part in enumerate(parts)
         ]
     else:
         buckets = [{"name": "1/1", "tests": files}]
@@ -216,12 +238,15 @@ def determine_integration_tests(branch: str | None = None) -> tuple[bool, list[s
     3. Integration test infrastructure files changed
        - 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
+
     Returns (run_all=False, [test_files...]) when:
 
-    4. Specific integration test files changed
+    5. Specific integration test files changed
        - Only those specific test files are returned
 
-    5. Components used by integration tests (or their dependencies) changed
+    6. Components used by integration tests (or their dependencies) changed
        - Only test files whose fixtures use the changed components are returned
 
     Args:
@@ -239,12 +264,15 @@ def determine_integration_tests(branch: str | None = None) -> tuple[bool, list[s
         # If any core files changed, run all integration tests
         return (True, [])
 
+    if any(f in INTEGRATION_TESTS_TRIGGER_FILES for f in files):
+        return (True, [])
+
     # If infrastructure Python files changed (conftest, utils, etc.), run all tests
     # Excludes test files (test_*.py), fixtures, and non-Python files (README.md)
     if any(
-        f.startswith("tests/integration/")
+        f.startswith(INTEGRATION_TESTS_PATH)
         and f.endswith(".py")
-        and not f.startswith("tests/integration/test_")
+        and not f.startswith(f"{INTEGRATION_TESTS_PATH}test_")
         and "/fixtures/" not in f
         for f in files
     ):
@@ -255,9 +283,9 @@ def determine_integration_tests(branch: str | None = None) -> tuple[bool, list[s
     fixture_to_test_files = get_fixture_to_test_files()
 
     for f in files:
-        if f.startswith("tests/integration/test_") and f.endswith(".py"):
+        if f.startswith(f"{INTEGRATION_TESTS_PATH}test_") and f.endswith(".py"):
             test_files.add(f)
-        elif f.startswith("tests/integration/fixtures/"):
+        elif f.startswith(f"{INTEGRATION_TESTS_PATH}fixtures/"):
             if f.endswith(".yaml"):
                 # Fixture YAML changed - add corresponding test file(s)
                 test_files.update(fixture_to_test_files.get(Path(f).stem, ()))
@@ -524,23 +552,6 @@ def _esp32_platformio_path_or_file_trigger(files: list[str]) -> bool:
     return False
 
 
-# Native-build infra: changes under esphome/espidf/, the shared
-# esphome/build_helpers/ package, or the modules the native ESP-IDF build
-# imports affect every esp32 IDF build (now the default toolchain) but aren't
-# components, so the component matrix wouldn't otherwise force any esp32
-# compile. When they change we fold the `esp32` component into the matrix so
-# the default native-IDF build path is still compiled on an infra-only PR.
-ESP_IDF_INFRA_TRIGGER_PATH_PREFIXES = ("esphome/espidf/", "esphome/build_helpers/")
-ESP_IDF_INFRA_TRIGGER_FILES = frozenset(
-    {
-        "esphome/build_gen/espidf.py",
-        "esphome/framework_helpers.py",
-        "esphome/platformio/library.py",
-        "esphome/platformio/extra_script.py",
-    }
-)
-
-
 def _esp_idf_infra_changed(files: list[str]) -> bool:
     """Whether any changed file is ESP-IDF build/runner infrastructure."""
     for file in files:
@@ -1410,6 +1421,7 @@ def main() -> None:
     output: dict[str, Any] = {
         "core_ci": run_core_ci,
         "integration_tests": run_integration,
+        "integration_run_all": integration_run_all,
         "integration_test_buckets": integration_test_buckets,
         "clang_tidy": run_clang_tidy,
         "clang_tidy_mode": clang_tidy_mode,

script/git-hooks/post-checkout

@@ -0,0 +1,20 @@
+#!/bin/sh
+# Prepare the dev environment for a new checkout or worktree.
+#
+# Installed into the git hooks directory by script/setup. Deliberately tiny and
+# self-contained: it stays valid on branches where script/setup does not exist,
+# and simply does nothing there.
+
+# $3 is 1 for a branch checkout, 0 for a file checkout.
+[ "$3" = "1" ] || exit 0
+
+top=$(git rev-parse --show-toplevel 2>/dev/null) || exit 0
+
+# This also runs on ordinary branch switches, where there is nothing to do.
+[ -x "$top/venv/bin/python" ] && exit 0
+[ -x "$top/script/setup" ] || exit 0
+
+# Clear VIRTUAL_ENV so a checkout made from a shell with an environment already
+# activated still gets its own, rather than having the active one repointed at
+# this working tree.
+exec env -u VIRTUAL_ENV "$top/script/setup"

script/helpers.py

@@ -43,6 +43,53 @@ ESPHOME_TESTS_COMPONENTS_PATH = "tests/components/"
 # Tuple of component and test paths for efficient startswith checks
 COMPONENT_AND_TESTS_PATHS = (ESPHOME_COMPONENTS_PATH, ESPHOME_TESTS_COMPONENTS_PATH)
 
+# Integration tests path prefix
+INTEGRATION_TESTS_PATH = "tests/integration/"
+
+# Per-file integration test durations from CI junit output; shared by the
+# reader (determine-jobs) and writer (update_integration_test_durations)
+INTEGRATION_TEST_DURATIONS_FILE = "tests/integration/integration_test_durations.json"
+
+
+def all_integration_test_files() -> list[str]:
+    """Return all integration test file paths, sorted, relative to repo root."""
+    return sorted(
+        p.relative_to(root_path).as_posix()
+        for p in (Path(root_path) / "tests" / "integration").glob("test_*.py")
+    )
+
+
+def load_integration_durations() -> dict[str, float]:
+    """Return recorded per-file pytest durations in seconds; empty when unavailable."""
+    try:
+        raw = json.loads(
+            (Path(root_path) / INTEGRATION_TEST_DURATIONS_FILE).read_text()
+        )
+        if not isinstance(raw, dict):
+            print(
+                f"integration durations unavailable: expected an object, "
+                f"got {type(raw).__name__}",
+                file=sys.stderr,
+            )
+            return {}
+    except (OSError, ValueError) as err:
+        # The file ships in the repo; degrade to unweighted bucketing, loudly
+        print(f"integration durations unavailable: {err}", file=sys.stderr)
+        return {}
+    durations = {
+        key: seconds
+        for key, value in raw.items()
+        if isinstance(value, (int, float)) and (seconds := float(value)) > 0
+    }
+    if len(durations) != len(raw):
+        # One bad entry must not discard the whole recording
+        print(
+            f"dropped {len(raw) - len(durations)} invalid duration entries",
+            file=sys.stderr,
+        )
+    return durations
+
+
 # Base bus components - these ARE the bus implementations and should not
 # be flagged as needing migration since they are the platform/base components
 BASE_BUS_COMPONENTS = {
@@ -809,17 +856,14 @@ def load_idedata(environment: str) -> dict[str, Any]:
     start_time = time.time()
     print(f"Loading IDE data for environment '{environment}'...")
 
-    # Reuse the clang-tidy input hash as the cache key: it already covers every
-    # file baked into the generated idedata (platformio.ini, sdkconfig.defaults,
-    # esphome/idf_component.yml), so this can't drift from that file list. A
-    # content hash -- unlike an mtime comparison -- stays correct across git
-    # checkouts, which don't preserve mtimes.
-    from clang_tidy_hash import calculate_clang_tidy_hash
+    # Content hash of the idedata inputs (data files and the generator code); a
+    # content hash, unlike mtimes, stays correct across git checkouts.
+    from clang_tidy_hash import idedata_cache_hash
 
     temp_idedata = Path(temp_folder) / f"idedata-{environment}.json"
     temp_hash = Path(temp_folder) / f"idedata-{environment}.hash"
 
-    cache_key = calculate_clang_tidy_hash()
+    cache_key = idedata_cache_hash(environment)
     changed = (
         not temp_idedata.is_file()
         or not temp_hash.is_file()
@@ -1548,3 +1592,21 @@ def get_cpp_changed_components(files: list[str]) -> list[str]:
         if file.startswith(ESPHOME_COMPONENTS_PATH):
             affected.update(find_children_of_component(components_graph, component))
     return sorted(c for c in affected if has_cpp_unit_tests(c, tests_dir))
+
+
+def lpt_partition(
+    items: list[str], weights: dict[str, float], count: int
+) -> list[list[str]]:
+    """Partition items into `count` weight-balanced groups (LPT greedy).
+
+    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.
+    """
+    groups: list[list[str]] = [[] for _ in range(count)]
+    group_weights = [0.0] * count
+    for item in sorted(items, key=lambda i: -weights[i]):
+        lightest = min(range(count), key=group_weights.__getitem__)
+        groups[lightest].append(item)
+        group_weights[lightest] += weights[item]
+    return groups

script/platformio_install_deps.py

@@ -3,58 +3,375 @@
 # 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
 
-config = configparser.ConfigParser(inline_comment_prefixes=(";",))
+# 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
 
-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")
+    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)
 
-args = parser.parse_args()
-
-config.read(args.file)
+# 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
 
 
-libs = []
-tools = []
-platforms = []
-# Extract from every lib_deps key in all sections
-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("-l")
-            libs.append(lib_dep)
-    if "platform" in conf and args.platforms:
-        platforms.append("-p")
-        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("-t")
-            tools.append(tool)
+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."""
 
-subprocess.check_call(
-    ["platformio", "pkg", "install", "-g", *libs, *platforms, *tools], close_fds=False
-)
+
+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/setup

@@ -7,13 +7,19 @@ cd "$(dirname "$0")/.."
 if [ -n "$VIRTUAL_ENV" ]; then
   # A virtual environment is already active (e.g. the devcontainer's pre-provisioned
   # esphome-venv). Install into it rather than creating a ./venv in the workspace.
-  created_venv=false
+  venv_state=active
+elif [ -x venv/bin/python ]; then
+  # Reuse the environment from an earlier run, so this script can be run again
+  # at any time to pick up dependency changes.
+  venv_state=reused
+  source venv/bin/activate
 else
-  created_venv=true
+  venv_state=created
+  # --clear replaces a partial environment left behind by an interrupted run.
   if [ -x "$(command -v uv)" ]; then
-    uv venv --seed venv
+    uv venv --clear --seed venv
   else
-    python3 -m venv venv
+    python3 -m venv --clear venv
   fi
   source venv/bin/activate
 fi
@@ -25,20 +31,41 @@ fi
 uv pip install setuptools wheel
 uv pip install -e ".[dev,test]" --config-settings editable_mode=compat
 
-# --overwrite replaces any hook already in place. Without it, prek finds a
-# previously installed pre-commit hook, moves it aside to
-# .git/hooks/pre-commit.legacy and keeps calling it, so every commit would
-# run both tools.
-prek install --overwrite
+# A worktree shares one git hooks directory with the main checkout it was
+# created from, so hooks are installed from the main checkout only. Installing
+# from a worktree would point the shared hook at that worktree's virtual
+# environment, breaking it for everyone once the worktree is removed.
+git_dir="$(git rev-parse --absolute-git-dir 2>/dev/null || true)"
+common_dir="$(git rev-parse --path-format=absolute --git-common-dir 2>/dev/null || true)"
+if [ -n "$common_dir" ] && [ "$git_dir" = "$common_dir" ]; then
+  # --overwrite replaces any hook already in place. Without it, prek finds a
+  # previously installed pre-commit hook, moves it aside to
+  # .git/hooks/pre-commit.legacy and keeps calling it, so every commit would
+  # run both tools.
+  prek install --overwrite
+
+  # Prepares the virtual environment for new checkouts and worktrees. Installed
+  # once here, it covers every worktree created from this checkout.
+  if [ -d "$common_dir/hooks" ]; then
+    cp script/git-hooks/post-checkout "$common_dir/hooks/post-checkout"
+    chmod +x "$common_dir/hooks/post-checkout"
+  fi
+fi
 
 mkdir -p .temp
 
 echo
 echo
-if [ "$created_venv" = true ]; then
-  echo "Virtual environment created at ./venv. Run 'source venv/bin/activate' to use it."
-else
-  echo "Dependencies installed into the active virtual environment:"
-  echo "  $VIRTUAL_ENV"
-  echo "It is already active in this shell, so no 'source venv/bin/activate' is needed."
-fi
+case "$venv_state" in
+  created)
+    echo "Virtual environment created at ./venv. Run 'source venv/bin/activate' to use it."
+    ;;
+  reused)
+    echo "Dependencies updated in the existing ./venv. Run 'source venv/bin/activate' to use it."
+    ;;
+  active)
+    echo "Dependencies installed into the active virtual environment:"
+    echo "  $VIRTUAL_ENV"
+    echo "It is already active in this shell, so no 'source venv/bin/activate' is needed."
+    ;;
+esac

script/update_integration_test_durations.py

@@ -0,0 +1,119 @@
+#!/usr/bin/env python3
+"""Merge CI junit output into tests/integration/integration_test_durations.json.
+
+The integration-tests CI job uploads one junit XML artifact per bucket on
+full matrix dev runs. Download a run's artifacts and merge the per file
+durations into the recording used by script/determine-jobs.py:
+
+    gh run download <run-id> --repo esphome/esphome -p "junit-integration-*" -D /tmp/junit
+    script/update_integration_test_durations.py /tmp/junit
+
+Missing files keep their previous recording and deleted files drop out; a
+run covering under 90% of the test files aborts unless --allow-partial.
+"""
+
+from __future__ import annotations
+
+import argparse
+from collections import defaultdict
+import json
+from pathlib import Path
+import sys
+import xml.etree.ElementTree as ET
+
+from helpers import (
+    INTEGRATION_TEST_DURATIONS_FILE,
+    INTEGRATION_TESTS_PATH,
+    all_integration_test_files,
+    load_integration_durations,
+    root_path,
+)
+
+DURATIONS_FILE = Path(root_path) / INTEGRATION_TEST_DURATIONS_FILE
+MIN_COVERAGE = 0.9
+# Exit code for the expected "run covers too few files" refusal, so the
+# refresh workflow can move on to the next candidate run
+EXIT_LOW_COVERAGE = 3
+
+
+def collect_durations(junit_dir: Path, known_files: set[str]) -> dict[str, float]:
+    """Sum junit testcase times per integration test file, in seconds."""
+    durations: defaultdict[str, float] = defaultdict(float)
+    unmatched = 0
+    xml_files = sorted(junit_dir.rglob("*.xml"))
+    if not xml_files:
+        raise SystemExit(f"no junit XML files found under {junit_dir}")
+    for xml_file in xml_files:
+        for testcase in ET.parse(xml_file).getroot().iter("testcase"):
+            # Skipped/errored testcases carry time="0"; recording them would
+            # overwrite a good previous duration
+            if any(
+                testcase.find(tag) is not None
+                for tag in ("skipped", "error", "failure")
+            ):
+                continue
+            # classname is the dotted module plus any test class, e.g.
+            # tests.integration.test_x or tests.integration.test_x.TestFoo
+            parts = testcase.get("classname", "").split(".")
+            if parts[:2] != ["tests", "integration"] or len(parts) < 3:
+                unmatched += 1
+                continue
+            path = f"{INTEGRATION_TESTS_PATH}{parts[2]}.py"
+            if path not in known_files:
+                print(f"skipping unknown test module {path}", file=sys.stderr)
+                continue
+            durations[path] += float(testcase.get("time", "0"))
+    if unmatched:
+        # A junit naming change would otherwise shrink the recording silently
+        raise SystemExit(
+            f"{unmatched} testcases with unexpected classnames; the junit layout changed"
+        )
+    # An all-skipped file totals 0.0; let the merge keep its previous entry
+    return {k: v for k, v in durations.items() if v > 0}
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "junit_dir", type=Path, help="directory containing downloaded junit XML files"
+    )
+    parser.add_argument(
+        "--allow-partial",
+        action="store_true",
+        help="merge a run covering under 90%% of the test files",
+    )
+    args = parser.parse_args()
+
+    on_disk = set(all_integration_test_files())
+    if not on_disk:
+        raise SystemExit("no integration test files found; wrong checkout root?")
+    collected = collect_durations(args.junit_dir, on_disk)
+    coverage = len(collected.keys() & on_disk) / len(on_disk)
+    if coverage < MIN_COVERAGE and not args.allow_partial:
+        print(
+            f"artifacts cover only {coverage:.0%} of {len(on_disk)} test files; "
+            "use a full matrix run or pass --allow-partial to merge anyway",
+            file=sys.stderr,
+        )
+        return EXIT_LOW_COVERAGE
+
+    # Validated load: a bad previous entry cannot survive the round trip, and
+    # an unreadable file aborts rather than being overwritten
+    previous = load_integration_durations()
+    if DURATIONS_FILE.is_file() and not previous:
+        raise SystemExit(f"{DURATIONS_FILE} is unreadable; refusing to overwrite it")
+    # New recordings win, absent files keep theirs, deleted files drop out
+    merged = {
+        path: collected.get(path, previous.get(path))
+        for path in sorted(on_disk)
+        if path in collected or path in previous
+    }
+    DURATIONS_FILE.write_text(
+        json.dumps({k: round(v, 2) for k, v in merged.items()}, indent=2) + "\n"
+    )
+    print(f"wrote {len(merged)} entries to {DURATIONS_FILE} ({coverage:.0%} fresh)")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())