scripts/update_lemonade_pins.py
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env python3
# SPDX-License-Identifier: AGPL-3.0-only
# Copyright 2026-present the Unsloth AI Inc. team. All rights reserved. See /studio/LICENSE.AGPL-3.0
"""Move studio/lemonade_prebuilt_pins.json to the latest stable Lemonade and FastFlowLM releases.
A release whose Linux build needs a system library newer than LINUX_ABI_LIMITS is skipped for the
newest one that does not, since a lemond that cannot start breaks the NPU for every Linux user on that distro.
Exit codes: 0 = current (or updated with --write); 1 = --check found a newer release; 2 = error.
"""
from __future__ import annotations
import argparse
import hashlib
import json
import os
import re
import sys
import tarfile
import tempfile
import urllib.request
from pathlib import Path
from typing import Callable, Optional
PINS_PATH = Path(__file__).resolve().parents[1] / "studio" / "lemonade_prebuilt_pins.json"
_API = "https://api.github.com"
_TIMEOUT_S = 60
_CHUNK = 1 << 20
# Newest symbol versions Ubuntu 24.04 ships; FastFlowLM's own Linux build already needs glibc 2.39.
LINUX_ABI_LIMITS = {
"GLIBC": (2, 39),
"GLIBCXX": (3, 4, 33),
"CXXABI": (1, 3, 15),
"OPENSSL": (3, 0, 9),
}
_SYMBOL_VERSION = re.compile(rb"(GLIBCXX|GLIBC|CXXABI|OPENSSL)_(\d+(?:\.\d+)+)")
Needs = dict[str, tuple[int, ...]]
_NEXT_PAGE = re.compile(r'<([^>]+)>;\s*rel="next"')
def _lemonade_assets(version: str) -> dict[str, str]:
return {
"linux-x64": f"lemonade-embeddable-{version}-ubuntu-x64.tar.gz",
"windows-x64": f"lemonade-embeddable-{version}-windows-x64.zip",
}
def _fastflowlm_assets(version: str) -> dict[str, str]:
# The names lemond builds for its /v1/install download.
return {
"linux-x64": f"fastflowlm_{version}_linux.tar.gz",
"windows-x64": f"fastflowlm_{version}_windows_amd64.zip",
}
# section -> (asset names, pin keeps the tag's "v", pin records sizes)
_SECTIONS: dict[str, tuple[Callable[[str], dict[str, str]], bool, bool]] = {
"lemonade": (_lemonade_assets, False, True),
"fastflowlm": (_fastflowlm_assets, True, False),
}
def _request(url: str) -> urllib.request.Request:
headers = {"User-Agent": "unsloth-pins-updater", "Accept": "application/vnd.github+json"}
token = os.environ.get("GITHUB_TOKEN") or os.environ.get("GH_TOKEN")
if token and url.startswith(_API):
headers["Authorization"] = f"Bearer {token}"
return urllib.request.Request(url, headers = headers)
def fetch_stable_releases(repo: str) -> list[dict]:
# Every page: prerelease candidates can push a compatible stable release past the first one.
releases, url = [], f"{_API}/repos/{repo}/releases?per_page=100"
while url:
with urllib.request.urlopen(_request(url), timeout = _TIMEOUT_S) as r:
releases += json.load(r)
next_page = _NEXT_PAGE.search(r.headers.get("Link") or "")
url = next_page.group(1) if next_page else None
return [r for r in releases if not r.get("draft") and not r.get("prerelease")]
def tar_abi_needs(fileobj) -> Needs:
"""Newest symbol version per LINUX_ABI_LIMITS library that any ELF file in a .tar.gz references."""
newest: Needs = {}
with tarfile.open(fileobj = fileobj, mode = "r:gz") as archive:
for member in archive:
handle = archive.extractfile(member) if member.isfile() else None
if handle is None or handle.read(4) != b"\x7fELF":
continue
for match in _SYMBOL_VERSION.finditer(handle.read()):
lib = match.group(1).decode()
version = tuple(int(part) for part in match.group(2).split(b"."))
newest[lib] = max(newest.get(lib, ()), version)
return newest
def hash_asset(url: str) -> tuple[str, int, Optional[Needs]]:
"""sha256, size and, for a Linux tarball, the newest system library versions its binaries need."""
digest = hashlib.sha256()
size = 0
with tempfile.TemporaryFile() as tmp:
with urllib.request.urlopen(_request(url), timeout = _TIMEOUT_S) as response:
while chunk := response.read(_CHUNK):
digest.update(chunk)
tmp.write(chunk)
size += len(chunk)
tmp.seek(0)
needs = tar_abi_needs(tmp) if url.endswith(".tar.gz") else None
return digest.hexdigest(), size, needs
class AbiTooNew(RuntimeError):
"""A release whose Linux build cannot start within LINUX_ABI_LIMITS."""
def _dotted(version: tuple[int, ...]) -> str:
return ".".join(map(str, version))
def version_key(version: str) -> tuple[int, ...]:
"""11.9.0 < 2026.41.1 and v1.0.3 < v1.0.7; anything else is not a release version."""
parts = version.lstrip("v").split(".")
if not parts or not all(part.isdigit() for part in parts):
raise ValueError(f"{version!r} is not a numeric release version.")
return tuple(int(part) for part in parts)
def updated_section(
name: str,
section: dict,
release: dict,
hasher: Callable[[str], tuple[str, int, Optional[Needs]]],
) -> Optional[dict]:
tag = str(release["tag_name"])
bare = tag[1:] if tag.startswith("v") else tag
asset_names, keep_v, with_size = _SECTIONS[name]
pinned = section["version"]
if version_key(tag) <= version_key(pinned):
return None
published = {asset["name"]: asset for asset in release.get("assets") or []}
assets = {}
for key, asset_name in asset_names(bare).items():
asset = published.get(asset_name)
if asset is None:
raise RuntimeError(f"{section['repo']} {tag} has no {asset_name}.")
sha256, size, needs = hasher(asset["browser_download_url"])
github_digest = asset.get("digest")
# Fail closed: without GitHub's digest the hash proves only what this runner downloaded.
if not github_digest:
raise RuntimeError(f"{asset_name}: GitHub publishes no digest; refusing to pin it.")
if github_digest != f"sha256:{sha256}":
raise RuntimeError(
f"{asset_name} hashed to sha256:{sha256}, but GitHub publishes {github_digest}."
)
if asset.get("size") is not None and asset["size"] != size:
raise RuntimeError(f"{asset_name} is {size} bytes, but GitHub lists {asset['size']}.")
too_new = [
f"{lib} {_dotted(version)} > {_dotted(LINUX_ABI_LIMITS[lib])}"
for lib, version in sorted((needs or {}).items())
if version > LINUX_ABI_LIMITS[lib]
]
if too_new:
raise AbiTooNew(f"{section['repo']} {tag}: {asset_name} needs {', '.join(too_new)}.")
assets[key] = {"name": asset_name, "sha256": sha256}
if with_size:
assets[key]["size"] = size
return {**section, "version": tag if keep_v else bare, "assets": assets}
def newer_releases(releases: list[dict], pinned: str) -> list[dict]:
"""Releases with a numeric tag above the pin, newest first."""
newer = []
for release in releases:
try:
key = version_key(str(release["tag_name"]))
except ValueError:
continue
if key > version_key(pinned):
newer.append((key, release))
return [release for _, release in sorted(newer, key = lambda item: item[0], reverse = True)]
def update_pins(
pins: dict,
releases: Optional[Callable[[str], list[dict]]] = None,
hasher: Optional[Callable[[str], tuple[str, int, Optional[Needs]]]] = None,
only: Optional[str] = None,
) -> tuple[dict, list[str]]:
releases = releases or fetch_stable_releases
hasher = hasher or hash_asset
result = json.loads(json.dumps(pins))
changes = []
for name in [only] if only else _SECTIONS:
section = result[name]
for release in newer_releases(releases(section["repo"]), section["version"]):
try:
moved = updated_section(name, section, release, hasher)
except AbiTooNew as exc:
# stderr: the workflow builds the PR title from stdout.
print(f"skipped: {exc}", file = sys.stderr)
continue
changes.append(f"{name}: {section['version']} -> {moved['version']}")
result[name] = moved
break
return result, changes
def main(argv: Optional[list[str]] = None) -> int:
parser = argparse.ArgumentParser(description = __doc__.splitlines()[0])
parser.add_argument("--pins", type = Path, default = PINS_PATH)
parser.add_argument("--only", choices = sorted(_SECTIONS), help = "Move one section only.")
mode = parser.add_mutually_exclusive_group()
mode.add_argument("--check", action = "store_true", help = "Exit 1 if a newer release exists.")
mode.add_argument("--write", action = "store_true", help = "Rewrite the pins file.")
args = parser.parse_args(argv)
try:
pins = json.loads(args.pins.read_text(encoding = "utf-8"))
new_pins, changes = update_pins(pins, only = args.only)
except Exception as exc: # noqa: BLE001 -- one line for the CI log, exit 2
print(f"error: {exc}", file = sys.stderr)
return 2
for line in changes or ["Lemonade and FastFlowLM pins are current."]:
print(line)
if changes and args.write:
args.pins.write_text(json.dumps(new_pins, indent = 2) + "\n", encoding = "utf-8")
return 1 if changes and args.check else 0
if __name__ == "__main__":
raise SystemExit(main())