Merge commit 'dfba4cc5ca' into studio-tokenless-llama-prebuilt

# Conflicts:
#	tests/studio/install/test_macos_version_compat.py

studio/install_llama_prebuilt.py

@@ -1208,14 +1208,102 @@ def github_releases(
     return releases
 
 
+_RELEASE_TAG_PATH_RE = re.compile(r"/releases/tag/(?P<tag>[^/?#]+)")
+
+
+def web_request_headers() -> dict[str, str]:
+    # github.com release endpoints are not rate-limited like api.github.com, so
+    # this path is intentionally unauthenticated: no token, no header logging.
+    return {"User-Agent": "unsloth-studio-llama-prebuilt"}
+
+
+def upstream_release_download_url(repo: str, tag: str, asset_name: str) -> str:
+    """Deterministic release-asset download URL that needs no REST API call.
+
+    github.com/<repo>/releases/download/<tag>/<asset> redirects to the release
+    CDN and is not part of the rate-limited api.github.com surface.
+    """
+    return (
+        f"https://github.com/{repo}/releases/download/"
+        f"{urllib.parse.quote(tag, safe = '')}/{asset_name}"
+    )
+
+
+def _tag_from_release_location(location: str | None) -> str | None:
+    if not location:
+        return None
+    try:
+        path = urllib.parse.urlparse(location).path
+    except Exception:
+        return None
+    match = _RELEASE_TAG_PATH_RE.search(path)
+    if not match:
+        return None
+    tag = urllib.parse.unquote(match.group("tag")).strip()
+    return tag or None
+
+
+def _resolve_latest_release_tag_via_redirect(repo: str) -> str | None:
+    """Resolve a repo's latest release tag without the rate-limited REST API.
+
+    github.com/<repo>/releases/latest 302-redirects to .../releases/tag/<TAG>;
+    reading that Location costs none of the anonymous api.github.com budget.
+    Returns None on failure so callers fall back to the REST listing.
+    """
+
+    class _NoFollowRedirect(urllib.request.HTTPRedirectHandler):
+        def redirect_request(self, req, fp, code, msg, headers, newurl):
+            # Never follow the redirect; read the tag from the 3xx Location.
+            return None
+
+    url = f"https://github.com/{repo}/releases/latest"
+    opener = urllib.request.build_opener(_NoFollowRedirect)
+    last_exc: Exception | None = None
+    for attempt in range(1, JSON_FETCH_ATTEMPTS + 1):
+        try:
+            request = urllib.request.Request(url, headers = web_request_headers())
+            try:
+                with opener.open(request, timeout = 30) as response:
+                    location = response.headers.get("Location")
+            except urllib.error.HTTPError as exc:
+                # Disabling redirects surfaces the 3xx as an HTTPError whose
+                # Location header still carries the resolved release-tag URL.
+                if exc.code in {301, 302, 303, 307, 308}:
+                    location = exc.headers.get("Location") if exc.headers else None
+                else:
+                    raise
+            tag = _tag_from_release_location(location)
+            # Accept only upstream-style tags (b1234); any other redirect target
+            # is unexpected, so fail closed to the REST/source path.
+            return tag if is_release_tag_like(tag) else None
+        except Exception as exc:
+            last_exc = exc
+            if attempt >= JSON_FETCH_ATTEMPTS or not is_retryable_url_error(exc):
+                break
+            sleep_backoff(attempt, exc = exc)
+    if last_exc is not None:
+        log(f"latest-release redirect resolution failed for {repo}: {last_exc}")
+    return None
+
+
 def latest_upstream_release_tag() -> str:
-    payload = fetch_json(UPSTREAM_RELEASES_API)
-    tag = payload.get("tag_name")
-    if not isinstance(tag, str) or not tag:
-        raise RuntimeError(
-            f"latest release tag was missing from {UPSTREAM_RELEASES_API}"
-        )
-    return tag
+    # Primary path is the GitHub REST API; fall back to the github.com release
+    # redirect (no token, not subject to the anonymous api.github.com 403 rate
+    # limit) when REST is unavailable or returns no usable tag.
+    try:
+        payload = fetch_json(UPSTREAM_RELEASES_API)
+        tag = payload.get("tag_name")
+    except (urllib.error.URLError, RuntimeError):
+        redirect_tag = _resolve_latest_release_tag_via_redirect(UPSTREAM_REPO)
+        if redirect_tag:
+            return redirect_tag
+        raise
+    if isinstance(tag, str) and tag:
+        return tag
+    redirect_tag = _resolve_latest_release_tag_via_redirect(UPSTREAM_REPO)
+    if redirect_tag:
+        return redirect_tag
+    raise RuntimeError(f"latest release tag was missing from {UPSTREAM_RELEASES_API}")
 
 
 def is_release_tag_like(value: str | None) -> bool:
@@ -1240,10 +1328,22 @@ def release_time_sort_key(release: dict[str, Any]) -> tuple[str, int]:
     return (timestamp, normalized_id)
 
 
+def _needs_dynamic_asset_enumeration(host: HostInfo | None) -> bool:
+    # Windows x64 NVIDIA/AMD hosts pick among several CUDA/HIP archives by
+    # probing a release's actual assets, which the redirect fast path does not
+    # enumerate. Keep them on the REST listing to avoid a silent CPU downgrade.
+    if host is None:
+        return False
+    return bool(
+        host.is_windows and host.is_x86_64 and (host.has_usable_nvidia or host.has_rocm)
+    )
+
+
 def iter_release_payloads_by_time(
     repo: str,
     published_release_tag: str = "",
     requested_tag: str = "",
+    host: HostInfo | None = None,
 ) -> Iterable[dict[str, Any]]:
     if published_release_tag:
         yield github_release(repo, published_release_tag)
@@ -1264,18 +1364,46 @@ def iter_release_payloads_by_time(
                 )
             else:
                 raise
-        except Exception:
-            raise
 
-    releases = [
-        release
-        for release in github_releases(
-            repo, max_pages = DEFAULT_GITHUB_RELEASE_SCAN_MAX_PAGES
-        )
-        if isinstance(release, dict)
-        and not release.get("draft")
-        and not release.get("prerelease")
-    ]
+    # Primary: the GitHub REST release listing. Fallback for upstream "latest":
+    # resolve the tag from the github.com release redirect (no api.github.com
+    # budget spent) and synthesize a release whose sentinel lets the planner
+    # build deterministic download URLs, so a rate-limited (HTTP 403) REST API
+    # does not force a source build. Gated to upstream single-asset platforms.
+    redirect_eligible = (
+        repo == UPSTREAM_REPO
+        and (not requested_tag or requested_tag == "latest")
+        and not _needs_dynamic_asset_enumeration(host)
+    )
+    try:
+        releases = [
+            release
+            for release in github_releases(
+                repo, max_pages = DEFAULT_GITHUB_RELEASE_SCAN_MAX_PAGES
+            )
+            if isinstance(release, dict)
+            and not release.get("draft")
+            and not release.get("prerelease")
+        ]
+    except (urllib.error.URLError, RuntimeError) as exc:
+        if redirect_eligible:
+            redirect_tag = _resolve_latest_release_tag_via_redirect(repo)
+            if redirect_tag:
+                log(
+                    f"GitHub REST release listing failed ({exc}); resolved "
+                    f"latest {repo} release {redirect_tag} via release redirect"
+                )
+                yield {
+                    "tag_name": redirect_tag,
+                    "assets": [],
+                    "_unsloth_download_repo": repo,
+                }
+                return
+            # Both REST and the redirect fallback failed; surface both causes.
+            raise RuntimeError(
+                f"{exc}; release redirect fallback for {repo} also failed"
+            ) from exc
+        raise
     releases.sort(key = release_time_sort_key, reverse = True)
     for release in releases:
         yield release
@@ -1477,6 +1605,16 @@ def direct_upstream_release_plan(
         return None
 
     assets = release_asset_map(release)
+    # Redirect fast-path releases carry no enumerated assets; build URLs from
+    # the tag instead. Without this sentinel, behavior is unchanged.
+    download_repo = release.get("_unsloth_download_repo")
+
+    def asset_url_for(asset_name: str) -> str | None:
+        url = assets.get(asset_name)
+        if not url and download_repo:
+            return upstream_release_download_url(download_repo, release_tag, asset_name)
+        return url
+
     attempts: list[AssetChoice] = []
     if host.is_windows and host.is_x86_64:
         if host.has_usable_nvidia:
@@ -1502,7 +1640,7 @@ def direct_upstream_release_plan(
             if lemonade_choice is not None:
                 attempts.append(lemonade_choice)
             hip_asset = f"llama-{release_tag}-bin-win-hip-radeon-x64.zip"
-            hip_url = assets.get(hip_asset)
+            hip_url = asset_url_for(hip_asset)
             if hip_url:
                 attempts.append(
                     AssetChoice(
@@ -1515,7 +1653,7 @@ def direct_upstream_release_plan(
                     )
                 )
         cpu_asset = f"llama-{release_tag}-bin-win-cpu-x64.zip"
-        cpu_url = assets.get(cpu_asset)
+        cpu_url = asset_url_for(cpu_asset)
         if cpu_url:
             attempts.append(
                 AssetChoice(
@@ -1533,7 +1671,7 @@ def direct_upstream_release_plan(
         # selector returned 0 attempts and the installer fell back to a
         # source build on every Windows ARM64 host.
         cpu_asset = f"llama-{release_tag}-bin-win-cpu-arm64.zip"
-        cpu_url = assets.get(cpu_asset)
+        cpu_url = asset_url_for(cpu_asset)
         if cpu_url:
             attempts.append(
                 AssetChoice(
@@ -1547,7 +1685,7 @@ def direct_upstream_release_plan(
             )
     elif host.is_macos and host.is_arm64:
         asset_name = f"llama-{release_tag}-bin-macos-arm64.tar.gz"
-        asset_url = assets.get(asset_name)
+        asset_url = asset_url_for(asset_name)
         if asset_url:
             attempts.append(
                 AssetChoice(
@@ -1561,7 +1699,7 @@ def direct_upstream_release_plan(
             )
     elif host.is_macos and host.is_x86_64:
         asset_name = f"llama-{release_tag}-bin-macos-x64.tar.gz"
-        asset_url = assets.get(asset_name)
+        asset_url = asset_url_for(asset_name)
         if asset_url:
             attempts.append(
                 AssetChoice(
@@ -1575,7 +1713,7 @@ def direct_upstream_release_plan(
             )
     elif host.is_linux and host.is_x86_64 and not host.has_usable_nvidia:
         asset_name = f"llama-{release_tag}-bin-ubuntu-x64.tar.gz"
-        asset_url = assets.get(asset_name)
+        asset_url = asset_url_for(asset_name)
         if asset_url:
             attempts.append(
                 AssetChoice(
@@ -1594,7 +1732,7 @@ def direct_upstream_release_plan(
         # source build on every Linux ARM64 host (DGX Spark, Ampere
         # Altra, GitHub-hosted ubuntu-24.04-arm runners, etc.).
         asset_name = f"llama-{release_tag}-bin-ubuntu-arm64.tar.gz"
-        asset_url = assets.get(asset_name)
+        asset_url = asset_url_for(asset_name)
         if asset_url:
             attempts.append(
                 AssetChoice(
@@ -1666,7 +1804,7 @@ def resolve_simple_install_release_plans(
 
     try:
         releases = iter_release_payloads_by_time(
-            repo, published_release_tag, requested_tag
+            repo, published_release_tag, requested_tag, host
         )
         for release in releases:
             try:

tests/studio/install/test_macos_version_compat.py

@@ -247,7 +247,7 @@ class TestMacosReleasePin:
     TAGS = [f"b{n}" for n in range(9442, 9400, -1)]  # newest-first, includes b9415
 
     def _patch_releases(self, monkeypatch):
-        def fake_iter(repo, published_release_tag, requested_tag):
+        def fake_iter(repo, published_release_tag, requested_tag, host = None):
             # The real iterator yields only the requested tag when one is pinned.
             if requested_tag and requested_tag != "latest":
                 return _fake_macos_releases([requested_tag])

tests/studio/install/test_selection_logic.py

@@ -2717,7 +2717,7 @@ class TestResolveSimpleMacosPin:
                 ],
             }
 
-        def fake_iter(repo, published_release_tag = "", requested_tag = ""):
+        def fake_iter(repo, published_release_tag = "", requested_tag = "", host = None):
             calls.append((repo, published_release_tag, requested_tag))
             # Emulate the real iterator: a specific tag yields only that release.
             if requested_tag and requested_tag != "latest":

tests/studio/install/test_tokenless_release_resolution.py

@@ -0,0 +1,430 @@
+"""Tests for tokenless (no GH_TOKEN) llama.cpp release resolution.
+
+Cover the github.com release-redirect fallback that resolves the latest upstream
+release when the rate-limited api.github.com REST surface is unavailable.
+"""
+
+import sys
+import urllib.error
+from email.message import Message
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[3]
+STUDIO_DIR = ROOT / "studio"
+if str(STUDIO_DIR) not in sys.path:
+    sys.path.insert(0, str(STUDIO_DIR))
+
+import install_llama_prebuilt as INSTALL_LLAMA_PREBUILT  # noqa: E402
+from install_llama_prebuilt import HostInfo  # noqa: E402
+
+MOD = INSTALL_LLAMA_PREBUILT
+
+
+def _host(**overrides) -> HostInfo:
+    base = dict(
+        system = "Darwin",
+        machine = "arm64",
+        is_windows = False,
+        is_linux = False,
+        is_macos = True,
+        is_x86_64 = False,
+        is_arm64 = True,
+        nvidia_smi = None,
+        driver_cuda_version = None,
+        compute_caps = [],
+        visible_cuda_devices = None,
+        has_physical_nvidia = False,
+        has_usable_nvidia = False,
+        has_rocm = False,
+        rocm_gfx_target = None,
+    )
+    base.update(overrides)
+    return HostInfo(**base)
+
+
+def _linux_cpu(**overrides) -> HostInfo:
+    return _host(
+        system = "Linux",
+        machine = "x86_64",
+        is_windows = False,
+        is_linux = True,
+        is_macos = False,
+        is_x86_64 = True,
+        is_arm64 = False,
+        **overrides,
+    )
+
+
+def _win_cuda(**overrides) -> HostInfo:
+    return _host(
+        system = "Windows",
+        machine = "AMD64",
+        is_windows = True,
+        is_linux = False,
+        is_macos = False,
+        is_x86_64 = True,
+        is_arm64 = False,
+        has_physical_nvidia = True,
+        has_usable_nvidia = True,
+        **overrides,
+    )
+
+
+def _headers(**pairs) -> Message:
+    message = Message()
+    for key, value in pairs.items():
+        message[key] = value
+    return message
+
+
+class _FakeRedirectResponse:
+    def __init__(self, headers: Message):
+        self._headers = headers
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *exc):
+        return False
+
+    @property
+    def headers(self) -> Message:
+        return self._headers
+
+
+def _rest_403(*args, **kwargs):
+    # Matches how fetch_json / github_releases surface an anonymous rate limit.
+    raise RuntimeError(
+        "GitHub API returned 403 for "
+        "https://api.github.com/repos/ggml-org/llama.cpp/releases?per_page=100&page=1"
+    )
+
+
+# --- pure helpers -----------------------------------------------------------
+
+
+def test_upstream_release_download_url_is_deterministic():
+    url = MOD.upstream_release_download_url(
+        "ggml-org/llama.cpp", "b9999", "llama-b9999-bin-macos-arm64.tar.gz"
+    )
+    assert url == (
+        "https://github.com/ggml-org/llama.cpp/releases/download/"
+        "b9999/llama-b9999-bin-macos-arm64.tar.gz"
+    )
+
+
+def test_tag_from_release_location_parses_tag():
+    assert (
+        MOD._tag_from_release_location(
+            "https://github.com/ggml-org/llama.cpp/releases/tag/b9437"
+        )
+        == "b9437"
+    )
+    # URL-encoded tag is decoded.
+    assert (
+        MOD._tag_from_release_location(
+            "https://github.com/owner/repo/releases/tag/v1.2.3%2Bcuda"
+        )
+        == "v1.2.3+cuda"
+    )
+
+
+@pytest.mark.parametrize(
+    "location",
+    [None, "", "https://github.com/owner/repo/releases", "not a url at all"],
+)
+def test_tag_from_release_location_rejects_non_tag_urls(location):
+    assert MOD._tag_from_release_location(location) is None
+
+
+# --- redirect resolver ------------------------------------------------------
+
+
+def test_resolve_latest_tag_via_redirect_reads_location(monkeypatch):
+    captured = {}
+
+    def fake_build_opener(*handlers):
+        class _Opener:
+            def open(self, request, timeout = None):
+                captured["url"] = request.full_url
+                captured["has_auth"] = "Authorization" in request.headers
+                return _FakeRedirectResponse(
+                    _headers(
+                        Location = "https://github.com/ggml-org/llama.cpp/releases/tag/b9999"
+                    )
+                )
+
+        return _Opener()
+
+    monkeypatch.setattr(MOD.urllib.request, "build_opener", fake_build_opener)
+    tag = MOD._resolve_latest_release_tag_via_redirect("ggml-org/llama.cpp")
+    assert tag == "b9999"
+    assert captured["url"] == "https://github.com/ggml-org/llama.cpp/releases/latest"
+    # The redirect call must never be authenticated (no token leakage).
+    assert captured["has_auth"] is False
+
+
+def test_resolve_latest_tag_via_redirect_handles_httperror_3xx(monkeypatch):
+    def fake_build_opener(*handlers):
+        class _Opener:
+            def open(self, request, timeout = None):
+                raise urllib.error.HTTPError(
+                    request.full_url,
+                    302,
+                    "Found",
+                    _headers(
+                        Location = "https://github.com/ggml-org/llama.cpp/releases/tag/b1234"
+                    ),
+                    None,
+                )
+
+        return _Opener()
+
+    monkeypatch.setattr(MOD.urllib.request, "build_opener", fake_build_opener)
+    assert MOD._resolve_latest_release_tag_via_redirect("ggml-org/llama.cpp") == "b1234"
+
+
+def test_resolve_latest_tag_via_redirect_returns_none_on_failure(monkeypatch):
+    def fake_build_opener(*handlers):
+        class _Opener:
+            def open(self, request, timeout = None):
+                raise urllib.error.URLError("connection reset")
+
+        return _Opener()
+
+    monkeypatch.setattr(MOD.urllib.request, "build_opener", fake_build_opener)
+    monkeypatch.setattr(MOD, "sleep_backoff", lambda *a, **k: None)
+    assert MOD._resolve_latest_release_tag_via_redirect("ggml-org/llama.cpp") is None
+
+
+# --- latest_upstream_release_tag: REST first, redirect fallback -------------
+
+
+def test_latest_upstream_release_tag_uses_rest_first(monkeypatch):
+    monkeypatch.setattr(MOD, "fetch_json", lambda url: {"tag_name": "b500"})
+
+    def boom(repo):
+        raise AssertionError("redirect must not be used when REST succeeds")
+
+    monkeypatch.setattr(MOD, "_resolve_latest_release_tag_via_redirect", boom)
+    assert MOD.latest_upstream_release_tag() == "b500"
+
+
+def test_latest_upstream_release_tag_falls_back_to_redirect_on_403(monkeypatch):
+    monkeypatch.setattr(MOD, "fetch_json", _rest_403)
+    monkeypatch.setattr(
+        MOD, "_resolve_latest_release_tag_via_redirect", lambda repo: "b600"
+    )
+    assert MOD.latest_upstream_release_tag() == "b600"
+
+
+def test_latest_upstream_release_tag_falls_back_to_redirect_on_missing_tag(monkeypatch):
+    monkeypatch.setattr(MOD, "fetch_json", lambda url: {})
+    monkeypatch.setattr(
+        MOD, "_resolve_latest_release_tag_via_redirect", lambda repo: "b601"
+    )
+    assert MOD.latest_upstream_release_tag() == "b601"
+
+
+# --- iter_release_payloads_by_time: REST first, redirect fallback -----------
+
+
+def test_iter_release_payloads_uses_rest_listing_first(monkeypatch):
+    rest_release = {
+        "tag_name": "b800",
+        "assets": [],
+        "published_at": "2026-01-01T00:00:00Z",
+        "id": 800,
+    }
+    monkeypatch.setattr(MOD, "github_releases", lambda *a, **k: [rest_release])
+
+    def boom(repo):
+        raise AssertionError("redirect must not be used when REST succeeds")
+
+    monkeypatch.setattr(MOD, "_resolve_latest_release_tag_via_redirect", boom)
+
+    releases = list(
+        MOD.iter_release_payloads_by_time(MOD.UPSTREAM_REPO, "", "latest", _host())
+    )
+    assert releases == [rest_release]
+
+
+def test_iter_release_payloads_falls_back_to_redirect_on_403(monkeypatch):
+    monkeypatch.setattr(MOD, "github_releases", _rest_403)
+    monkeypatch.setattr(
+        MOD, "_resolve_latest_release_tag_via_redirect", lambda repo: "b900"
+    )
+
+    releases = list(
+        MOD.iter_release_payloads_by_time(MOD.UPSTREAM_REPO, "", "latest", _host())
+    )
+    assert len(releases) == 1
+    assert releases[0]["tag_name"] == "b900"
+    assert releases[0]["assets"] == []
+    assert releases[0]["_unsloth_download_repo"] == MOD.UPSTREAM_REPO
+
+
+def test_windows_cuda_host_does_not_use_redirect_fallback(monkeypatch):
+    win = _win_cuda()
+    assert MOD._needs_dynamic_asset_enumeration(win) is True
+    assert MOD._needs_dynamic_asset_enumeration(_host()) is False
+
+    monkeypatch.setattr(MOD, "github_releases", _rest_403)
+
+    def boom(repo):
+        raise AssertionError("Windows CUDA hosts must not use the redirect fallback")
+
+    monkeypatch.setattr(MOD, "_resolve_latest_release_tag_via_redirect", boom)
+
+    with pytest.raises(RuntimeError):
+        list(MOD.iter_release_payloads_by_time(MOD.UPSTREAM_REPO, "", "latest", win))
+
+
+# --- end to end: REST 403 falls back to a redirect-resolved prebuilt --------
+
+
+def test_macos_latest_prebuilt_via_redirect_when_rest_403(monkeypatch):
+    monkeypatch.setattr(MOD, "github_releases", _rest_403)
+    monkeypatch.setattr(
+        MOD, "_resolve_latest_release_tag_via_redirect", lambda repo: "b9999"
+    )
+
+    requested_tag, plans = MOD.resolve_simple_install_release_plans(
+        "latest", _host(), "ggml-org/llama.cpp", ""
+    )
+    assert requested_tag == "latest"
+    assert len(plans) == 1
+    plan = plans[0]
+    assert plan.llama_tag == "b9999"
+    assert len(plan.attempts) == 1
+    assert plan.attempts[0].url == (
+        "https://github.com/ggml-org/llama.cpp/releases/download/"
+        "b9999/llama-b9999-bin-macos-arm64.tar.gz"
+    )
+    assert plan.attempts[0].name == "llama-b9999-bin-macos-arm64.tar.gz"
+
+
+def test_linux_cpu_latest_prebuilt_via_redirect_when_rest_403(monkeypatch):
+    monkeypatch.setattr(MOD, "github_releases", _rest_403)
+    monkeypatch.setattr(
+        MOD, "_resolve_latest_release_tag_via_redirect", lambda repo: "b9999"
+    )
+
+    _requested, plans = MOD.resolve_simple_install_release_plans(
+        "latest", _linux_cpu(), "ggml-org/llama.cpp", ""
+    )
+    assert plans[0].attempts[0].url == (
+        "https://github.com/ggml-org/llama.cpp/releases/download/"
+        "b9999/llama-b9999-bin-ubuntu-x64.tar.gz"
+    )
+
+
+# --- negative paths: fail closed and asset precedence ----------------------
+
+
+def test_resolve_latest_tag_via_redirect_rejects_non_release_tag(monkeypatch):
+    class _Resp:
+        def __init__(self, location):
+            self.headers = {"Location": location}
+
+        def __enter__(self):
+            return self
+
+        def __exit__(self, *a):
+            return False
+
+    def fake_build_opener(handler):
+        class _Op:
+            def open(self, request, timeout = None):
+                return _Resp(
+                    "https://github.com/ggml-org/llama.cpp/releases/tag/nightly"
+                )
+
+        return _Op()
+
+    monkeypatch.setattr(MOD.urllib.request, "build_opener", fake_build_opener)
+    # A non b1234 redirect target must fail closed to the REST/source path.
+    assert MOD._resolve_latest_release_tag_via_redirect("ggml-org/llama.cpp") is None
+
+
+def test_iter_release_payloads_both_paths_fail_raises(monkeypatch):
+    def boom(repo, max_pages = 5):
+        raise RuntimeError("GitHub API returned 403")
+
+    monkeypatch.setattr(MOD, "github_releases", boom)
+    monkeypatch.setattr(
+        MOD, "_resolve_latest_release_tag_via_redirect", lambda repo: None
+    )
+    with pytest.raises(RuntimeError) as excinfo:
+        list(MOD.iter_release_payloads_by_time(MOD.UPSTREAM_REPO, "", "latest"))
+    message = str(excinfo.value)
+    assert "release redirect fallback" in message
+    assert "also failed" in message
+
+
+def test_iter_release_payloads_pinned_tag_never_uses_redirect(monkeypatch):
+    def missing(repo, tag):
+        raise urllib.error.HTTPError(
+            f"https://api.github.com/repos/{repo}/releases/tags/{tag}",
+            404,
+            "Not Found",
+            None,
+            None,
+        )
+
+    def listing_403(repo, max_pages = 5):
+        raise RuntimeError("GitHub API returned 403")
+
+    monkeypatch.setattr(MOD, "github_release", missing)
+    monkeypatch.setattr(MOD, "github_releases", listing_403)
+    monkeypatch.setattr(
+        MOD,
+        "_resolve_latest_release_tag_via_redirect",
+        lambda repo: (_ for _ in ()).throw(AssertionError("redirect used")),
+    )
+    # A pinned tag is not "latest", so the redirect fast path must never run.
+    with pytest.raises(RuntimeError):
+        list(MOD.iter_release_payloads_by_time(MOD.UPSTREAM_REPO, "", "b1234"))
+
+
+def test_direct_release_real_asset_takes_precedence_over_redirect(monkeypatch):
+    host = _host(
+        system = "Darwin",
+        machine = "arm64",
+        is_windows = False,
+        is_linux = False,
+        is_macos = True,
+        is_x86_64 = False,
+        is_arm64 = True,
+    )
+    custom_url = "https://cdn.example.test/custom-macos-arm64.tar.gz"
+    release = {
+        "tag_name": "b1234",
+        "assets": [
+            {
+                "name": "llama-b1234-bin-macos-arm64.tar.gz",
+                "browser_download_url": custom_url,
+            }
+        ],
+    }
+    monkeypatch.setattr(MOD, "github_releases", lambda repo, max_pages = 5: [release])
+    monkeypatch.setattr(
+        MOD,
+        "fetch_json",
+        lambda *a, **k: (_ for _ in ()).throw(AssertionError("fetch_json used")),
+    )
+    monkeypatch.setattr(
+        MOD,
+        "_resolve_latest_release_tag_via_redirect",
+        lambda repo: (_ for _ in ()).throw(AssertionError("redirect used")),
+    )
+    _requested, plans = MOD.resolve_simple_install_release_plans(
+        "latest", host, "ggml-org/llama.cpp", ""
+    )
+    assert plans
+    attempt_urls = [a.url for p in plans for a in p.attempts]
+    # The real REST asset URL wins; the synthetic redirect URL is additive only.
+    assert custom_url in attempt_urls
+    assert all("releases/download/b1234" not in u for u in attempt_urls)