Address review: name the clones-first ordering, harden the name fallback

esphome/platformio/prefetch.py

@@ -16,7 +16,7 @@ name and promote with an atomic rename.
 
 from __future__ import annotations
 
-from collections.abc import Iterator
+from collections.abc import Iterable, Iterator
 from concurrent.futures import ThreadPoolExecutor
 from contextlib import contextmanager, suppress
 import hashlib
@@ -381,9 +381,22 @@ def _is_vcs_spec_uri(url: str) -> bool:
 
 
 def _spec_name(spec: Any, url: str) -> str:
-    """The spec's name; the URL basename fallback is defensive only
-    (PackageSpec derives a name from the URI itself)."""
-    return spec.name or url.split("#", 1)[0].rsplit("/", 1)[-1]
+    """The spec's name; the URL fallbacks are defensive only
+    (PackageSpec derives a name from the URI itself). Never empty:
+    an empty name would coalesce distinct to_install keys."""
+    return spec.name or url.split("#", 1)[0].rstrip("/").rsplit("/", 1)[-1] or url
+
+
+def _entry_is_vcs(entry: tuple[str, Any]) -> bool:
+    """Whether this pre-install entry is cloned rather than unpacked."""
+    url = entry[1].uri
+    return bool(url and _is_vcs_spec_uri(url))
+
+
+def _clones_first(entries: Iterable[tuple[str, Any]]) -> list[tuple[str, Any]]:
+    """Clones first: they wait on the network, so they must not queue
+    behind CPU-bound archive extractions in the pre-install pool."""
+    return sorted(entries, key=lambda entry: not _entry_is_vcs(entry))
 
 
 def _uri_jobs(
@@ -411,6 +424,10 @@ def _uri_jobs(
             continue
         name = _spec_name(spec, url)
         if is_vcs:
+            # Kept even with a derived name, unlike archives below: the
+            # platform tool packages this exists for carry name= without
+            # being "custom". A manifest-name collision fails that wave
+            # entry and pio run installs it serially.
             installable.append((name, spec))
             continue
         # PlatformIO downloads URL specs with no checksum
@@ -750,7 +767,7 @@ def _preinstall(
             raise
 
     _LOGGER.info(
-        "Installing %d PlatformIO package(s) with %d extraction worker(s): %s",
+        "Installing %d PlatformIO package(s) with %d worker(s): %s",
         len(entries),
         workers,
         ", ".join(name for name, *_ in entries),
@@ -931,12 +948,7 @@ def _prefetch(build_dir: Path, env: str) -> None:
             if name not in failed_names
         }
         if to_install:
-            # Clones first: they wait on the network, so they must not
-            # queue behind CPU-bound archive extractions
-            ordered = sorted(
-                to_install.values(),
-                key=lambda entry: not ((url := entry[1].uri) and _is_vcs_spec_uri(url)),
-            )
+            ordered = _clones_first(to_install.values())
             try:
                 _preinstall(mgr, ordered)
                 if is_platform:

tests/unit_tests/test_platformio_prefetch.py

@@ -626,6 +626,8 @@ def test_uri_jobs_vcs_specs_installable_without_probe(tmp_path: Path) -> None:
                 _FakeSpec(uri="hg+https://x/old", name="mercurial"),
                 # Name falls back to the URL basename, fragment excluded
                 _FakeSpec(uri="git+https://x/noname#v2", name=None),
+                # A trailing slash must not derive an empty (colliding) name
+                _FakeSpec(uri="git+https://x/trail/", name=None),
                 _FakeSpec(uri="file:///local/dir", name="local"),
                 _FakeSpec(uri="symlink:///local/dir", name="link"),
             ],
@@ -633,7 +635,7 @@ def test_uri_jobs_vcs_specs_installable_without_probe(tmp_path: Path) -> None:
         )
     mock_head.assert_not_called()
     assert (jobs, failed) == ([], 0)
-    assert [n for n, _ in installable] == ["tool", "mercurial", "noname"]
+    assert [n for n, _ in installable] == ["tool", "mercurial", "noname", "trail"]
 
     m.get_package.return_value = object()  # already installed: warm and silent
     with patch("esphome.net_retry.http_request"):
@@ -642,6 +644,21 @@ def test_uri_jobs_vcs_specs_installable_without_probe(tmp_path: Path) -> None:
         ) == ([], 0, [])
 
 
+def test_clones_first_orders_vcs_before_archives() -> None:
+    """The pre-install pool receives clones first: they wait on the
+    network and must not queue behind CPU-bound archive extractions."""
+    archive = ("zip", _FakeSpec(uri="https://x/a.zip", name="zip"))
+    registry = ("reg", _FakeSpec(uri=None, name="reg"))
+    clone = ("repo", _FakeSpec(uri="git+https://x/repo.git", name="repo"))
+    ordered = pf._clones_first([archive, registry, clone])
+    assert ordered[0] == clone
+    # Stable partition: non-clone relative order is preserved
+    assert ordered[1:] == [archive, registry]
+    assert pf._entry_is_vcs(clone)
+    assert not pf._entry_is_vcs(archive)
+    assert not pf._entry_is_vcs(registry)
+
+
 def test_uri_jobs_head_failure_counts_as_unresolved(
     tmp_path: Path, caplog: pytest.LogCaptureFixture
 ) -> None: