Classify VCS URIs positively and pin the clone floor

esphome/platformio/prefetch.py

@@ -377,10 +377,21 @@ def _registry_jobs(
     return jobs, failed, installable
 
 
+# The schemes pio's VCSClientFactory dispatches on (Git/Hg/SvnClient)
+_VCS_URI_PREFIXES = ("git+", "hg+", "svn+", "git://", "hg://", "svn://")
+
+
 def _is_vcs_spec_uri(url: str) -> bool:
     """Whether pio's ``install_from_uri`` would clone this URI rather than
-    copy or download it (PackageSpec normalizes git URLs to ``git+``)."""
-    return not url.startswith(("file://", "symlink://", "http://", "https://"))
+    copy or download it (PackageSpec normalizes git URLs to ``git+``).
+    Positive match, with a .git path as the backstop; an unrecognized
+    scheme is skipped here and left to pio run."""
+    if url.startswith(("file://", "symlink://", "http://", "https://")):
+        return False
+    if url.startswith(_VCS_URI_PREFIXES) or url.split("#", 1)[0].endswith(".git"):
+        return True
+    _LOGGER.debug("Unrecognized package URI scheme, leaving it to pio run: %s", url)
+    return False
 
 
 def _spec_name(spec: Any, url: str) -> str:
@@ -390,8 +401,9 @@ def _spec_name(spec: Any, url: str) -> str:
     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."""
+def _entry_is_vcs(entry: tuple[str, Any] | tuple[str, Any, Any]) -> bool:
+    """Whether this (name, spec[, compatibility]) pre-install entry is
+    cloned rather than unpacked."""
     url = entry[1].uri
     return bool(url and _is_vcs_spec_uri(url))
 
@@ -757,7 +769,8 @@ def _preinstall(
     clones = sum(1 for entry in entries if _entry_is_vcs(entry))
     # Clones are network-bound: let them run wide even on small-core
     # runners. Capped, since every worker builds a sibling manager and
-    # may run a postinstall script
+    # may run a postinstall script; the tail of a mixed wave runs its
+    # (largely I/O-bound) extractions at the same width
     workers = min(
         max(get_usable_cpu_count(), min(clones, _CLONE_WORKERS)), len(entries)
     )

tests/unit_tests/test_platformio_prefetch.py

@@ -636,6 +636,12 @@ 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", "trail"]
+    # Positive classification: an unknown scheme is left to pio run
+    with patch("esphome.net_retry.http_request") as mock_head:
+        assert pf._uri_jobs(
+            m, [_FakeSpec(uri="weird://x/pkg", name="weird")], set()
+        ) == ([], 0, [])
+    mock_head.assert_not_called()
 
     m.get_package.return_value = object()  # already installed: warm and silent
     with patch("esphome.net_retry.http_request"):
@@ -1750,6 +1756,56 @@ def test_preinstall_uses_distinct_managers_in_parallel(tmp_path: Path) -> None:
     assert id(seed) not in used
 
 
+def test_preinstall_clone_floor_widens_small_core_pool(tmp_path: Path) -> None:
+    """Network-bound clones run wide even on a 1-CPU host: the barrier
+    deadlocks unless all four clone entries get concurrent workers."""
+    barrier = threading.Barrier(4, timeout=5)
+    used: set = set()
+
+    class _WaveManager:
+        package_dir = str(tmp_path)
+        compatibility = None
+
+        def __init__(self, package_dir, **kwargs) -> None:
+            assert package_dir == str(tmp_path)
+
+        def lock(self) -> None:
+            pass
+
+        def unlock(self) -> None:
+            pass
+
+        def memcache_reset(self) -> None:
+            pass
+
+        def get_tmp_dir(self) -> str:
+            return str(tmp_path)
+
+        def get_download_dir(self) -> str:
+            return str(tmp_path)
+
+        def get_package(self, spec):
+            return None
+
+        def get_pkg_dependencies(self, pkg):
+            return None
+
+        def _install(self, spec, skip_dependencies, compatibility=None) -> None:
+            used.add(id(self))
+            barrier.wait()
+
+    seed = _WaveManager(str(tmp_path))
+    with patch.object(pf, "get_usable_cpu_count", return_value=1):
+        pf._preinstall(
+            seed,
+            [
+                (f"r{i}", _FakeSpec(uri=f"git+https://x/r{i}.git", name=f"r{i}"))
+                for i in range(4)
+            ],
+        )
+    assert len(used) == 4
+
+
 def test_sibling_manager_and_sigterm() -> None:
     """Sibling managers inherit compatibility; SIGTERM raises SystemExit."""
     calls = []