diff --git a/script/platformio_install_deps.py b/script/platformio_install_deps.py
index 8f7261efc3..1c4fb28b30 100755
--- a/script/platformio_install_deps.py
+++ b/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()
diff --git a/tests/script/test_platformio_install_deps.py b/tests/script/test_platformio_install_deps.py
new file mode 100644
index 0000000000..a263d7937f
--- /dev/null
+++ b/tests/script/test_platformio_install_deps.py
@@ -0,0 +1,649 @@
+"""Tests for script/platformio_install_deps.py."""
+
+from argparse import Namespace
+import importlib.util
+import inspect
+from pathlib import Path
+import shutil
+import sys
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+from platformio import fs
+from platformio.cache import ContentCache
+from platformio.exception import InvalidJSONFile
+from platformio.package.manager._install import PackageManagerInstallMixin
+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, PackageItem, PackageSpec
+import pytest
+from semantic_version import Version
+
+_SCRIPT = Path(__file__).parents[2] / "script" / "platformio_install_deps.py"
+
+
+def _load_script():
+    spec = importlib.util.spec_from_file_location("platformio_install_deps", _SCRIPT)
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    # The real ContentCache would create dirs under the user's core dir
+    module.ContentCache = lambda *_: None
+    return module
+
+
+def test_spec_key_collapses_destinations() -> None:
+    """Two specs delivering one package share a directory and one key."""
+    mod = _load_script()
+    assert mod.spec_key("esphome/noise-c @ 0.1.21") == "noise-c"
+    assert mod.spec_key("esphome/noise-c@0.1.21") == "noise-c"
+    assert mod.spec_key("ESP32Async/AsyncTCP @ ^3.4.10") == mod.spec_key(
+        "esp32async/asynctcp @ 3.5.0"
+    )
+    url = "https://github.com/pioarduino/platform-espressif32/releases/download/{v}/platform-espressif32.zip"
+    assert mod.spec_key(url.format(v="55.03.311")) == mod.spec_key(
+        url.format(v="54.03.20")
+    )
+
+
+def test_parse_specs_and_cli_args(tmp_path: Path) -> None:
+    """Parsing skips unpinned and interpolated entries; the CLI rebuild
+    keeps the original flag pairing."""
+    ini = tmp_path / "platformio.ini"
+    ini.write_text(
+        "[env:a]\n"
+        "platform = fake/platform@1\n"
+        "lib_deps =\n"
+        "    esphome/noise-c @ 0.1.21\n"
+        "    ${common.lib_deps}\n"
+        "    internal_lib\n"
+        "[env:b]\n"
+        "lib_deps =\n"
+        "    esphome/noise-c @ 0.1.21\n"
+    )
+    mod = _load_script()
+    args = Namespace(libraries=True, platforms=True, tools=False)
+    libs, platforms, tools = mod.parse_specs(str(ini), args)
+    # exact-string duplicates collapse; distinct version pins survive
+    assert libs == ["esphome/noise-c @ 0.1.21"]
+    assert platforms == ["fake/platform@1"]
+    assert tools == []
+    assert mod.build_cli_args(libs, platforms, tools) == [
+        "-l",
+        "esphome/noise-c @ 0.1.21",
+        "-p",
+        "fake/platform@1",
+    ]
+
+
+class _FakeManager:
+    """Scripted manager_cls: records installs, raises on demand."""
+
+    installed: set = set()
+    fail: set = set()
+    calls: list = []
+    lock_events: list = []
+    base_dir: str = ""  # per-test tmp base; set by _reset_fake
+
+    def __init__(self, package_dir) -> None:
+        assert package_dir is None
+
+    @staticmethod
+    def _key(spec) -> str:
+        return spec if isinstance(spec, str) else str(spec)
+
+    def get_package(self, spec):
+        if self._key(spec) in self.installed:
+            return SimpleNamespace(path="/tmp/fake-pkg", spec=self._key(spec))
+        return None
+
+    def memcache_reset(self) -> None:
+        type(self).resets = getattr(type(self), "resets", 0) + 1
+
+    @property
+    def package_dir(self) -> str:
+        return str(Path(type(self).base_dir) / "packages")
+
+    def get_download_dir(self) -> str:
+        return str(Path(type(self).base_dir) / "downloads")
+
+    def get_tmp_dir(self) -> str:
+        return str(Path(type(self).base_dir) / "tmp")
+
+    def lock(self) -> None:
+        type(self).lock_events.append("lock")
+
+    def unlock(self) -> None:
+        type(self).lock_events.append("unlock")
+
+    def _install(self, spec, skip_dependencies, compatibility=None):
+        assert skip_dependencies is True
+        if self._key(spec) in self.fail:
+            raise RuntimeError("boom")
+        type(self).calls.append(spec)
+        type(self).compat_calls.append((self._key(spec), compatibility))
+        type(self).installed.add(self._key(spec))  # atomic under the GIL
+
+    def get_pkg_dependencies(self, pkg):
+        return getattr(type(self), "deps", {}).get(pkg.spec)
+
+    dependency_to_spec = staticmethod(BasePackageManager.dependency_to_spec)
+
+
+def _reset_fake(base_dir: str = "", **kwargs) -> type:
+    # A fresh subclass per test: nothing leaks between tests through the
+    # class-level scripted state
+    return type(
+        "_ScriptedManager",
+        (_FakeManager,),
+        {
+            "base_dir": base_dir,
+            "installed": kwargs.get("installed", set()),
+            "fail": kwargs.get("fail", set()),
+            "calls": [],
+            "compat_calls": [],
+            "lock_events": [],
+        },
+    )
+
+
+def test_parallel_install_empty_specs_is_a_no_op(tmp_path: Path) -> None:
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path))
+    mod.parallel_install(cls, [])
+    assert cls.calls == [] and cls.lock_events == []
+
+
+def test_parallel_install_behavior(tmp_path: Path) -> None:
+    """Duplicates collapse to one install, installed specs are filtered,
+    URL specs stay out of the wave, and the lock wraps the pool."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path), installed={"esphome/already @ 1.0"})
+    mod.parallel_install(
+        cls,
+        [
+            "esphome/noise-c @ 0.1.21",
+            "esphome/noise-c @ 0.1.21",
+            "esphome/already @ 1.0",
+            "https://x/framework.tar.xz",
+        ],
+    )
+    assert cls.calls == ["esphome/noise-c @ 0.1.21"]
+    assert cls.lock_events == ["lock", "unlock"]
+
+
+def test_parallel_install_failure_cleans_torn_destination(
+    tmp_path: Path, capsys
+) -> None:
+    """A failed install resets the memcache, removes what get_package can
+    see, and reports; the others still install."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path), fail={"esphome/bad @ 1.0"})
+
+    removed = []
+
+    torn = str(tmp_path / "packages" / "torn-pkg")  # never created; only rmtree'd
+
+    def get_package(self, spec):
+        if spec == "esphome/bad @ 1.0" and getattr(cls, "resets", 0):
+            return SimpleNamespace(path=torn, spec=spec)
+        return _FakeManager.get_package(self, spec)
+
+    cls.get_package = get_package  # throwaway subclass; nothing to restore
+    with patch.object(mod.fs, "rmtree", side_effect=removed.append):
+        mod.parallel_install(cls, ["esphome/bad @ 1.0", "esphome/good @ 1.0"])
+    assert "esphome/good @ 1.0" in cls.calls
+    assert removed == [torn]
+    out = capsys.readouterr().out
+    assert "Pre-install of esphome/bad @ 1.0 failed" in out
+    assert "Pre-install failed for 1 of 2 package(s)" in out
+
+
+def test_parallel_install_runs_dependency_waves(tmp_path: Path) -> None:
+    """Dependencies of wave-installed packages install in a second wave,
+    deduped by name; name-only platform libs stay with the serial pass."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path))
+    cls.deps = {
+        "esphome/noise-c @ 0.1.21": [
+            {"owner": "esphome", "name": "libsodium", "version": "^1.0"},
+            {"name": "SPI"},
+        ],
+        "esphome/wg @ 1.0": [
+            {"owner": "esphome", "name": "libsodium", "version": "^1.0"},
+        ],
+    }
+    mod.parallel_install(cls, ["esphome/noise-c @ 0.1.21", "esphome/wg @ 1.0"])
+    assert len(cls.calls) == 3  # the shared dep installs exactly once
+    assert {mod.spec_key(c) for c in cls.calls} == {"noise-c", "wg", "libsodium"}
+    # Wave-1 strings carry no compatibility; the dependency wave does
+    compats = dict(cls.compat_calls)
+    assert compats["esphome/noise-c @ 0.1.21"] is None
+    dep_compat = next(v for k, v in cls.compat_calls if "libsodium" in k)
+    assert dep_compat is not None  # mirrors pio's install_dependency
+
+
+def test_dependency_wave_excludes_url_specs(tmp_path: Path) -> None:
+    """A dependency pinned to a URL surfaces as spec.uri; it must stay out
+    of the wave like string URL specs do."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path))
+    cls.deps = {
+        "esphome/noise-c @ 0.1.21": [
+            {"name": "vendored", "version": "https://github.com/x/y.git"},
+        ],
+    }
+    mod.parallel_install(cls, ["esphome/noise-c @ 0.1.21"])
+    assert {mod.spec_key(c) for c in cls.calls} == {"noise-c"}
+
+
+def test_failed_cleanup_fails_the_build(tmp_path: Path) -> None:
+    """A torn destination still on disk after rmtree must fail the build:
+    fs.rmtree never raises (its onexc handler prints), so only the
+    destination's absence proves the cleanup worked."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path), fail={"esphome/bad @ 1.0"})
+    torn = tmp_path / "packages" / "torn-pkg"
+    torn.mkdir(parents=True)
+
+    def get_package(self, spec):
+        if getattr(cls, "resets", 0):
+            return SimpleNamespace(path=str(torn), spec=spec)
+        return None
+
+    cls.get_package = get_package  # throwaway subclass; nothing to restore
+    with (
+        patch.object(mod.fs, "rmtree", lambda path: None),  # onexc swallowed
+        pytest.raises(mod.CleanupError, match="could not remove"),
+    ):
+        mod.parallel_install(cls, ["esphome/bad @ 1.0"])
+    assert cls.lock_events == ["lock", "unlock"]  # still released
+
+
+def test_unverifiable_torn_destination_fails_the_build(tmp_path: Path) -> None:
+    """When the scan fails, the spec's own .piopm decides: an unremovable
+    leftover fails the build."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path), fail={"esphome/bad @ 1.0"})
+    dest = Path(cls.base_dir) / "packages" / "bad"
+    dest.mkdir(parents=True)
+    (dest / ".piopm").write_text('{"spec": {"owner": "esphome", "name": "bad"}}')
+
+    def bad_reset(self):
+        raise OSError("scan broken")
+
+    cls.memcache_reset = bad_reset
+    with (
+        patch.object(mod.fs, "rmtree", lambda path: None),  # onexc swallowed
+        pytest.raises(mod.CleanupError, match="could not remove"),
+    ):
+        mod.parallel_install(cls, ["esphome/bad @ 1.0"])
+
+
+def test_unverifiable_scan_without_leftover_degrades(tmp_path: Path, capsys) -> None:
+    """A failing scan with no destination on disk is never a build
+    failure blaming this spec."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path), fail={"esphome/bad @ 1.0"})
+    resets = {"n": 0}
+
+    def bad_reset(self):
+        # Fail clean_torn's reset; the coordinator's later reset works
+        resets["n"] += 1
+        if resets["n"] <= 1:
+            raise OSError("scan broken")
+
+    cls.memcache_reset = bad_reset
+    mod.parallel_install(cls, ["esphome/bad @ 1.0"])
+    assert "No resolvable destination to clean" in capsys.readouterr().out
+
+
+def test_unresolvable_torn_destination_is_printed(tmp_path: Path, capsys) -> None:
+    """A failed install with no resolvable package prints, so an invisible
+    torn directory is at least traceable."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path), fail={"esphome/bad @ 1.0"})
+    mod.parallel_install(cls, ["esphome/bad @ 1.0"])
+    assert "No resolvable destination to clean" in capsys.readouterr().out
+
+
+def test_unparsable_torn_destination_is_removed(tmp_path: Path, capsys) -> None:
+    """A torn dir get_package cannot resolve but whose .piopm names the
+    spec is removed instead of surviving into the serial pass."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path), fail={"esphome/bad @ 1.0"})
+    dest = Path(cls.base_dir) / "packages" / "bad"
+    dest.mkdir(parents=True)
+    (dest / ".piopm").write_text('{"spec": {"owner": "esphome", "name": "bad"}}')
+
+    with patch.object(mod.fs, "rmtree", shutil.rmtree):
+        mod.parallel_install(cls, ["esphome/bad @ 1.0"])
+    assert not dest.exists()
+    assert "Removed torn destination" in capsys.readouterr().out
+
+
+def test_parse_specs_tools_branch(tmp_path: Path) -> None:
+    """platform_packages parsing keeps owner'd tools and rewrites github
+    URL pins to bare URLs the wave then skips via parsed.uri."""
+    mod = _load_script()
+    ini = tmp_path / "platformio.ini"
+    ini.write_text(
+        "[env:t]\n"
+        "platform_packages =\n"
+        "    ${common.platform_packages}\n"
+        "    platformio/tool-scons@~4.40801.0\n"
+        "    framework-arduinopico@https://github.com/earlephilhower/arduino-pico/releases/download/6.0.0/rp2040-6.0.0.zip\n"
+    )
+    args = Namespace(libraries=False, platforms=False, tools=True)
+    libs, platforms, tools = mod.parse_specs(str(ini), args)
+    assert libs == [] and platforms == []
+    assert tools == [
+        "platformio/tool-scons@~4.40801.0",
+        "https://github.com/earlephilhower/arduino-pico/releases/download/6.0.0/rp2040-6.0.0.zip",
+    ]
+    assert mod.build_cli_args([], [], tools)[:2] == ["-t", tools[0]]
+
+
+def test_warm_store_still_walks_dependencies(tmp_path: Path) -> None:
+    """Already-installed top-level packages still feed the dependency
+    wave; a warm store can be missing a transitive dep."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path), installed={"esphome/noise-c @ 0.1.21"})
+    cls.deps = {
+        "esphome/noise-c @ 0.1.21": [
+            {"owner": "esphome", "name": "libsodium", "version": "^1.0"},
+        ],
+    }
+    mod.parallel_install(cls, ["esphome/noise-c @ 0.1.21"])
+    assert [mod.spec_key(c) for c in cls.calls] == ["libsodium"]
+
+
+def test_worker_system_exit_still_cleans(tmp_path: Path, capsys) -> None:
+    """A worker SystemExit runs the torn cleanup before propagating; the
+    serial pass must never trust its leftovers."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path))
+    torn = tmp_path / "packages" / "torn-pkg"
+    torn.mkdir(parents=True)
+
+    def exiting_install(self, spec, skip_dependencies, compatibility=None):
+        raise SystemExit(0)
+
+    def get_package(self, spec):
+        if getattr(cls, "resets", 0):
+            return SimpleNamespace(path=str(torn), spec=spec)
+        return None
+
+    cls._install = exiting_install
+    cls.get_package = get_package
+
+    def real_rmtree(path):
+        Path(path).rmdir()
+
+    with (
+        patch.object(mod.fs, "rmtree", real_rmtree),
+        pytest.raises(SystemExit),
+    ):
+        mod.parallel_install(cls, ["esphome/bad @ 1.0"])
+    assert not torn.exists()
+
+
+def test_unlock_failure_is_fatal(tmp_path: Path) -> None:
+    """A failed unlock must fail the build: the serial pass in another
+    process would block on the held flock."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path))
+
+    def bad_unlock(self):
+        raise OSError("flock broke")
+
+    cls.unlock = bad_unlock
+    with pytest.raises(mod.LockReleaseError, match="manager lock"):
+        mod.parallel_install(cls, ["esphome/good @ 1.0"])
+
+
+def test_unlock_failure_keeps_inflight_error_as_context(tmp_path: Path) -> None:
+    """An in-flight CleanupError stays attached when the unlock fault
+    takes over the raise."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path), fail={"esphome/bad @ 1.0"})
+    torn = tmp_path / "packages" / "bad"
+    torn.mkdir(parents=True)
+
+    def get_package(self, spec):
+        if getattr(cls, "resets", 0):
+            return SimpleNamespace(path=str(torn), spec=spec)
+        return None
+
+    def bad_unlock(self):
+        raise OSError("flock broke")
+
+    cls.get_package = get_package
+    cls.unlock = bad_unlock
+    with (
+        patch.object(mod.fs, "rmtree", lambda path: None),  # leaves torn
+        pytest.raises(mod.LockReleaseError) as err,
+    ):
+        mod.parallel_install(cls, ["esphome/bad @ 1.0"])
+    assert isinstance(err.value.__cause__.__context__, mod.CleanupError)
+
+
+def test_chdir_failure_does_not_fail_the_wave(tmp_path: Path, monkeypatch) -> None:
+    """A lost cwd is suppressed: further waves may misbehave and fall to
+    the serial pass, whose cwd is pinned."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path))
+    monkeypatch.setattr(mod.os, "chdir", MagicMock(side_effect=OSError("gone")))
+    mod.parallel_install(cls, ["esphome/good @ 1.0"])
+    assert cls.calls == ["esphome/good @ 1.0"]
+
+
+def test_piopm_match_removes_manifest_named_torn_dir(tmp_path: Path, capsys) -> None:
+    """A torn dir named by its manifest (not the registry spec) is found
+    through its .piopm and removed."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path), fail={"esphome/bad @ 1.0"})
+    torn = tmp_path / "packages" / "ManifestName"
+    torn.mkdir(parents=True)
+    (torn / ".piopm").write_text('{"spec": {"owner": "esphome", "name": "bad"}}')
+    innocent = tmp_path / "packages" / "innocent"
+    innocent.mkdir()
+    (innocent / ".piopm").write_text('{"spec": {"owner": "o", "name": "other"}}')
+    with patch.object(mod.fs, "rmtree", shutil.rmtree):
+        mod.parallel_install(cls, ["esphome/bad @ 1.0"])
+    assert not torn.exists()
+    assert innocent.exists()  # another package's valid metadata survives
+    assert "Removed torn destination" in capsys.readouterr().out
+
+
+def test_unscannable_package_dir_fails_the_build(tmp_path: Path) -> None:
+    """A storage dir the cleanup cannot scan is not proof of cleanliness."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path), fail={"esphome/bad @ 1.0"})
+    real_iterdir = Path.iterdir
+
+    def broken_iterdir(self):
+        if self.name == "packages":
+            raise PermissionError("denied")
+        return real_iterdir(self)
+
+    with (
+        patch.object(Path, "iterdir", broken_iterdir),
+        pytest.raises(mod.CleanupError, match="cleanup failed"),
+    ):
+        mod.parallel_install(cls, ["esphome/bad @ 1.0"])
+
+
+def test_stray_file_in_package_dir_is_ignored(tmp_path: Path) -> None:
+    """A plain file (or a pio-link) beside the packages is skipped by
+    pio's own scan and must never hard-fail the build."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path), fail={"esphome/bad @ 1.0"})
+    (tmp_path / "packages").mkdir(parents=True)
+    (tmp_path / "packages" / "stray.pio-link").write_text("x")
+    (tmp_path / "packages" / "no-metadata").mkdir()  # pio overwrites these
+    mod.parallel_install(cls, ["esphome/bad @ 1.0"])
+    assert (tmp_path / "packages" / "stray.pio-link").exists()
+    assert (tmp_path / "packages" / "no-metadata").exists()
+
+
+def test_unreadable_piopm_dir_is_removed(tmp_path: Path) -> None:
+    """A persistently corrupt .piopm under this spec's own name would
+    crash pio's storage scan; the dir is removed rather than left to
+    break the serial pass."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path), fail={"esphome/bad @ 1.0"})
+    torn = tmp_path / "packages" / "bad"
+    torn.mkdir(parents=True)
+    (torn / ".piopm").write_text("{not json")
+    mod.parallel_install(cls, ["esphome/bad @ 1.0"])
+    assert not torn.exists()
+
+
+def test_unreadable_piopm_under_other_name_survives(tmp_path: Path) -> None:
+    """A corrupt .piopm in another package's dir may be a worker mid-copy;
+    a failing spec must not remove a directory it does not own."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path), fail={"esphome/bad @ 1.0"})
+    other = tmp_path / "packages" / "innocent"
+    other.mkdir(parents=True)
+    (other / ".piopm").write_text("{not json")
+    mod.parallel_install(cls, ["esphome/bad @ 1.0"])
+    assert other.exists()
+
+
+def test_unexpected_cleanup_class_becomes_cleanup_error(tmp_path: Path) -> None:
+    """Cleanup failures of any class fail the build; nothing may be
+    downgraded to the serial fallback over a torn directory."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path), fail={"esphome/bad @ 1.0"})
+
+    with (
+        patch.object(
+            mod, "piopm_matches", MagicMock(side_effect=ValueError("bad spec"))
+        ),
+        pytest.raises(mod.CleanupError, match="cleanup failed"),
+    ):
+        mod.parallel_install(cls, ["esphome/bad @ 1.0"])
+
+
+def test_main_cleanup_error_fails_before_generic_fallback(tmp_path: Path) -> None:
+    """A CleanupError must escape main's serial fallback: the clause order
+    decides whether a stuck torn package fails the image build."""
+    mod = _load_script()
+    ini = tmp_path / "platformio.ini"
+    ini.write_text("[env:t]\nlib_deps =\n    esphome/x @ 1.0\n")
+    with (
+        patch.object(
+            mod, "parallel_install", side_effect=mod.CleanupError("stuck torn pkg")
+        ),
+        patch.object(mod.subprocess, "check_call"),
+        patch.object(sys, "argv", ["platformio_install_deps.py", str(ini), "-l"]),
+        pytest.raises(mod.CleanupError),
+    ):
+        mod.main()
+
+
+def test_main_generic_failure_still_runs_serial_pass(tmp_path: Path) -> None:
+    """A non-CleanupError wave failure prints, dumps the traceback, and
+    still reaches the authoritative serial pass with the pinned cwd."""
+    mod = _load_script()
+    ini = tmp_path / "platformio.ini"
+    ini.write_text("[env:t]\nlib_deps =\n    esphome/x @ 1.0\n")
+    with (
+        patch.object(mod, "parallel_install", side_effect=RuntimeError("boom")),
+        patch.object(mod.subprocess, "check_call") as mock_call,
+        patch.object(sys, "argv", ["platformio_install_deps.py", str(ini), "-l"]),
+    ):
+        mod.main()
+    mock_call.assert_called_once()
+    args, kwargs = mock_call.call_args
+    assert args[0][:4] == ["platformio", "pkg", "install", "-g"]
+    assert "esphome/x @ 1.0" in args[0]
+    assert kwargs["cwd"] == Path.cwd()
+
+
+def test_content_cache_creates_its_dir(tmp_path: Path, monkeypatch) -> None:
+    """The cold-cache hardening relies on ContentCache.__init__ creating
+    the namespace dir; pin the side effect, not mere callability."""
+    monkeypatch.setenv("PLATFORMIO_CACHE_DIR", str(tmp_path / "cache"))
+    ContentCache("http")
+    assert (tmp_path / "cache" / "http").is_dir()
+
+
+def test_piopm_matches_without_name_matches_nothing(tmp_path: Path) -> None:
+    """A spec with no derivable name can never match a directory."""
+    mod = _load_script()
+    assert mod.piopm_matches(str(tmp_path), "") == []
+
+
+def test_unresolvable_spec_stays_out_of_the_wave(tmp_path: Path, capsys) -> None:
+    """A spec with no derivable name is left to the serial pass; a raw
+    string key would break the one-per-destination dedupe."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path))
+    nameless = PackageSpec(requirements="^1.0")
+    mod.parallel_install(cls, [nameless])
+    assert cls.calls == []
+    assert "Skipping unresolvable spec" in capsys.readouterr().out
+
+
+def test_parallel_install_unlocks_when_pool_fails(tmp_path: Path) -> None:
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path))
+    with (
+        patch.object(mod, "ThreadPoolExecutor", side_effect=RuntimeError("no")),
+        pytest.raises(RuntimeError),
+    ):
+        mod.parallel_install(cls, ["esphome/a @ 1.0"])
+    assert cls.lock_events == ["lock", "unlock"]
+
+
+def test_parse_specs_unreadable_ini_fails_loudly(tmp_path: Path) -> None:
+    """A bad path must not silently build an image with no dependencies."""
+    mod = _load_script()
+    args = Namespace(libraries=True, platforms=False, tools=False)
+    with pytest.raises(SystemExit):
+        mod.parse_specs(str(tmp_path / "missing.ini"), args)
+
+
+def test_platformio_surface_for_install_deps_script() -> None:
+    """A PlatformIO bump that changes these members must fail here, not
+    silently turn the docker image's parallel preinstall into a no-op."""
+    # The script calls these positionally; pin the positions, not just
+    # membership, so a parameter reorder trips the wire too
+    params = inspect.signature(PackageManagerInstallMixin._install).parameters
+    assert list(params)[1] == "spec"
+    assert "skip_dependencies" in params
+    assert "compatibility" in params
+    for cls in (ToolPackageManager, LibraryPackageManager):
+        assert list(inspect.signature(cls.__init__).parameters)[1] == "package_dir"
+    for name in (
+        "lock",
+        "unlock",
+        "get_package",
+        "memcache_reset",
+        "get_pkg_dependencies",
+        "dependency_to_spec",
+        "get_download_dir",
+        "get_tmp_dir",
+    ):
+        assert callable(getattr(BasePackageManager, name))
+    # Losing any of these turns the wave into main()'s silent serial
+    # fallback: ensure_spec runs in the coordinator, the spec attributes
+    # feed the dedupe, cleanup, and dependency filters
+    assert callable(BasePackageManager.ensure_spec)
+    spec = PackageSpec("owner/name @ ^1.0")
+    assert spec.name == "name"
+    assert spec.owner == "owner"
+    assert spec.uri is None
+    assert spec.external is False
+    assert Version("1.5.0") in spec.requirements
+    # The failure-cleanup path degrades to a single line if these vanish
+    assert callable(fs.rmtree)
+    assert callable(fs.load_json)
+    # piopm_matches only tolerates a corrupt .piopm through this base;
+    # losing it would flip a wave failure from degrade to build failure
+    assert issubclass(InvalidJSONFile, ValueError)
+    assert PackageItem("pkg-dir").path == "pkg-dir"
+    assert callable(PackageCompatibility.from_dependency)