# SPDX-License-Identifier: AGPL-3.0-only # Copyright 2026-present the Unsloth AI Inc. team. All rights reserved. See /studio/LICENSE.AGPL-3.0 """GGUF variant resolution.""" from __future__ import annotations import asyncio import threading import time from collections import OrderedDict from typing import NamedTuple, Optional from fastapi import HTTPException from loggers import get_logger from hub.schemas.inventory import GgufVariantDetail, GgufVariantsResponse from hub.utils import download_manifest from hub.utils import download_registry from hub.utils import inventory_scan as hf_cache_scan from hub.utils.hf_errors import hf_error_status from hub.utils.hf_cache_state import ( INCOMPLETE_SUFFIX, iter_destructive_repo_cache_dirs, ) from hub.utils.gguf import ( extract_quant_label, iter_hf_cache_snapshots, is_big_endian_gguf_path, list_empty_gguf_variant_dirs, list_gguf_variants, list_gguf_variants_from_hf_cache, list_local_gguf_variants, list_partial_gguf_variants_from_state, pick_best_gguf, ) from hub.utils.paths import ( is_local_path, is_valid_repo_id as _is_valid_repo_id, ) from hub.services.models.common import ( _is_mmproj_filename, _is_mtp_drafter_path, _iter_gguf_paths, ) from hub.utils.gguf_plan import ( GgufVariantPlan as _GgufVariantRequirement, build_gguf_variant_plans, is_main_gguf_variant_path, ) logger = get_logger(__name__) _VARIANT_HASH_CACHE: "OrderedDict[tuple[str, str, str, bool], tuple[frozenset[str], float]]" = ( OrderedDict() ) _VARIANT_REQUIREMENT_CACHE: "OrderedDict[tuple[str, str, str], tuple[_GgufVariantRequirement, float]]" = OrderedDict() _VARIANT_REQUIREMENT_NEG_CACHE: "OrderedDict[tuple[str, str], float]" = OrderedDict() _VARIANT_HASH_MAX = 512 # Blob hashes are derived from the same mutable remote revision metadata as # variant requirements, so they must not outlive that freshness window. _VARIANT_HASH_POS_TTL = 60.0 # Refresh resolved variant requirements so a moved repo revision is picked up # within the session instead of being pinned for the backend's lifetime. _VARIANT_REQUIREMENT_POS_TTL = 60.0 # Suppress retries on a metadata-fetch failure so a slow/flaky link doesn't # re-hammer the API on every page refresh. _VARIANT_REQUIREMENT_NEG_TTL = 60.0 # Fail fast on a slow link so the variant render isn't blocked for seconds. _GGUF_METADATA_TIMEOUT_SECONDS = 5.0 _VARIANT_HASH_LOCK = threading.Lock() class VariantIncompleteDeleteResult(NamedTuple): deleted: int unresolved: bool def _variant_hash_cache_key( repo_id: str, variant: str, hf_token: Optional[str] ) -> tuple[str, str, str]: return ( repo_id.lower(), variant.lower(), hf_cache_scan.token_fingerprint(hf_token), ) def _variant_blob_hash_cache_key( repo_id: str, variant: str, hf_token: Optional[str], include_companions: bool ) -> tuple[str, str, str, bool]: base = _variant_hash_cache_key(repo_id, variant, hf_token) return (*base, include_companions) def _variant_repo_cache_key(repo_id: str, hf_token: Optional[str]) -> tuple[str, str]: return (repo_id.lower(), hf_cache_scan.token_fingerprint(hf_token)) def _variant_requirement_neg_cache_active(key: tuple[str, str]) -> bool: with _VARIANT_HASH_LOCK: cached_at = _VARIANT_REQUIREMENT_NEG_CACHE.get(key) if cached_at is None: return False if (time.monotonic() - cached_at) < _VARIANT_REQUIREMENT_NEG_TTL: _VARIANT_REQUIREMENT_NEG_CACHE.move_to_end(key) return True _VARIANT_REQUIREMENT_NEG_CACHE.pop(key, None) return False def _variant_requirement_neg_cache_set(key: tuple[str, str]) -> None: with _VARIANT_HASH_LOCK: _VARIANT_REQUIREMENT_NEG_CACHE[key] = time.monotonic() _VARIANT_REQUIREMENT_NEG_CACHE.move_to_end(key) while len(_VARIANT_REQUIREMENT_NEG_CACHE) > _VARIANT_HASH_MAX: _VARIANT_REQUIREMENT_NEG_CACHE.popitem(last = False) def _variant_requirement_neg_cache_clear(key: tuple[str, str]) -> None: with _VARIANT_HASH_LOCK: _VARIANT_REQUIREMENT_NEG_CACHE.pop(key, None) def _variant_hash_cache_get(key: tuple[str, str, str, bool]) -> Optional[frozenset[str]]: with _VARIANT_HASH_LOCK: cached = _VARIANT_HASH_CACHE.get(key) if cached is None: return None hashes, ts = cached if (time.monotonic() - ts) >= _VARIANT_HASH_POS_TTL: _VARIANT_HASH_CACHE.pop(key, None) return None _VARIANT_HASH_CACHE.move_to_end(key) return hashes def _variant_hash_cache_set(key: tuple[str, str, str, bool], hashes: frozenset[str]) -> None: with _VARIANT_HASH_LOCK: _VARIANT_HASH_CACHE[key] = (hashes, time.monotonic()) _VARIANT_HASH_CACHE.move_to_end(key) while len(_VARIANT_HASH_CACHE) > _VARIANT_HASH_MAX: _VARIANT_HASH_CACHE.popitem(last = False) def _variant_requirement_cache_get(key: tuple[str, str, str]) -> Optional[_GgufVariantRequirement]: with _VARIANT_HASH_LOCK: cached = _VARIANT_REQUIREMENT_CACHE.get(key) if cached is None: return None requirement, ts = cached if (time.monotonic() - ts) >= _VARIANT_REQUIREMENT_POS_TTL: _VARIANT_REQUIREMENT_CACHE.pop(key, None) return None _VARIANT_REQUIREMENT_CACHE.move_to_end(key) return requirement def _variant_requirement_cache_set_many( repo_id: str, hf_token: Optional[str], requirements: dict[str, _GgufVariantRequirement] ) -> None: with _VARIANT_HASH_LOCK: now = time.monotonic() for quant, requirement in requirements.items(): key = _variant_hash_cache_key(repo_id, quant, hf_token) _VARIANT_REQUIREMENT_CACHE[key] = (requirement, now) _VARIANT_REQUIREMENT_CACHE.move_to_end(key) while len(_VARIANT_REQUIREMENT_CACHE) > _VARIANT_HASH_MAX: _VARIANT_REQUIREMENT_CACHE.popitem(last = False) def _build_gguf_variant_requirements(siblings: list) -> dict[str, _GgufVariantRequirement]: return build_gguf_variant_plans(siblings) def gguf_variant_requirements( repo_id: str, variant: str, hf_token: Optional[str] = None, ) -> Optional[_GgufVariantRequirement]: key = _variant_hash_cache_key(repo_id, variant, hf_token) cached = _variant_requirement_cache_get(key) if cached is not None: return cached requirements = _fetch_gguf_variant_requirements(repo_id, hf_token) return requirements.get(variant.lower()) def _fetch_gguf_variant_requirements( repo_id: str, hf_token: Optional[str] = None, *, siblings: Optional[list] = None, ) -> dict[str, _GgufVariantRequirement]: repo_key = _variant_repo_cache_key(repo_id, hf_token) if siblings is None: if _variant_requirement_neg_cache_active(repo_key): return {} try: from huggingface_hub import HfApi info = HfApi(token = hf_token).model_info( repo_id, files_metadata = True, timeout = _GGUF_METADATA_TIMEOUT_SECONDS, ) except Exception as e: logger.warning( "model_info failed resolving GGUF files for %s: %s", repo_id, download_registry.scrub_secrets(str(e), hf_token = hf_token), ) _variant_requirement_neg_cache_set(repo_key) return {} siblings = list(info.siblings) requirements = _build_gguf_variant_requirements(siblings) if requirements: _variant_requirement_cache_set_many(repo_id, hf_token, requirements) _variant_requirement_neg_cache_clear(repo_key) return requirements def _gguf_all_variant_requirements( repo_id: str, hf_token: Optional[str] = None, *, siblings: Optional[list] = None, ) -> dict[str, _GgufVariantRequirement]: return _fetch_gguf_variant_requirements(repo_id, hf_token, siblings = siblings) def _manifest_variant_blob_hashes( repo_id: str, variant: str, *, include_companions: bool = True, ) -> frozenset[str]: manifest = download_manifest.read_manifest("model", repo_id, variant) if manifest is None: return frozenset() variant_key = variant.lower() hashes: set[str] = set() for expected in manifest.expected_files: if not expected.sha256: continue if include_companions: hashes.add(expected.sha256) continue if is_main_gguf_variant_path(expected.path, variant_key): hashes.add(expected.sha256) return frozenset(hashes) def gguf_variant_blob_hashes( repo_id: str, variant: str, hf_token: Optional[str] = None, *, include_companions: bool = True, allow_remote: bool = True, ) -> frozenset[str]: key = _variant_blob_hash_cache_key( repo_id, variant, hf_token, include_companions, ) cached = _variant_hash_cache_get(key) if cached is not None: return cached hashes = _manifest_variant_blob_hashes( repo_id, variant, include_companions = include_companions, ) if hashes: _variant_hash_cache_set(key, hashes) return hashes requirement_key = _variant_hash_cache_key(repo_id, variant, hf_token) requirement = _variant_requirement_cache_get(requirement_key) if requirement is None and allow_remote: requirement = gguf_variant_requirements(repo_id, variant, hf_token) if requirement is not None: hashes = requirement.required_hashes if include_companions else requirement.main_hashes if hashes: _variant_hash_cache_set(key, hashes) return hashes return frozenset() def _partial_transport_for_variant(repo_id: str, variant: str) -> Optional[str]: return hf_cache_scan.partial_transport_for("model", repo_id, variant) def _local_main_gguf_blobs_by_quant(repo_id: str) -> dict[str, dict[str, set[str]]]: """Map quant -> repo-relative expected GGUF filename -> cached blob hashes. Shared companions are copied into each main-quant bucket so update checks can detect mmproj/MTP-only upstream changes without a separate remote call. """ result: dict[str, dict[str, set[str]]] = {} companion_blobs: dict[str, set[str]] = {} try: from hub.services.models import cache_inventory scans = cache_inventory.all_hf_cache_scans() except Exception as e: logger.warning("Failed to scan local GGUF blobs for %s: %s", repo_id, e) return result target_lower = repo_id.lower() for hf_cache in scans: for repo_info in hf_cache.repos: if str(getattr(repo_info, "repo_type", "")) != "model": continue if str(getattr(repo_info, "repo_id", "")).lower() != target_lower: continue for path, hashes in cache_inventory._repo_gguf_blob_map( repo_info, include_companions = True, ).items(): normalized = str(path).replace("\\", "/") if not hashes: continue if _is_mmproj_filename(normalized) or _is_mtp_drafter_path(normalized): companion_blobs.setdefault(normalized, set()).update( str(blob) for blob in hashes if blob ) continue quant = extract_quant_label(normalized).lower() if is_big_endian_gguf_path(normalized, quant): continue bucket = result.setdefault(quant, {}).setdefault(normalized, set()) bucket.update(str(blob) for blob in hashes if blob) if companion_blobs: for local_blobs in result.values(): for path, hashes in companion_blobs.items(): local_blobs.setdefault(path, set()).update(hashes) return result def _variant_update_available_from_requirement( local_blobs: dict[str, set[str]], requirement: Optional[_GgufVariantRequirement], variant: str ) -> bool: if requirement is None or not local_blobs: return False local_by_posix = {path.replace("\\", "/"): blobs for path, blobs in local_blobs.items()} for expected in requirement.expected_files: path = str(expected.path).replace("\\", "/") if not ( is_main_gguf_variant_path(path, variant) or _is_mmproj_filename(path) or _is_mtp_drafter_path(path) ): continue remote_blob = expected.sha256 if not remote_blob: continue local_set = local_by_posix.get(path) if not local_set or remote_blob not in local_set: return True return False def delete_variant_incomplete_blobs_result( repo_id: str, variant: str, hf_token: Optional[str], *, extra_hashes: frozenset[str] = frozenset(), companions: bool = True, ) -> VariantIncompleteDeleteResult: # With a sibling still downloading, ``companions=False`` keeps a shared mmproj # from being unlinked out from under it; the repo's last delete reclaims it. target_hashes = ( gguf_variant_blob_hashes(repo_id, variant, hf_token, include_companions = companions) | extra_hashes ) if not target_hashes: has_variant_partial_state = hf_cache_scan.is_variant_partial( repo_id, variant, incomplete_blob_hashes = set(), variant_blob_hashes = frozenset(), ) has_repo_partials = bool(download_registry.incomplete_blob_hashes("model", repo_id)) return VariantIncompleteDeleteResult( deleted = 0, unresolved = has_variant_partial_state and has_repo_partials, ) deleted = 0 # Destructive iterator: only the exact-case match (or abort if ambiguous), # so a case-variant sibling repo's partials are never unlinked. for entry in iter_destructive_repo_cache_dirs("model", repo_id): blobs_dir = entry / "blobs" if not blobs_dir.is_dir(): continue for h in target_hashes: incomplete = blobs_dir / f"{h}{INCOMPLETE_SUFFIX}" if incomplete.exists(): try: incomplete.unlink() deleted += 1 except OSError as e: logger.warning(f"Failed to unlink {incomplete}: {e}") return VariantIncompleteDeleteResult(deleted = deleted, unresolved = False) def _mark_empty_dir_cleanables( repo_id: str, response: GgufVariantsResponse ) -> GgufVariantsResponse: """Surface empty leftover ``/`` folders (interrupted downloads) as partial so the UI can delete them -- on local/offline paths too, not just a remote listing. A listed quant is flipped to partial; an unlisted one is appended as a zero-byte cleanable entry.""" try: empty_labels = list_empty_gguf_variant_dirs(repo_id) except Exception as e: logger.warning(f"Failed to scan empty GGUF variant folders for {repo_id}: {e}") return response if not empty_labels: return response empty_by_key = {label.lower(): label for label in empty_labels} variants = list(response.variants) listed = {v.quant.lower() for v in variants} for i, v in enumerate(variants): if v.quant.lower() in empty_by_key and not v.downloaded and not v.partial: variants[i] = v.model_copy(update = {"partial": True}) for key, label in sorted(empty_by_key.items()): if key not in listed: variants.append(GgufVariantDetail(filename = f"{label}.gguf", quant = label, partial = True)) return response.model_copy(update = {"variants": variants}) async def get_gguf_variants_response( repo_id: str, prefer_local_cache: bool = False, offline: bool = False, local_path: Optional[str] = None, hf_token: Optional[str] = None, ): """ List available GGUF quantization variants for a HuggingFace repo or a local directory (e.g. LM Studio model folder). Returns all available quantization variants (Q4_K_M, Q8_0, BF16, etc.) with file sizes, whether the model supports vision, and the recommended default variant. """ def _compute() -> GgufVariantsResponse: def _local_response( response_repo_id: str, variants, has_vision: bool ) -> GgufVariantsResponse: filenames = [v.filename for v in variants] best = pick_best_gguf(filenames) default_variant = extract_quant_label(best) if best else None return GgufVariantsResponse( repo_id = response_repo_id, variants = [ GgufVariantDetail( filename = v.filename, quant = v.quant, display_label = v.display_label, size_bytes = v.size_bytes, download_size_bytes = v.size_bytes, downloaded = True, ) for v in variants ], has_vision = has_vision, default_variant = default_variant, ) def _partial_local_response( response_repo_id: str, variants, has_vision: bool ) -> GgufVariantsResponse: filenames = [v.filename for v in variants] best = pick_best_gguf(filenames) default_variant = extract_quant_label(best) if best else None return GgufVariantsResponse( repo_id = response_repo_id, variants = [ GgufVariantDetail( filename = v.filename, quant = v.quant, display_label = v.display_label, size_bytes = v.size_bytes, download_size_bytes = v.download_size_bytes or v.size_bytes, downloaded = False, partial = True, partial_transport = _partial_transport_for_variant( response_repo_id, v.quant, ), ) for v in variants ], has_vision = has_vision, default_variant = default_variant, ) # Local directory path (e.g. LM Studio models) — scan filesystem if is_local_path(repo_id): variants, has_vision = list_local_gguf_variants(repo_id) return _local_response(repo_id, variants, has_vision) # Reject invalid remote repo_ids up front (like download/delete) so a # malformed id returns 400 instead of a 500 from the HF client. if not _is_valid_repo_id(repo_id): raise HTTPException(status_code = 400, detail = f"Invalid repo_id: {repo_id!r}") local_only = prefer_local_cache or offline if local_only: cached = list_gguf_variants_from_hf_cache(repo_id) if cached is not None: variants, has_vision = cached return _local_response(repo_id, variants, has_vision) if local_path and is_local_path(local_path): variants, has_vision = list_local_gguf_variants(local_path) if variants or has_vision: return _local_response(repo_id, variants, has_vision) partial = list_partial_gguf_variants_from_state(repo_id) if partial is not None: variants, has_vision = partial return _partial_local_response(repo_id, variants, has_vision) if local_path and offline: return GgufVariantsResponse( repo_id = repo_id, variants = [], has_vision = False, default_variant = None, ) if offline: raise HTTPException( status_code = 404, detail = "No cached GGUF variants available while offline.", ) try: variants, has_vision, siblings = list_gguf_variants(repo_id, hf_token = hf_token) except Exception: cached = list_gguf_variants_from_hf_cache(repo_id) if cached is not None: variants, has_vision = cached return _local_response(repo_id, variants, has_vision) partial = list_partial_gguf_variants_from_state(repo_id) if partial is not None: variants, has_vision = partial return _partial_local_response(repo_id, variants, has_vision) raise filenames = [v.filename for v in variants] best = pick_best_gguf(filenames) default_variant = extract_quant_label(best) if best else None # Per-snapshot accounting: a variant counts as present only when one # snapshot holds all its files (split GGUFs need every shard together), # sizes are max across snapshots so shared blobs aren't double-counted, # and keys are lowercased since cache dir casing can differ from repo_id. cached_filenames_by_snapshot: list[dict[str, int]] = [] cached_quant_bytes_by_snapshot: list[dict[str, int]] = [] if _is_valid_repo_id(repo_id): for snap in iter_hf_cache_snapshots(repo_id): try: gguf_paths = list(_iter_gguf_paths(snap)) except (OSError, RuntimeError, ValueError) as e: logger.debug("Skipping GGUF cache snapshot %s: %s", snap, e) continue by_filename: dict[str, int] = {} by_quant: dict[str, int] = {} for f in gguf_paths: try: rel = f.relative_to(snap).as_posix() size = f.stat().st_size except (OSError, RuntimeError, ValueError) as e: logger.debug("Skipping GGUF cache file %s: %s", f, e) continue key = rel.lower() by_filename[key] = max(by_filename.get(key, 0), size) if _is_mmproj_filename(f.name) or _is_mtp_drafter_path(rel): continue q = extract_quant_label(rel) if is_big_endian_gguf_path(rel, q): continue q = q.lower() by_quant[q] = by_quant.get(q, 0) + size if by_filename: cached_filenames_by_snapshot.append(by_filename) if by_quant: cached_quant_bytes_by_snapshot.append(by_quant) requirements_by_quant = { v.quant.lower(): _variant_requirement_cache_get( _variant_hash_cache_key(repo_id, v.quant, hf_token) ) for v in variants } if any(req is None for req in requirements_by_quant.values()): fetched_requirements = _gguf_all_variant_requirements( repo_id, hf_token, siblings = siblings ) for v in variants: key = v.quant.lower() if requirements_by_quant.get(key) is None: requirements_by_quant[key] = fetched_requirements.get(key) def _filenames_cached(filenames: frozenset[str], expected_size: int) -> bool: if not filenames: return False wanted = [name.lower() for name in filenames] # All files must live in a single snapshot, not spread across several. for by_filename in cached_filenames_by_snapshot: cached = 0 for name in wanted: size = by_filename.get(name) if size is None: break cached += size else: return expected_size <= 0 or cached >= expected_size * 0.99 return False def _any_mmproj_cached(filenames: frozenset[str]) -> bool: if any( by_filename.get(name.lower()) is not None for by_filename in cached_filenames_by_snapshot for name in filenames ): return True return any( _is_mmproj_filename(name.rsplit("/", 1)[-1]) for by_filename in cached_filenames_by_snapshot for name in by_filename ) def _quant_bytes_present(quant: str, size_bytes: int) -> bool: # Small rounding tolerance for symlinks vs real sizes. if size_bytes <= 0: return False return any( by_quant.get(quant, 0) >= size_bytes * 0.99 for by_quant in cached_quant_bytes_by_snapshot ) def _is_fully_downloaded(variant) -> bool: quant = variant.quant.lower() requirement = requirements_by_quant.get(quant) # Vision repos ship an mmproj adapter; any precision on disk suffices. if ( requirement is not None and _filenames_cached( requirement.main_filenames, requirement.main_size_bytes, ) and ( not requirement.mmproj_filenames or _any_mmproj_cached(requirement.mmproj_filenames) ) ): return True # Byte fallback so a present quant isn't demoted by a filename mismatch; # vision repos still need an mmproj cached (any precision). if not _quant_bytes_present(quant, variant.size_bytes): return False if ( requirement is not None and requirement.mmproj_filenames and not _any_mmproj_cached(requirement.mmproj_filenames) ): return False return True partial_quants: set[str] = set() partial_quant_transports: dict[str, Optional[str]] = {} try: incomplete_hashes = download_registry.incomplete_blob_hashes("model", repo_id) except Exception as e: logger.warning(f"Failed to compute partial GGUF variants for {repo_id}: {e}") incomplete_hashes = set() scan_snapshot_dir = hf_cache_scan.resolve_snapshot_dir_for_scan("model", repo_id) # Manifest + marker + main incomplete-blob check: catches variants whose # download was cancelled or whose expected shards are missing/undersized. for variant in variants: try: requirement = requirements_by_quant.get(variant.quant.lower()) variant_hashes = requirement.main_hashes if requirement is not None else None if variant_hashes is None and incomplete_hashes: variant_hashes = gguf_variant_blob_hashes( repo_id, variant.quant, hf_token, include_companions = False, ) if hf_cache_scan.is_variant_partial( repo_id, variant.quant, scan_snapshot_dir, incomplete_blob_hashes = incomplete_hashes, variant_blob_hashes = variant_hashes, ): partial_quants.add(variant.quant) partial_quant_transports[variant.quant] = _partial_transport_for_variant( repo_id, variant.quant, ) except Exception as e: logger.warning( f"Manifest-based partial check failed for " f"{repo_id}/{variant.quant}: {e}" ) if incomplete_hashes: for variant in variants: requirement = requirements_by_quant.get(variant.quant.lower()) if requirement is None: continue # companion_hashes adds the MTP drafter (mmproj_hashes covers # every mmproj precision in the repo, not just the planned one). if ( (requirement.mmproj_hashes | requirement.companion_hashes) & incomplete_hashes ) and _filenames_cached( requirement.main_filenames, requirement.main_size_bytes, ): partial_quants.add(variant.quant) partial_quant_transports.setdefault( variant.quant, _partial_transport_for_variant(repo_id, variant.quant), ) local_blobs_by_quant = _local_main_gguf_blobs_by_quant(repo_id) def _variant_detail(v) -> GgufVariantDetail: is_partial = v.quant in partial_quants requirement = requirements_by_quant.get(v.quant.lower()) downloaded = _is_fully_downloaded(v) and not is_partial return GgufVariantDetail( filename = v.filename, quant = v.quant, display_label = v.display_label, size_bytes = v.size_bytes, download_size_bytes = ( requirement.download_size_bytes if requirement is not None else v.size_bytes ), downloaded = downloaded, update_available = downloaded and _variant_update_available_from_requirement( local_blobs_by_quant.get(v.quant.lower(), {}), requirement, v.quant, ), partial = is_partial, partial_transport = (partial_quant_transports.get(v.quant) if is_partial else None), ) return GgufVariantsResponse( repo_id = repo_id, variants = [_variant_detail(v) for v in variants], has_vision = has_vision, default_variant = default_variant, ) def _compute_with_cleanables() -> GgufVariantsResponse: skip = is_local_path(repo_id) or not _is_valid_repo_id(repo_id) try: response = _compute() except Exception: # Offline / metadata fetch failed with only an empty leftover # / folder cached: still surface it so the UI can delete it, # otherwise re-raise the original error. if skip: raise enriched = _mark_empty_dir_cleanables( repo_id, GgufVariantsResponse(repo_id = repo_id, variants = []) ) if enriched.variants: return enriched raise if skip: return response return _mark_empty_dir_cleanables(repo_id, response) try: return await asyncio.to_thread(_compute_with_cleanables) except HTTPException: raise except Exception as e: scrubbed = download_registry.scrub_secrets(str(e), hf_token = hf_token) # Client-side HF error (missing repo, gated, bad token): pass the status through. status = hf_error_status(e) if status is not None: raise HTTPException(status_code = status, detail = scrubbed) logger.error("Error listing GGUF variants for %s: %s", repo_id, scrubbed) raise HTTPException( status_code = 500, detail = "Failed to list GGUF variants: " + scrubbed, )