ci: skip agents with broken upstream URLs instead of failing the whole update run

.github/workflows/build_registry.py

@@ -7,17 +7,14 @@
 import os
 import re
 import sys
-import urllib.error
-import urllib.request
 import xml.etree.ElementTree as ET
 from pathlib import Path
 
 from registry_utils import (
-    extract_npm_package_name,
     extract_npm_package_version,
-    extract_pypi_package_name,
     normalize_version,
     should_skip_dir,
+    validate_distribution_urls,
 )
 
 try:
@@ -48,40 +45,6 @@
 PREFERRED_ICON_SIZE = 16
 ALLOWED_FILL_STROKE_VALUES = {"currentcolor", "none", "inherit"}
 
-# URL validation
-SKIP_URL_VALIDATION = os.environ.get("SKIP_URL_VALIDATION", "").lower() in (
-    "1",
-    "true",
-    "yes",
-)
-
-
-def url_exists(url: str, method: str = "HEAD", retries: int = 3) -> bool:
-    """Check if a URL exists using HEAD or GET request with retries."""
-    import time
-
-    for attempt in range(retries):
-        try:
-            req = urllib.request.Request(url, method=method)
-            req.add_header("User-Agent", "ACP-Registry-Validator/1.0")
-            with urllib.request.urlopen(req, timeout=15) as response:
-                return response.status in (200, 301, 302)
-        except urllib.error.HTTPError as e:
-            # Some servers don't support HEAD, try GET
-            if method == "HEAD" and e.code in (403, 405):
-                return url_exists(url, method="GET", retries=retries - attempt)
-            if attempt < retries - 1 and e.code in (429, 500, 502, 503, 504):
-                time.sleep(2**attempt)
-                continue
-            return False
-        except (urllib.error.URLError, TimeoutError, OSError):
-            if attempt < retries - 1:
-                time.sleep(2**attempt)
-                continue
-            return False
-    return False
-
-
 def extract_version_from_url(url: str) -> str | None:
     """Extract version from binary archive URL."""
     # GitHub releases: /download/v1.0.0/ or /releases/v1.0.0/
@@ -147,44 +110,6 @@ def validate_distribution_versions(agent_version: str, distribution: dict) -> li
     return errors
 
 
-def validate_distribution_urls(distribution: dict) -> list[str]:
-    """Validate that distribution URLs exist."""
-    if SKIP_URL_VALIDATION:
-        return []
-
-    errors = []
-
-    # Check binary archive URLs
-    if "binary" in distribution:
-        for platform, target in distribution["binary"].items():
-            if "archive" in target:
-                url = target["archive"]
-                if not url_exists(url):
-                    errors.append(f"Binary archive URL not accessible for {platform}: {url}")
-
-    # Check npm package URLs (registry.npmjs.org)
-    seen_npm = set()
-    for dist_type in ("npx",):
-        if dist_type in distribution:
-            package = distribution[dist_type].get("package", "")
-            pkg_name = extract_npm_package_name(package)
-            if pkg_name and pkg_name not in seen_npm:
-                seen_npm.add(pkg_name)
-                npm_url = f"https://registry.npmjs.org/{pkg_name}"
-                if not url_exists(npm_url):
-                    errors.append(f"npm package not found: {pkg_name}")
-
-    # Check PyPI package URLs
-    if "uvx" in distribution:
-        package = distribution["uvx"].get("package", "")
-        pkg_name = extract_pypi_package_name(package)
-        pypi_url = f"https://pypi.org/pypi/{pkg_name}/json"
-        if not url_exists(pypi_url):
-            errors.append(f"PyPI package not found: {pkg_name}")
-
-    return errors
-
-
 def validate_icon_monochrome(root: ET.Element) -> list[str]:
     """Validate that icon uses currentColor and no hardcoded colors.
 

.github/workflows/registry_utils.py

@@ -1,8 +1,12 @@
 """Shared utilities for ACP registry scripts."""
 
 import json
+import os
 import re
 import sys
+import time
+import urllib.error
+import urllib.request
 from pathlib import Path
 
 SKIP_DIRS = {
@@ -57,6 +61,80 @@ def normalize_version(version: str) -> str:
     return ".".join(parts[:3])
 
 
+# URL validation: env var lets CI / local runs disable network calls.
+SKIP_URL_VALIDATION = os.environ.get("SKIP_URL_VALIDATION", "").lower() in (
+    "1",
+    "true",
+    "yes",
+)
+
+
+def url_exists(url: str, method: str = "HEAD", retries: int = 3) -> bool:
+    """Check if a URL exists using HEAD or GET request with retries."""
+    for attempt in range(retries):
+        try:
+            req = urllib.request.Request(url, method=method)
+            req.add_header("User-Agent", "ACP-Registry-Validator/1.0")
+            with urllib.request.urlopen(req, timeout=15) as response:
+                return response.status in (200, 301, 302)
+        except urllib.error.HTTPError as e:
+            # Some servers don't support HEAD, try GET
+            if method == "HEAD" and e.code in (403, 405):
+                return url_exists(url, method="GET", retries=retries - attempt)
+            if attempt < retries - 1 and e.code in (429, 500, 502, 503, 504):
+                time.sleep(2**attempt)
+                continue
+            return False
+        except (urllib.error.URLError, TimeoutError, OSError):
+            if attempt < retries - 1:
+                time.sleep(2**attempt)
+                continue
+            return False
+    return False
+
+
+def validate_distribution_urls(distribution: dict) -> list[str]:
+    """Validate that distribution URLs exist (binary archives, npm, PyPI).
+
+    Returns a list of human-readable error strings (empty on success).
+    Honors SKIP_URL_VALIDATION env var.
+    """
+    if SKIP_URL_VALIDATION:
+        return []
+
+    errors = []
+
+    # Check binary archive URLs
+    if "binary" in distribution:
+        for platform, target in distribution["binary"].items():
+            if "archive" in target:
+                url = target["archive"]
+                if not url_exists(url):
+                    errors.append(f"Binary archive URL not accessible for {platform}: {url}")
+
+    # Check npm package URLs (registry.npmjs.org)
+    seen_npm = set()
+    for dist_type in ("npx",):
+        if dist_type in distribution:
+            package = distribution[dist_type].get("package", "")
+            pkg_name = extract_npm_package_name(package)
+            if pkg_name and pkg_name not in seen_npm:
+                seen_npm.add(pkg_name)
+                npm_url = f"https://registry.npmjs.org/{pkg_name}"
+                if not url_exists(npm_url):
+                    errors.append(f"npm package not found: {pkg_name}")
+
+    # Check PyPI package URLs
+    if "uvx" in distribution:
+        package = distribution["uvx"].get("package", "")
+        pkg_name = extract_pypi_package_name(package)
+        pypi_url = f"https://pypi.org/pypi/{pkg_name}/json"
+        if not url_exists(pypi_url):
+            errors.append(f"PyPI package not found: {pkg_name}")
+
+    return errors
+
+
 def load_quarantine(registry_dir: Path) -> dict[str, str]:
     """Load quarantine list from registry directory.
 

