Diagnosable prefetch failures, Ctrl-C-safe pool, typed dependency versions

The library prefetch pool now cancels queued futures on interrupt like
the espidf pool it mirrors, prefetch and HEAD failures log their cause
at debug (a suppressed bar is traceable), and non-2xx HEAD statuses read
as unknown size. normalize_dependencies rejects a non-string version in
both spellings (a container would raise from set.add(); an int fails
opaquely in the registry), and the walk's version-less skip logs the
dependency at debug until the arduino-backend reconciliation lands.

esphome/platformio/library.py

@@ -729,6 +729,15 @@ def normalize_dependencies(
                     manifest_name,
                 )
                 continue
+            if "version" in entry and not isinstance(entry["version"], str):
+                # A container would raise from set.add(); an int fails
+                # opaquely inside the registry resolution
+                _LOGGER.warning(
+                    "Ignoring unrecognized dependency entry %r of %s",
+                    entry,
+                    manifest_name,
+                )
+                continue
             normalized.append(entry)
         return normalized
     if not isinstance(dependencies, (list, tuple)):
@@ -751,6 +760,15 @@ def normalize_dependencies(
                     manifest_name,
                 )
                 continue
+            if "version" in entry and not isinstance(entry["version"], str):
+                # A container would raise from set.add(); an int fails
+                # opaquely inside the registry resolution
+                _LOGGER.warning(
+                    "Ignoring unrecognized dependency entry %r of %s",
+                    entry,
+                    manifest_name,
+                )
+                continue
             normalized.append(entry)
         elif isinstance(entry, str) and entry:
             # PIO also accepts a bare list of names ("dependencies": ["Wire"])
@@ -902,8 +920,12 @@ def _content_lengths(urls: list[str]) -> list[int]:
     def head(url: str) -> int:
         try:
             resp = requests.head(url, timeout=10, allow_redirects=True)
+            if not resp.ok:
+                _LOGGER.debug("HEAD %s returned %s", url, resp.status_code)
+                return 0
             return int(resp.headers.get("content-length", 0))
-        except Exception:  # noqa: BLE001  # pylint: disable=broad-exception-caught
+        except Exception as err:  # noqa: BLE001  # pylint: disable=broad-exception-caught
+            _LOGGER.debug("HEAD %s failed: %s", url, err)
             return 0
 
     with ThreadPoolExecutor(max_workers=min(_DOWNLOAD_WORKERS, len(urls))) as ex:
@@ -954,16 +976,19 @@ def _prefetch_wave(
         tracker = progress.tracker()
         try:
             component.download(salt=salt, namespace=namespace, progress=tracker)
-        except Exception:  # noqa: BLE001  # pylint: disable=broad-exception-caught
+        except Exception as err:  # noqa: BLE001  # pylint: disable=broad-exception-caught
             # The sequential call below retries and reports the failure
+            _LOGGER.debug("Prefetch of %s failed: %s", component.name, err)
             tracker(0)
 
+    ex = ThreadPoolExecutor(max_workers=min(_DOWNLOAD_WORKERS, len(components)))
     try:
-        with ThreadPoolExecutor(
-            max_workers=min(_DOWNLOAD_WORKERS, len(components))
-        ) as ex:
-            list(ex.map(_fetch, components))
+        for future in [ex.submit(_fetch, component) for component in components]:
+            future.result()
     finally:
+        # On Ctrl-C drop the queued archives instead of downloading them
+        # all before the process can exit; in-flight ones still finish.
+        ex.shutdown(wait=True, cancel_futures=True)
         progress.done()
 
 
@@ -1157,6 +1182,13 @@ def convert_libraries(
                 component.data.get("dependencies"), component.name
             ):
                 if "version" not in dependency:
+                    # Cannot resolve from the registry; the arduino-backend
+                    # PR adds the reconciliation that reports real drops
+                    _LOGGER.debug(
+                        "Skip version-less dependency %r of %s",
+                        dependency.get("name"),
+                        component.name,
+                    )
                     continue
                 if not dependency_is_usable(
                     dependency, backend.platform, backend.framework, component.name

tests/unit_tests/test_platformio_library.py

@@ -597,7 +597,7 @@ def test_lex_build_flags_dangling_flag_does_not_cross_entries(
 
 
 def test_prefetch_wave_downloads_registry_archives_in_parallel(
-    monkeypatch: pytest.MonkeyPatch,
+    setup_core, monkeypatch: pytest.MonkeyPatch
 ) -> None:
     """Registry archives in one wave download concurrently, deduped by URL;
     git/local sources and failures are left to the sequential call."""
@@ -636,12 +636,18 @@ def test_content_lengths_head_requests(monkeypatch: pytest.MonkeyPatch) -> None:
     def fake_head(url, timeout, allow_redirects):
         if "bad" in url:
             raise requests.ConnectionError("down")
-        return SimpleNamespace(headers={"content-length": "123"})
+        if "gone" in url:
+            return SimpleNamespace(ok=False, status_code=404, headers={})
+        return SimpleNamespace(ok=True, headers={"content-length": "123"})
 
     monkeypatch.setattr(
         lib.requests if hasattr(lib, "requests") else requests, "head", fake_head
     )
-    assert lib._content_lengths(["https://x/a", "https://x/bad"]) == [123, 0]
+    assert lib._content_lengths(["https://x/a", "https://x/bad", "https://x/gone"]) == [
+        123,
+        0,
+        0,
+    ]
 
 
 def test_prefetch_wave_cache_probe_failure_still_prefetches(
@@ -732,6 +738,11 @@ def test_normalize_dependencies_forms(caplog) -> None:
         {"": "1.0", "Wire": {"name": 123, "version": "1.0"}, "SPI": "*"}, "libx"
     ) == [{"name": "SPI", "owner": None, "version": "*"}]
     assert caplog.text.count("unrecognized dependency entry") == 5
+    # A container or numeric version would raise from set.add() or fail
+    # opaquely in the registry; both spellings warn and drop
+    assert normalize_dependencies({"Foo": ["1.0", "2.0"]}, "libx") == []
+    assert normalize_dependencies([{"name": "Foo", "version": 1}], "libx") == []
+    assert caplog.text.count("unrecognized dependency entry") == 7
 
 
 @pytest.mark.parametrize(