.github/workflows/tests/test_registry_utils.py

@@ -3,6 +3,7 @@
 import json
 import tempfile
 from pathlib import Path
+from unittest.mock import patch
 
 from registry_utils import (
     extract_npm_package_name,
@@ -11,6 +12,7 @@
     load_quarantine,
     normalize_version,
     should_skip_dir,
+    validate_distribution_urls,
 )
 
 
@@ -98,6 +100,72 @@ def test_invalid_json(self):
             assert load_quarantine(Path(d)) == {}
 
 
+class TestValidateDistributionUrls:
+    """Distribution URL validation (moved from build_registry.py)."""
+
+    def test_returns_no_errors_when_all_urls_reachable(self):
+        distribution = {
+            "binary": {
+                "darwin-aarch64": {
+                    "archive": "https://example.com/agent-darwin.tar.gz",
+                    "cmd": "./agent",
+                },
+                "linux-x86_64": {
+                    "archive": "https://example.com/agent-linux.tar.gz",
+                    "cmd": "./agent",
+                },
+            }
+        }
+        with patch("registry_utils.url_exists", return_value=True):
+            assert validate_distribution_urls(distribution) == []
+
+    def test_reports_unreachable_binary_archive(self):
+        distribution = {
+            "binary": {
+                "darwin-aarch64": {
+                    "archive": "https://example.com/ok.tar.gz",
+                    "cmd": "./agent",
+                },
+                "windows-x86_64": {
+                    "archive": "https://example.com/missing-windows.zip",
+                    "cmd": "agent.exe",
+                },
+            }
+        }
+
+        def fake_url_exists(url, *_args, **_kwargs):
+            return "missing" not in url
+
+        with patch("registry_utils.url_exists", side_effect=fake_url_exists):
+            errors = validate_distribution_urls(distribution)
+
+        assert len(errors) == 1
+        assert "windows-x86_64" in errors[0]
+        assert "missing-windows.zip" in errors[0]
+
+    def test_skip_url_validation_env_returns_empty(self, monkeypatch):
+        monkeypatch.setenv("SKIP_URL_VALIDATION", "1")
+        # Re-import to pick up the patched env var.
+        import importlib
+
+        import registry_utils
+
+        importlib.reload(registry_utils)
+        try:
+            distribution = {
+                "binary": {
+                    "darwin-aarch64": {
+                        "archive": "https://example.com/anything.tar.gz",
+                        "cmd": "./agent",
+                    }
+                }
+            }
+            assert registry_utils.validate_distribution_urls(distribution) == []
+        finally:
+            monkeypatch.delenv("SKIP_URL_VALIDATION", raising=False)
+            importlib.reload(registry_utils)
+
+
 class TestShouldSkipDir:
     def test_skips_hidden_runtime_dirs(self):
         assert should_skip_dir(".sandbox")