3427e3fd62* Studio: fix Downloaded model list disappearing and order it by last download The chat model picker scan for cached GGUF and safetensors models aborted whenever an auxiliary Hugging Face cache dir (such as ~/.cache/huggingface/hub) was unreadable, returning an empty list. That hid the Downloaded section and let already downloaded models appear under Recommended. Isolate each cache probe so an inaccessible directory is skipped instead of failing the scan. Also order Downloaded newest-first using cached blob mtimes (multi-quant repos group by their most recent quant), keep the section visible while searching, and make the per-quant downloaded check per-snapshot and mmproj aware so a Recommended quant is never falsely marked downloaded. * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * Studio: harden gguf-variants scan and dedupe by newest timestamp Guard f.stat() per file so a broken symlink or unreadable file in a snapshot no longer aborts the downloaded check early, and match quant labels case-insensitively. When the same repo is present in multiple caches with equal size, keep the newest last_modified so Downloaded ordering reflects the most recent copy. * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * Studio: apply cache-scan guards to sibling endpoints found in review Extend the inaccessible-cache guard and mmproj/stat hardening to the parallel HF cache code paths flagged in review: - list_local_models and the Hub inventory scan now skip an unreadable auxiliary cache instead of returning 500. - The GGUF download-progress endpoint excludes mmproj adapters and guards f.stat() so one bad file does not zero a repo's progress. - The offline snapshot scanner guards its is_dir() probes. - The chat-only picker no longer renders a blank list when a search matches only cached non-GGUF models. * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci --------- Co-authored-by: danielhanchen <michaelhan2050@gmail.com> Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
2746 lines
98 KiB
Python
2746 lines
98 KiB
Python
# SPDX-License-Identifier: AGPL-3.0-only
|
|
# Copyright 2026-present the Unsloth AI Inc. team. All rights reserved. See /studio/LICENSE.AGPL-3.0
|
|
|
|
"""Model management API routes."""
|
|
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import shutil
|
|
import sys
|
|
import uuid
|
|
from pathlib import Path
|
|
from fastapi import APIRouter, Body, Depends, HTTPException, Query
|
|
from typing import List, Optional
|
|
import structlog
|
|
from loggers import get_logger
|
|
from utils.utils import log_and_http_error
|
|
|
|
import re as _re
|
|
|
|
_VALID_REPO_ID = _re.compile(r"^[A-Za-z0-9._-]+/[A-Za-z0-9._-]+$")
|
|
|
|
|
|
def _is_valid_repo_id(repo_id: str) -> bool:
|
|
return bool(_VALID_REPO_ID.fullmatch(repo_id))
|
|
|
|
|
|
def _safe_is_dir(path) -> bool:
|
|
"""``Path.is_dir()`` returning ``False`` instead of raising.
|
|
|
|
Python >= 3.12 propagates ``PermissionError`` from ``is_dir()``;
|
|
folder-scan endpoints probe system locations (e.g. root-owned
|
|
``/usr/share/ollama``) and must treat un-stat-able paths as "not a
|
|
directory", never 500.
|
|
"""
|
|
try:
|
|
return Path(path).is_dir()
|
|
except OSError:
|
|
return False
|
|
|
|
|
|
def _is_hidden_model(*values: str | None) -> bool:
|
|
"""True if any id/path is the RAG embedding model (EMBEDDING_MODEL or
|
|
EMBED_GGUF_REPO basename), so pickers hide it (GGUF and non-GGUF)."""
|
|
from core.rag import config as rag_config
|
|
|
|
needles = (
|
|
rag_config.EMBEDDING_MODEL.split("/")[-1].lower(),
|
|
rag_config.EMBED_GGUF_REPO.split("/")[-1].lower(),
|
|
)
|
|
return any(v and any(n in v.lower() for n in needles) for v in values)
|
|
|
|
|
|
def _safe_resolve(path: Path) -> Optional[str]:
|
|
"""resolve() to a string, or None when the path is inaccessible."""
|
|
try:
|
|
return str(path.resolve())
|
|
except OSError:
|
|
return None
|
|
|
|
|
|
backend_path = Path(__file__).parent.parent.parent
|
|
if str(backend_path) not in sys.path:
|
|
sys.path.insert(0, str(backend_path))
|
|
|
|
from auth.authentication import get_current_subject
|
|
|
|
try:
|
|
from utils.models import (
|
|
scan_trained_models,
|
|
scan_exported_models,
|
|
get_base_model_from_checkpoint,
|
|
load_model_defaults,
|
|
get_base_model_from_lora,
|
|
is_vision_model,
|
|
is_embedding_model,
|
|
scan_checkpoints,
|
|
list_gguf_variants,
|
|
ModelConfig,
|
|
)
|
|
from utils.models.model_config import (
|
|
_pick_best_gguf,
|
|
_extract_quant_label,
|
|
is_audio_input_type,
|
|
)
|
|
from core.inference import get_inference_backend
|
|
from utils.paths import (
|
|
is_local_path,
|
|
outputs_root,
|
|
exports_root,
|
|
resolve_cached_repo_id_case,
|
|
resolve_output_dir,
|
|
resolve_export_dir,
|
|
)
|
|
except ImportError:
|
|
# Fallback: import from parent directory.
|
|
parent_backend = backend_path.parent / "backend"
|
|
if str(parent_backend) not in sys.path:
|
|
sys.path.insert(0, str(parent_backend))
|
|
from utils.models import (
|
|
scan_trained_models,
|
|
scan_exported_models,
|
|
get_base_model_from_checkpoint,
|
|
load_model_defaults,
|
|
get_base_model_from_lora,
|
|
is_vision_model,
|
|
is_embedding_model,
|
|
scan_checkpoints,
|
|
list_gguf_variants,
|
|
ModelConfig,
|
|
)
|
|
from utils.models.model_config import (
|
|
_pick_best_gguf,
|
|
_extract_quant_label,
|
|
is_audio_input_type,
|
|
)
|
|
from core.inference import get_inference_backend
|
|
from utils.paths import (
|
|
is_local_path,
|
|
outputs_root,
|
|
exports_root,
|
|
resolve_cached_repo_id_case,
|
|
resolve_output_dir,
|
|
resolve_export_dir,
|
|
)
|
|
|
|
from models import (
|
|
CheckpointInfo,
|
|
CheckpointListResponse,
|
|
LocalModelInfo,
|
|
LocalModelListResponse,
|
|
ModelCheckpoints,
|
|
ModelDetails,
|
|
LoRAScanResponse,
|
|
LoRAInfo,
|
|
ModelListResponse,
|
|
)
|
|
from models.models import (
|
|
BrowseEntry,
|
|
BrowseFoldersResponse,
|
|
GgufVariantDetail,
|
|
GgufVariantsResponse,
|
|
ModelType,
|
|
ScanFolderInfo,
|
|
AddScanFolderRequest,
|
|
)
|
|
from models.responses import (
|
|
LoRABaseModelResponse,
|
|
VisionCheckResponse,
|
|
EmbeddingCheckResponse,
|
|
)
|
|
|
|
router = APIRouter()
|
|
logger = get_logger(__name__)
|
|
|
|
|
|
def derive_model_type(
|
|
is_vision: bool,
|
|
audio_type: Optional[str],
|
|
is_embedding: bool = False,
|
|
) -> ModelType:
|
|
"""Collapse individual capability flags into a single model modality string."""
|
|
if is_embedding:
|
|
return "embeddings"
|
|
if audio_type is not None:
|
|
return "audio"
|
|
if is_vision:
|
|
return "vision"
|
|
return "text"
|
|
|
|
|
|
def _resolve_hf_cache_dir() -> Path:
|
|
"""Resolve local HF cache root used by hub downloads."""
|
|
try:
|
|
from huggingface_hub.constants import HF_HUB_CACHE
|
|
return Path(HF_HUB_CACHE)
|
|
except Exception:
|
|
return Path.home() / ".cache" / "huggingface" / "hub"
|
|
|
|
|
|
def _is_model_directory(d: Path) -> bool:
|
|
"""Return ``True`` when *d* looks like a model directory.
|
|
|
|
Requires both a config (``config.json``/``adapter_config.json``) and
|
|
weight files. Excludes ``mmproj`` GGUFs (vision projectors) and
|
|
non-weight ``.bin`` files (``tokenizer.bin`` etc.) to avoid false
|
|
positives.
|
|
"""
|
|
|
|
def _is_weight_file(f: Path) -> bool:
|
|
suffix = f.suffix.lower()
|
|
if suffix == ".safetensors":
|
|
return True
|
|
if suffix == ".gguf":
|
|
return "mmproj" not in f.name.lower()
|
|
if suffix == ".bin":
|
|
name = f.name.lower()
|
|
return (
|
|
name.startswith("pytorch_model")
|
|
or name.startswith("model")
|
|
or name.startswith("adapter_model")
|
|
or name.startswith("consolidated")
|
|
)
|
|
return False
|
|
|
|
try:
|
|
has_config = (d / "config.json").exists() or (d / "adapter_config.json").exists()
|
|
if not has_config:
|
|
return False
|
|
return any(_is_weight_file(f) for f in d.iterdir() if f.is_file())
|
|
except OSError:
|
|
return False
|
|
|
|
|
|
def _scan_models_dir(models_dir: Path, *, limit: int | None = None) -> List[LocalModelInfo]:
|
|
if not models_dir.exists() or not models_dir.is_dir():
|
|
return []
|
|
|
|
_is_self_model = _is_model_directory(models_dir)
|
|
|
|
if _is_self_model:
|
|
try:
|
|
updated_at = models_dir.stat().st_mtime
|
|
except OSError:
|
|
updated_at = None
|
|
return [
|
|
LocalModelInfo(
|
|
id = str(models_dir),
|
|
display_name = models_dir.name,
|
|
path = str(models_dir),
|
|
source = "models_dir",
|
|
updated_at = updated_at,
|
|
),
|
|
]
|
|
|
|
found: List[LocalModelInfo] = []
|
|
for child in models_dir.iterdir():
|
|
if limit is not None and len(found) >= limit:
|
|
break
|
|
try:
|
|
if not child.is_dir():
|
|
continue
|
|
has_model_files = (
|
|
(child / "config.json").exists()
|
|
or (child / "adapter_config.json").exists()
|
|
or any(child.glob("*.safetensors"))
|
|
or any(child.glob("*.bin"))
|
|
or any(child.glob("*.gguf"))
|
|
)
|
|
except OSError:
|
|
# Skip unreadable children rather than failing the scan.
|
|
continue
|
|
if not has_model_files:
|
|
continue
|
|
try:
|
|
updated_at = child.stat().st_mtime
|
|
except OSError:
|
|
updated_at = None
|
|
found.append(
|
|
LocalModelInfo(
|
|
id = str(child),
|
|
display_name = child.name,
|
|
path = str(child),
|
|
source = "models_dir",
|
|
updated_at = updated_at,
|
|
),
|
|
)
|
|
# Also scan standalone .gguf files in the models directory.
|
|
if limit is None or len(found) < limit:
|
|
for gguf_file in models_dir.glob("*.gguf"):
|
|
if limit is not None and len(found) >= limit:
|
|
break
|
|
if gguf_file.is_file():
|
|
try:
|
|
updated_at = gguf_file.stat().st_mtime
|
|
except OSError:
|
|
updated_at = None
|
|
found.append(
|
|
LocalModelInfo(
|
|
id = str(gguf_file),
|
|
display_name = gguf_file.stem,
|
|
path = str(gguf_file),
|
|
source = "models_dir",
|
|
updated_at = updated_at,
|
|
),
|
|
)
|
|
|
|
return found
|
|
|
|
|
|
def _scan_hf_cache(cache_dir: Path) -> List[LocalModelInfo]:
|
|
if not cache_dir.exists() or not cache_dir.is_dir():
|
|
return []
|
|
|
|
found: List[LocalModelInfo] = []
|
|
for repo_dir in cache_dir.glob("models--*"):
|
|
if not repo_dir.is_dir():
|
|
continue
|
|
|
|
repo_name = repo_dir.name[len("models--") :]
|
|
if not repo_name:
|
|
continue
|
|
model_id = repo_name.replace("--", "/")
|
|
|
|
try:
|
|
updated_at = repo_dir.stat().st_mtime
|
|
except OSError:
|
|
updated_at = None
|
|
|
|
found.append(
|
|
LocalModelInfo(
|
|
id = model_id,
|
|
model_id = model_id,
|
|
display_name = model_id.split("/")[-1],
|
|
path = str(repo_dir),
|
|
source = "hf_cache",
|
|
updated_at = updated_at,
|
|
),
|
|
)
|
|
return found
|
|
|
|
|
|
def _scan_lmstudio_dir(lm_dir: Path) -> List[LocalModelInfo]:
|
|
"""Scan an LM Studio models directory for model files.
|
|
|
|
LM Studio uses a ``publisher/model-name`` folder structure with GGUF
|
|
files, or standalone GGUF files at the top level.
|
|
"""
|
|
if not lm_dir.exists() or not lm_dir.is_dir():
|
|
return []
|
|
|
|
# If lm_dir is itself a model directory (not a publisher structure),
|
|
# return it as a single entry rather than skipping it silently.
|
|
if _is_model_directory(lm_dir):
|
|
try:
|
|
updated_at = lm_dir.stat().st_mtime
|
|
except OSError:
|
|
updated_at = None
|
|
return [
|
|
LocalModelInfo(
|
|
id = str(lm_dir),
|
|
display_name = lm_dir.name,
|
|
path = str(lm_dir),
|
|
source = "lmstudio",
|
|
updated_at = updated_at,
|
|
),
|
|
]
|
|
|
|
found: List[LocalModelInfo] = []
|
|
for child in lm_dir.iterdir():
|
|
try:
|
|
if not child.is_dir():
|
|
if child.suffix == ".gguf" and child.is_file():
|
|
try:
|
|
updated_at = child.stat().st_mtime
|
|
except OSError:
|
|
updated_at = None
|
|
found.append(
|
|
LocalModelInfo(
|
|
id = str(child),
|
|
display_name = child.stem,
|
|
path = str(child),
|
|
source = "lmstudio",
|
|
updated_at = updated_at,
|
|
),
|
|
)
|
|
continue
|
|
|
|
# Surface a model-directory child directly instead of
|
|
# descending into it as a publisher.
|
|
if _is_model_directory(child):
|
|
try:
|
|
updated_at = child.stat().st_mtime
|
|
except OSError:
|
|
updated_at = None
|
|
found.append(
|
|
LocalModelInfo(
|
|
id = str(child),
|
|
display_name = child.name,
|
|
path = str(child),
|
|
source = "lmstudio",
|
|
updated_at = updated_at,
|
|
),
|
|
)
|
|
continue
|
|
|
|
# child is a publisher directory; scan its subdirectories.
|
|
for model_dir in child.iterdir():
|
|
try:
|
|
if model_dir.is_dir():
|
|
has_model = (
|
|
any(model_dir.glob("*.gguf"))
|
|
or (model_dir / "config.json").exists()
|
|
or any(model_dir.glob("*.safetensors"))
|
|
)
|
|
if not has_model:
|
|
continue
|
|
model_id = f"{child.name}/{model_dir.name}"
|
|
try:
|
|
updated_at = model_dir.stat().st_mtime
|
|
except OSError:
|
|
updated_at = None
|
|
found.append(
|
|
LocalModelInfo(
|
|
id = str(model_dir),
|
|
model_id = model_id,
|
|
display_name = model_dir.name,
|
|
path = str(model_dir),
|
|
source = "lmstudio",
|
|
updated_at = updated_at,
|
|
),
|
|
)
|
|
elif model_dir.suffix == ".gguf" and model_dir.is_file():
|
|
try:
|
|
updated_at = model_dir.stat().st_mtime
|
|
except OSError:
|
|
updated_at = None
|
|
found.append(
|
|
LocalModelInfo(
|
|
id = str(model_dir),
|
|
model_id = f"{child.name}/{model_dir.stem}",
|
|
display_name = model_dir.stem,
|
|
path = str(model_dir),
|
|
source = "lmstudio",
|
|
updated_at = updated_at,
|
|
),
|
|
)
|
|
except OSError:
|
|
continue
|
|
except OSError:
|
|
continue
|
|
return found
|
|
|
|
|
|
def _ollama_links_dir(ollama_dir: Path) -> Optional[Path]:
|
|
"""Return a writable directory for Ollama ``.gguf`` symlinks.
|
|
|
|
Prefers ``<ollama_dir>/.studio_links/`` so links sit next to their
|
|
blobs; falls back to a per-ollama-dir namespace under Studio's cache
|
|
when the models dir is read-only (common for system installs).
|
|
"""
|
|
from utils.paths.storage_roots import cache_root
|
|
|
|
primary = ollama_dir / ".studio_links"
|
|
try:
|
|
primary.mkdir(exist_ok = True)
|
|
return primary
|
|
except OSError as e:
|
|
logger.debug(
|
|
"Ollama dir %s not writable for .studio_links (%s); falling back to Studio cache",
|
|
ollama_dir,
|
|
e,
|
|
)
|
|
|
|
# Fallback: namespace by a hash of ollama_dir so two roots don't
|
|
# collide. Cache path, not a security boundary.
|
|
try:
|
|
digest = hashlib.sha256(str(ollama_dir.resolve()).encode()).hexdigest()[:12]
|
|
except OSError:
|
|
digest = "default"
|
|
fallback = cache_root() / "ollama_links" / digest
|
|
try:
|
|
fallback.mkdir(parents = True, exist_ok = True)
|
|
return fallback
|
|
except OSError as e:
|
|
logger.warning(
|
|
"Could not create Ollama symlink cache at %s: %s",
|
|
fallback,
|
|
e,
|
|
)
|
|
return None
|
|
|
|
|
|
def _scan_ollama_dir(ollama_dir: Path, limit: Optional[int] = None) -> List[LocalModelInfo]:
|
|
"""Scan an Ollama models directory for downloaded models.
|
|
|
|
Ollama uses a content-addressable layout
|
|
(``manifests/<host>/<namespace>/<model>/<tag>`` + ``blobs/sha256-...``);
|
|
we ``rglob`` all manifests so every layout depth is found. Each
|
|
manifest is JSON with a ``layers`` array: the
|
|
``application/vnd.ollama.image.model`` layer holds the GGUF weights
|
|
and ``...image.projector`` is the vision adapter.
|
|
|
|
Ollama blobs lack the ``.gguf`` extension the loading pipeline
|
|
requires, so we create ``.gguf``-named links to them (one subdir per
|
|
model, keyed by a short hash of the manifest path, so
|
|
``detect_mmproj_file`` only sees that model's projector). Links are
|
|
symlinks when possible, else hardlinks; the link dir is
|
|
``.studio_links/`` when writable, else Studio's cache.
|
|
"""
|
|
manifests_root = ollama_dir / "manifests"
|
|
if not manifests_root.is_dir():
|
|
return []
|
|
|
|
found: List[LocalModelInfo] = []
|
|
blobs_dir = ollama_dir / "blobs"
|
|
links_root = _ollama_links_dir(ollama_dir)
|
|
if links_root is None:
|
|
logger.warning(
|
|
"Skipping Ollama scan for %s: no writable location for .gguf links",
|
|
ollama_dir,
|
|
)
|
|
return []
|
|
|
|
def _make_link(link_dir: Path, link_name: str, target: Path) -> Optional[str]:
|
|
"""Create a .gguf-named link to an Ollama blob.
|
|
|
|
Tries symlink, then hardlink; skips the model if neither works
|
|
(a multi-GB copy in a sync request would block the backend).
|
|
Idempotent: skips recreation when a valid link already exists.
|
|
"""
|
|
link_dir.mkdir(parents = True, exist_ok = True)
|
|
link_path = link_dir / link_name
|
|
resolved = target.resolve()
|
|
|
|
# Skip if the link already points at the same blob. Use samefile
|
|
# only; size checks can reuse stale links after `ollama pull`.
|
|
try:
|
|
if link_path.exists() and os.path.samefile(str(link_path), str(resolved)):
|
|
return str(link_path)
|
|
except OSError as e:
|
|
logger.debug("Error checking existing link %s: %s", link_path, e)
|
|
|
|
tmp_path = link_dir / f".{link_name}.tmp-{uuid.uuid4().hex[:8]}"
|
|
try:
|
|
if tmp_path.is_symlink() or tmp_path.exists():
|
|
tmp_path.unlink()
|
|
try:
|
|
tmp_path.symlink_to(resolved)
|
|
except OSError:
|
|
try:
|
|
os.link(str(resolved), str(tmp_path))
|
|
except OSError:
|
|
logger.warning(
|
|
"Could not create link for Ollama blob %s "
|
|
"(symlinks and hardlinks both failed). "
|
|
"Skipping model to avoid blocking the API.",
|
|
target,
|
|
)
|
|
return None
|
|
os.replace(str(tmp_path), str(link_path))
|
|
return str(link_path)
|
|
except OSError as e:
|
|
logger.debug("Could not create Ollama link %s: %s", link_path, e)
|
|
try:
|
|
if tmp_path.is_symlink() or tmp_path.exists():
|
|
tmp_path.unlink()
|
|
except OSError as cleanup_err:
|
|
logger.debug("Could not clean up tmp path %s: %s", tmp_path, cleanup_err)
|
|
return None
|
|
|
|
try:
|
|
for tag_file in manifests_root.rglob("*"):
|
|
if not tag_file.is_file():
|
|
continue
|
|
|
|
rel = tag_file.relative_to(manifests_root)
|
|
parts = rel.parts
|
|
if len(parts) < 3:
|
|
continue
|
|
|
|
host = parts[0]
|
|
repo_parts = list(parts[1:-1])
|
|
tag = parts[-1]
|
|
|
|
if host == "registry.ollama.ai" and repo_parts and repo_parts[0] == "library":
|
|
repo_name = "/".join(repo_parts[1:])
|
|
elif host == "registry.ollama.ai":
|
|
repo_name = "/".join(repo_parts)
|
|
else:
|
|
repo_name = "/".join([host] + repo_parts)
|
|
|
|
if not repo_name:
|
|
continue
|
|
|
|
display = f"{repo_name}:{tag}"
|
|
|
|
manifest_key = rel.as_posix()
|
|
stem_hash = hashlib.sha256(manifest_key.encode()).hexdigest()[:10]
|
|
|
|
try:
|
|
manifest = json.loads(tag_file.read_text())
|
|
except (json.JSONDecodeError, OSError) as e:
|
|
logger.debug(
|
|
"Skipping unreadable/invalid Ollama manifest %s: %s",
|
|
tag_file,
|
|
e,
|
|
)
|
|
continue
|
|
|
|
config_digest = manifest.get("config", {}).get("digest", "")
|
|
model_type = ""
|
|
file_type = ""
|
|
if config_digest and blobs_dir.is_dir():
|
|
config_blob = blobs_dir / config_digest.replace(":", "-")
|
|
if config_blob.is_file():
|
|
try:
|
|
cfg = json.loads(config_blob.read_text())
|
|
model_type = cfg.get("model_type", "")
|
|
file_type = cfg.get("file_type", "")
|
|
except (json.JSONDecodeError, OSError) as e:
|
|
logger.debug(
|
|
"Could not parse Ollama config blob %s: %s",
|
|
config_blob,
|
|
e,
|
|
)
|
|
|
|
model_link_dir = links_root / stem_hash
|
|
|
|
gguf_link_path: Optional[str] = None
|
|
quant = f"-{file_type}" if file_type else ""
|
|
safe_name = repo_name.replace("/", "-")
|
|
for layer in manifest.get("layers") or []:
|
|
media = layer.get("mediaType", "")
|
|
digest = layer.get("digest", "")
|
|
if not digest:
|
|
continue
|
|
|
|
if media == "application/vnd.ollama.image.model":
|
|
candidate = blobs_dir / digest.replace(":", "-")
|
|
if candidate.is_file():
|
|
link_name = f"{safe_name}-{tag}{quant}.gguf"
|
|
gguf_link_path = _make_link(model_link_dir, link_name, candidate)
|
|
|
|
elif media == "application/vnd.ollama.image.projector":
|
|
candidate = blobs_dir / digest.replace(":", "-")
|
|
if candidate.is_file():
|
|
mmproj_name = f"{safe_name}-{tag}-mmproj.gguf"
|
|
_make_link(model_link_dir, mmproj_name, candidate)
|
|
|
|
if not gguf_link_path:
|
|
continue
|
|
|
|
suffix = ""
|
|
if model_type:
|
|
suffix += f" ({model_type}"
|
|
if file_type:
|
|
suffix += f" {file_type}"
|
|
suffix += ")"
|
|
|
|
try:
|
|
updated_at = tag_file.stat().st_mtime
|
|
except OSError:
|
|
updated_at = None
|
|
|
|
found.append(
|
|
LocalModelInfo(
|
|
id = gguf_link_path,
|
|
model_id = f"ollama/{repo_name}:{tag}",
|
|
display_name = display + suffix,
|
|
path = gguf_link_path,
|
|
source = "custom",
|
|
updated_at = updated_at,
|
|
),
|
|
)
|
|
if limit is not None and len(found) >= limit:
|
|
return found
|
|
except OSError as e:
|
|
logger.warning("Error scanning Ollama directory %s: %s", ollama_dir, e)
|
|
return found
|
|
|
|
|
|
@router.get("/local", response_model = LocalModelListResponse)
|
|
async def list_local_models(
|
|
models_dir: str = Query(
|
|
default = "./models", description = "Directory to scan for local model folders"
|
|
),
|
|
current_subject: str = Depends(get_current_subject),
|
|
):
|
|
"""List local model candidates from the models dir, HF caches, and LM Studio dirs."""
|
|
from utils.paths import (
|
|
legacy_hf_cache_dir,
|
|
hf_default_cache_dir,
|
|
lmstudio_model_dirs,
|
|
)
|
|
|
|
# Resolve all scan directories up front.
|
|
hf_cache_dir = _resolve_hf_cache_dir()
|
|
legacy_hf = legacy_hf_cache_dir()
|
|
hf_default = hf_default_cache_dir()
|
|
lm_dirs = lmstudio_model_dirs()
|
|
|
|
# Validate models_dir against an allowlist of trusted dirs. Only the
|
|
# trusted Path objects are used for FS access; the user string is
|
|
# used for matching only, never for path construction.
|
|
allowed_roots: list[Path] = [Path("./models").resolve(), hf_cache_dir]
|
|
if _safe_is_dir(legacy_hf):
|
|
allowed_roots.append(legacy_hf)
|
|
if _safe_is_dir(hf_default):
|
|
allowed_roots.append(hf_default)
|
|
try:
|
|
from utils.paths import studio_root, outputs_root
|
|
allowed_roots.extend([studio_root(), outputs_root()])
|
|
except Exception:
|
|
pass
|
|
|
|
requested = os.path.realpath(os.path.expanduser(models_dir))
|
|
models_root = None
|
|
for root in allowed_roots:
|
|
root_str = os.path.realpath(str(root))
|
|
if requested == root_str or requested.startswith(root_str + os.sep):
|
|
models_root = root # trusted root, not the user-supplied path
|
|
break
|
|
if models_root is None:
|
|
raise HTTPException(
|
|
status_code = 403,
|
|
detail = "Directory not allowed",
|
|
)
|
|
|
|
try:
|
|
local_models = _scan_models_dir(models_root) + _scan_hf_cache(hf_cache_dir)
|
|
|
|
# Resolve once; an inaccessible aux cache must skip that scan, not 500.
|
|
hf_cache_real = _safe_resolve(hf_cache_dir)
|
|
legacy_real = _safe_resolve(legacy_hf)
|
|
default_real = _safe_resolve(hf_default)
|
|
|
|
# Scan legacy Unsloth HF cache for backward compatibility.
|
|
if _safe_is_dir(legacy_hf) and legacy_real != hf_cache_real:
|
|
local_models += _scan_hf_cache(legacy_hf)
|
|
|
|
# Scan HF system default cache (may differ under env overrides).
|
|
if (
|
|
_safe_is_dir(hf_default)
|
|
and default_real != hf_cache_real
|
|
and default_real != legacy_real
|
|
):
|
|
local_models += _scan_hf_cache(hf_default)
|
|
|
|
# Scan LM Studio directories.
|
|
for lm_dir in lm_dirs:
|
|
local_models += _scan_lmstudio_dir(lm_dir)
|
|
|
|
# Scan user-added custom folders (per-folder cap).
|
|
from storage.studio_db import list_scan_folders
|
|
|
|
_MAX_MODELS_PER_FOLDER = 200
|
|
try:
|
|
custom_folders = list_scan_folders()
|
|
except Exception as e:
|
|
logger.warning("Could not load custom scan folders: %s", e)
|
|
custom_folders = []
|
|
for folder in custom_folders:
|
|
folder_path = Path(folder["path"])
|
|
try:
|
|
# Filter Ollama .studio_links/ from generic scanners to
|
|
# avoid duplicates and leaking internal paths into the UI.
|
|
_generic = [
|
|
m
|
|
for m in (
|
|
_scan_models_dir(folder_path, limit = _MAX_MODELS_PER_FOLDER)
|
|
+ _scan_hf_cache(folder_path)
|
|
+ _scan_lmstudio_dir(folder_path)
|
|
)
|
|
if not any(p in (".studio_links", "ollama_links") for p in Path(m.path).parts)
|
|
]
|
|
custom_models = _generic
|
|
if len(custom_models) < _MAX_MODELS_PER_FOLDER:
|
|
custom_models += _scan_ollama_dir(
|
|
folder_path,
|
|
limit = _MAX_MODELS_PER_FOLDER - len(custom_models),
|
|
)
|
|
except OSError as e:
|
|
logger.warning("Skipping unreadable scan folder %s: %s", folder_path, e)
|
|
continue
|
|
local_models += [m.model_copy(update = {"source": "custom"}) for m in custom_models]
|
|
|
|
# Deduplicate, but always keep custom folder entries (keyed by
|
|
# (id, source)) so they show in the "Custom Folders" UI section
|
|
# even when the model is also in the HF cache.
|
|
deduped: dict[str, LocalModelInfo] = {}
|
|
for model in local_models:
|
|
key = f"{model.id}\x00custom" if model.source == "custom" else model.id
|
|
if key not in deduped:
|
|
deduped[key] = model
|
|
|
|
models = sorted(
|
|
deduped.values(),
|
|
key = lambda item: (item.updated_at or 0),
|
|
reverse = True,
|
|
)
|
|
models = [m for m in models if not _is_hidden_model(m.id, m.path)]
|
|
|
|
return LocalModelListResponse(
|
|
models_dir = str(models_root),
|
|
hf_cache_dir = str(hf_cache_dir),
|
|
lmstudio_dirs = [str(d) for d in lm_dirs],
|
|
models = models,
|
|
)
|
|
except Exception as e:
|
|
raise log_and_http_error(
|
|
e,
|
|
500,
|
|
"Failed to list local models",
|
|
event = "models.list_local_models_failed",
|
|
log = logger,
|
|
)
|
|
|
|
|
|
@router.get("/scan-folders")
|
|
async def get_scan_folders(current_subject: str = Depends(get_current_subject)):
|
|
"""List all registered custom model scan folders."""
|
|
from storage.studio_db import list_scan_folders
|
|
return {"folders": list_scan_folders()}
|
|
|
|
|
|
@router.post("/scan-folders", response_model = ScanFolderInfo, status_code = 201)
|
|
async def add_scan_folder_endpoint(
|
|
body: AddScanFolderRequest, current_subject: str = Depends(get_current_subject)
|
|
):
|
|
"""Register a new directory to scan for local models."""
|
|
from storage.studio_db import add_scan_folder
|
|
|
|
try:
|
|
folder = add_scan_folder(body.path)
|
|
except ValueError as e:
|
|
logger.warning("Scan folder rejected: %s (path=%s)", e, body.path)
|
|
# Forward the curated, path-free validation message.
|
|
rejection_message = str(e)
|
|
raise HTTPException(status_code = 400, detail = rejection_message)
|
|
logger.info("Scan folder added: %s", folder.get("path"))
|
|
return folder
|
|
|
|
|
|
@router.delete("/scan-folders/{folder_id}")
|
|
async def remove_scan_folder_endpoint(
|
|
folder_id: int, current_subject: str = Depends(get_current_subject)
|
|
):
|
|
"""Remove a registered custom scan folder."""
|
|
from storage.studio_db import remove_scan_folder
|
|
|
|
remove_scan_folder(folder_id)
|
|
logger.info("Scan folder removed: id=%s", folder_id)
|
|
return {"ok": True}
|
|
|
|
|
|
@router.get("/recommended-folders")
|
|
async def get_recommended_folders(current_subject: str = Depends(get_current_subject)):
|
|
"""Return well-known model directories that exist on this machine.
|
|
|
|
Lightweight alternative to ``browse-folders`` for the frontend's
|
|
one-click "Recommended" chips; returns existing paths only (HF
|
|
cache, LM Studio, Ollama, ``~/models``, etc.).
|
|
"""
|
|
from utils.paths.storage_roots import lmstudio_model_dirs
|
|
|
|
folders: list[str] = []
|
|
seen: set[str] = set()
|
|
|
|
def _add(p: Optional[Path]) -> None:
|
|
if p is None:
|
|
return
|
|
try:
|
|
resolved = str(p.resolve())
|
|
except OSError:
|
|
return
|
|
if resolved in seen:
|
|
return
|
|
if _safe_is_dir(resolved) and os.access(resolved, os.R_OK | os.X_OK):
|
|
seen.add(resolved)
|
|
folders.append(resolved)
|
|
|
|
# LM Studio model directories.
|
|
try:
|
|
for p in lmstudio_model_dirs():
|
|
_add(p)
|
|
except Exception as e:
|
|
logger.warning("Failed to scan for LM Studio model directories: %s", e)
|
|
|
|
# Ollama model directories.
|
|
ollama_env = os.environ.get("OLLAMA_MODELS")
|
|
if ollama_env:
|
|
_add(Path(ollama_env).expanduser())
|
|
for candidate in (
|
|
Path.home() / ".ollama" / "models",
|
|
Path("/usr/share/ollama/.ollama/models"),
|
|
Path("/var/lib/ollama/.ollama/models"),
|
|
):
|
|
_add(candidate)
|
|
|
|
return {"folders": folders}
|
|
|
|
|
|
# Max children to stat when checking if a directory "looks like" it
|
|
# holds models; keeps the browser snappy on huge dirs.
|
|
_BROWSE_MODEL_HINT_PROBE = 64
|
|
# Hard cap on subdirectory entries returned, so browsing ``/usr/lib``
|
|
# can't stat-storm the process or flood the client.
|
|
_BROWSE_ENTRY_CAP = 2000
|
|
|
|
|
|
def _count_model_files(directory: Path, cap: int = 200) -> int:
|
|
"""Count GGUF/safetensors files immediately inside *directory*.
|
|
|
|
Surfaces a count-hint so the UI can mark a weights-only leaf dir as a
|
|
valid "Use this folder" target. Bounded by *visited entries* (stops
|
|
after ``cap``), so the hint never costs more than a bounded walk.
|
|
"""
|
|
n = 0
|
|
visited = 0
|
|
try:
|
|
for f in directory.iterdir():
|
|
visited += 1
|
|
if visited > cap:
|
|
break
|
|
try:
|
|
if f.is_file():
|
|
low = f.name.lower()
|
|
if low.endswith((".gguf", ".safetensors")):
|
|
n += 1
|
|
except OSError:
|
|
continue
|
|
except PermissionError as e:
|
|
logger.debug("browse-folders: permission denied counting %s: %s", directory, e)
|
|
return 0
|
|
except OSError as e:
|
|
logger.debug("browse-folders: OS error counting %s: %s", directory, e)
|
|
return 0
|
|
return n
|
|
|
|
|
|
def _has_direct_model_signal(directory: Path) -> bool:
|
|
"""Return True if an immediate child signals a model: a
|
|
GGUF/safetensors/config.json file or a ``models--*`` subdir (HF
|
|
cache). Bounded by ``_BROWSE_MODEL_HINT_PROBE``."""
|
|
try:
|
|
it = directory.iterdir()
|
|
except OSError:
|
|
return False
|
|
try:
|
|
for i, child in enumerate(it):
|
|
if i >= _BROWSE_MODEL_HINT_PROBE:
|
|
break
|
|
try:
|
|
name = child.name
|
|
if child.is_file():
|
|
low = name.lower()
|
|
if low.endswith((".gguf", ".safetensors")):
|
|
return True
|
|
if low in ("config.json", "adapter_config.json"):
|
|
return True
|
|
elif child.is_dir() and name.startswith("models--"):
|
|
return True
|
|
except OSError:
|
|
continue
|
|
except OSError:
|
|
return False
|
|
return False
|
|
|
|
|
|
def _looks_like_model_dir(directory: Path) -> bool:
|
|
"""Bounded heuristic to flag dirs worth exploring in the browser.
|
|
|
|
False negatives are fine (the real scanner is authoritative). Three
|
|
signals, cheapest first: (1) name ``models--*`` (HF cache layout),
|
|
(2) an immediate child weight/config file, (3) a grandchild with a
|
|
direct signal (LM Studio / Ollama ``publisher/model`` layout, probing
|
|
the first ``_BROWSE_MODEL_HINT_PROBE`` child dirs).
|
|
"""
|
|
if directory.name.startswith("models--"):
|
|
return True
|
|
if _has_direct_model_signal(directory):
|
|
return True
|
|
# Grandchild probe: LM Studio / Ollama publisher/model layout.
|
|
try:
|
|
it = directory.iterdir()
|
|
except OSError:
|
|
return False
|
|
try:
|
|
for i, child in enumerate(it):
|
|
if i >= _BROWSE_MODEL_HINT_PROBE:
|
|
break
|
|
try:
|
|
if not child.is_dir():
|
|
continue
|
|
except OSError:
|
|
continue
|
|
if child.name.startswith("models--"):
|
|
return True
|
|
if _has_direct_model_signal(child):
|
|
return True
|
|
except OSError:
|
|
return False
|
|
return False
|
|
|
|
|
|
def _build_browse_allowlist() -> list[Path]:
|
|
"""Return the root directories the folder browser may walk.
|
|
|
|
The same list seeds the sidebar suggestion chips, so chip targets are
|
|
always reachable. Roots: HOME, resolved HF cache dirs, Studio's
|
|
outputs/exports/studio root, registered scan folders, and well-known
|
|
local-LLM dirs (LM Studio, Ollama, ``~/models``); each added only if
|
|
it resolves to a real directory.
|
|
"""
|
|
from utils.paths import (
|
|
hf_default_cache_dir,
|
|
legacy_hf_cache_dir,
|
|
well_known_model_dirs,
|
|
)
|
|
from storage.studio_db import list_scan_folders
|
|
|
|
candidates: list[Path] = []
|
|
|
|
def _add(p: Optional[Path]) -> None:
|
|
if p is None:
|
|
return
|
|
try:
|
|
resolved = p.resolve()
|
|
except OSError:
|
|
return
|
|
if _safe_is_dir(resolved):
|
|
candidates.append(resolved)
|
|
|
|
_add(Path.home())
|
|
_add(_resolve_hf_cache_dir())
|
|
try:
|
|
_add(hf_default_cache_dir())
|
|
except Exception: # noqa: BLE001 -- best-effort
|
|
pass
|
|
try:
|
|
_add(legacy_hf_cache_dir())
|
|
except Exception: # noqa: BLE001 -- best-effort
|
|
pass
|
|
try:
|
|
from utils.paths import (
|
|
exports_root,
|
|
outputs_root,
|
|
studio_root,
|
|
)
|
|
|
|
_add(studio_root())
|
|
_add(outputs_root())
|
|
_add(exports_root())
|
|
except Exception as exc: # noqa: BLE001 -- best-effort
|
|
logger.debug("browse-folders: studio roots unavailable: %s", exc)
|
|
try:
|
|
for folder in list_scan_folders():
|
|
p = folder.get("path")
|
|
if p:
|
|
_add(Path(p))
|
|
except Exception as exc: # noqa: BLE001 -- best-effort
|
|
logger.debug("browse-folders: could not load scan folders: %s", exc)
|
|
try:
|
|
for p in well_known_model_dirs():
|
|
_add(p)
|
|
except Exception as exc: # noqa: BLE001 -- best-effort
|
|
logger.debug("browse-folders: well-known dirs unavailable: %s", exc)
|
|
|
|
# Dedupe while preserving order.
|
|
seen: set[str] = set()
|
|
deduped: list[Path] = []
|
|
for p in candidates:
|
|
key = str(p)
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
deduped.append(p)
|
|
return deduped
|
|
|
|
|
|
def _is_path_inside_allowlist(target: Path, allowed_roots: list[Path]) -> bool:
|
|
"""True if *target* equals or descends from any allowed root.
|
|
|
|
Uses ``os.path.realpath`` so symlinks can't escape the sandbox.
|
|
"""
|
|
try:
|
|
target_real = os.path.realpath(str(target))
|
|
except OSError:
|
|
return False
|
|
for root in allowed_roots:
|
|
try:
|
|
root_real = os.path.realpath(str(root))
|
|
except OSError:
|
|
continue
|
|
if target_real == root_real or target_real.startswith(root_real + os.sep):
|
|
return True
|
|
return False
|
|
|
|
|
|
def _normalize_browse_request_path(path: Optional[str]) -> str:
|
|
"""Normalize the browse request path lexically, without touching the FS."""
|
|
if path is None or not path.strip():
|
|
return os.path.normpath(str(Path.home()))
|
|
|
|
expanded = os.path.expanduser(path.strip())
|
|
if not os.path.isabs(expanded):
|
|
expanded = os.path.join(str(Path.cwd()), expanded)
|
|
return os.path.normpath(expanded)
|
|
|
|
|
|
def _browse_relative_parts(requested_path: str, root: Path) -> Optional[list[str]]:
|
|
"""Return validated relative path components under ``root``."""
|
|
root_text = os.path.normpath(str(root))
|
|
try:
|
|
rel_text = os.path.relpath(requested_path, root_text)
|
|
except ValueError:
|
|
return None
|
|
|
|
if rel_text == ".":
|
|
return []
|
|
if rel_text == ".." or rel_text.startswith(f"..{os.sep}"):
|
|
return None
|
|
|
|
parts = [part for part in rel_text.split(os.sep) if part not in ("", ".")]
|
|
altsep = os.altsep
|
|
for part in parts:
|
|
if part == ".." or os.sep in part or (altsep and altsep in part):
|
|
return None
|
|
return parts
|
|
|
|
|
|
def _match_browse_child(current: Path, name: str) -> Optional[Path]:
|
|
"""Return the immediate child named ``name`` under ``current``."""
|
|
try:
|
|
for child in current.iterdir():
|
|
if child.name == name:
|
|
return child
|
|
except PermissionError:
|
|
raise HTTPException(
|
|
status_code = 403,
|
|
detail = f"Permission denied reading {current.name}",
|
|
) from None
|
|
except OSError as exc:
|
|
logger.warning("browse-folders: could not read %s: %s", current, exc, exc_info = True)
|
|
raise HTTPException(
|
|
status_code = 500,
|
|
detail = f"Could not read {os.path.basename(str(current))}",
|
|
) from exc
|
|
return None
|
|
|
|
|
|
def _resolve_browse_target(path: Optional[str], allowed_roots: list[Path]) -> Path:
|
|
"""Resolve a requested browse path by walking from trusted allowlist roots."""
|
|
requested_path = _normalize_browse_request_path(path)
|
|
resolved_roots: list[Path] = []
|
|
seen_roots: set[str] = set()
|
|
for root in sorted(allowed_roots, key = lambda p: len(str(p)), reverse = True):
|
|
try:
|
|
resolved = root.resolve()
|
|
except OSError:
|
|
continue
|
|
key = str(resolved)
|
|
if key in seen_roots:
|
|
continue
|
|
seen_roots.add(key)
|
|
resolved_roots.append(resolved)
|
|
|
|
for root in resolved_roots:
|
|
parts = _browse_relative_parts(requested_path, root)
|
|
if parts is None:
|
|
continue
|
|
|
|
current = root
|
|
for part in parts:
|
|
child = _match_browse_child(current, part)
|
|
if child is None:
|
|
raise HTTPException(
|
|
status_code = 404,
|
|
detail = f"Path does not exist: {os.path.basename(requested_path)}",
|
|
)
|
|
try:
|
|
resolved_child = child.resolve()
|
|
except OSError as exc:
|
|
logger.warning(
|
|
"browse-folders: invalid path component %r under %s: %s",
|
|
part,
|
|
current,
|
|
exc,
|
|
exc_info = True,
|
|
)
|
|
raise HTTPException(
|
|
status_code = 400,
|
|
detail = "Invalid path",
|
|
) from exc
|
|
if not _is_path_inside_allowlist(resolved_child, resolved_roots):
|
|
raise HTTPException(
|
|
status_code = 403,
|
|
detail = (
|
|
"Path is not in the browseable allowlist. Register it via "
|
|
"POST /api/models/scan-folders first, or pick a directory "
|
|
"under your home folder."
|
|
),
|
|
)
|
|
current = resolved_child
|
|
|
|
if not current.is_dir():
|
|
raise HTTPException(
|
|
status_code = 400,
|
|
detail = f"Not a directory: {os.path.basename(str(current))}",
|
|
)
|
|
return current
|
|
|
|
raise HTTPException(
|
|
status_code = 403,
|
|
detail = (
|
|
"Path is not in the browseable allowlist. Register it via "
|
|
"POST /api/models/scan-folders first, or pick a directory "
|
|
"under your home folder."
|
|
),
|
|
)
|
|
|
|
|
|
@router.get("/browse-folders", response_model = BrowseFoldersResponse)
|
|
async def browse_folders(
|
|
path: Optional[str] = Query(
|
|
None,
|
|
description = (
|
|
"Directory to list. If omitted, defaults to the current user's "
|
|
"home directory. Tilde (`~`) and relative paths are expanded. "
|
|
"Must resolve inside the allowlist of browseable roots (HOME, "
|
|
"HF cache, Studio dirs, registered scan folders, well-known "
|
|
"model dirs)."
|
|
),
|
|
),
|
|
show_hidden: bool = Query(
|
|
False,
|
|
description = "Include entries whose name starts with a dot",
|
|
),
|
|
current_subject: str = Depends(get_current_subject),
|
|
):
|
|
"""List immediate subdirectories of *path* for the Custom Folders picker.
|
|
|
|
Lets the frontend render a modal folder browser without a native OS
|
|
dialog. Read-only: enumerates visible subdirectories so the user can
|
|
click to a folder and hand the string to POST /api/models/scan-folders.
|
|
|
|
Sandbox: bounded to :func:`_build_browse_allowlist`; paths outside it
|
|
return 403, and symlinks are resolved via ``os.path.realpath`` first
|
|
so traversal can't escape. Sorting: model-bearing dirs, then plain,
|
|
then hidden (if ``show_hidden=true``).
|
|
"""
|
|
from utils.paths import hf_default_cache_dir, well_known_model_dirs
|
|
from storage.studio_db import list_scan_folders
|
|
|
|
# Build once; the sandbox check and suggestion chips share it.
|
|
allowed_roots = _build_browse_allowlist()
|
|
|
|
try:
|
|
target = _resolve_browse_target(path, allowed_roots)
|
|
except HTTPException:
|
|
requested_path = _normalize_browse_request_path(path)
|
|
if path is not None and path.strip():
|
|
logger.warning(
|
|
"browse-folders: rejected path %r (normalized=%s)",
|
|
path,
|
|
requested_path,
|
|
)
|
|
raise
|
|
|
|
# Enumerate immediate subdirectories with a bounded cap.
|
|
entries: list[BrowseEntry] = []
|
|
truncated = False
|
|
visited = 0
|
|
try:
|
|
it = target.iterdir()
|
|
except PermissionError:
|
|
raise HTTPException(
|
|
status_code = 403,
|
|
detail = f"Permission denied reading {os.path.basename(str(target))}",
|
|
)
|
|
except OSError as exc:
|
|
logger.warning("browse-folders: could not read %s: %s", target, exc, exc_info = True)
|
|
raise HTTPException(
|
|
status_code = 500,
|
|
detail = f"Could not read {os.path.basename(str(target))}",
|
|
)
|
|
|
|
try:
|
|
for child in it:
|
|
# Bound by *visited*, not *appended*: a cap on len(entries)
|
|
# would never trigger in dirs full of files. Counting visits
|
|
# caps worst-case work at ``_BROWSE_ENTRY_CAP`` calls.
|
|
visited += 1
|
|
if visited > _BROWSE_ENTRY_CAP:
|
|
truncated = True
|
|
break
|
|
try:
|
|
if not child.is_dir():
|
|
continue
|
|
except OSError:
|
|
continue
|
|
name = child.name
|
|
is_hidden = name.startswith(".")
|
|
if is_hidden and not show_hidden:
|
|
continue
|
|
entries.append(
|
|
BrowseEntry(
|
|
name = name,
|
|
has_models = _looks_like_model_dir(child),
|
|
hidden = is_hidden,
|
|
)
|
|
)
|
|
except PermissionError as exc:
|
|
logger.debug(
|
|
"browse-folders: permission denied during enumeration of %s: %s",
|
|
target,
|
|
exc,
|
|
)
|
|
except OSError as exc:
|
|
# Rare: iterdir succeeded but reading an entry failed.
|
|
logger.warning("browse-folders: partial enumeration of %s: %s", target, exc)
|
|
|
|
# Model-bearing first, then plain, then hidden; case-insensitive
|
|
# alphabetical within each bucket.
|
|
def _sort_key(e: BrowseEntry) -> tuple[int, str]:
|
|
bucket = 0 if e.has_models else (2 if e.hidden else 1)
|
|
return (bucket, e.name.lower())
|
|
|
|
entries.sort(key = _sort_key)
|
|
|
|
# Parent is None at the filesystem root and when it would leave the
|
|
# sandbox (else the up-row would 403 on click); users can still hop
|
|
# to other allowed roots via the suggestion chips.
|
|
parent: Optional[str]
|
|
if target.parent == target or not _is_path_inside_allowlist(target.parent, allowed_roots):
|
|
parent = None
|
|
else:
|
|
parent = str(target.parent)
|
|
|
|
# Handy starting points for the quick-pick chips.
|
|
suggestions: list[str] = []
|
|
seen_sug: set[str] = set()
|
|
|
|
def _add_sug(p: Optional[Path]) -> None:
|
|
if p is None:
|
|
return
|
|
try:
|
|
resolved = str(p.resolve())
|
|
except OSError:
|
|
return
|
|
if resolved in seen_sug:
|
|
return
|
|
if _safe_is_dir(resolved):
|
|
seen_sug.add(resolved)
|
|
suggestions.append(resolved)
|
|
|
|
# Home first -- the safe fallback when everything else is cold.
|
|
_add_sug(Path.home())
|
|
# The HF cache root the process is actually using.
|
|
try:
|
|
_add_sug(hf_default_cache_dir())
|
|
except Exception:
|
|
pass
|
|
# Already-registered scan folders (user-curated).
|
|
try:
|
|
for folder in list_scan_folders():
|
|
_add_sug(Path(folder.get("path", "")))
|
|
except Exception as exc:
|
|
logger.debug("browse-folders: could not load scan folders: %s", exc)
|
|
# Dirs used by other local-LLM tools (LM Studio, Ollama, ~/models);
|
|
# the helper returns only existing paths, so no dead chips.
|
|
try:
|
|
for p in well_known_model_dirs():
|
|
_add_sug(p)
|
|
except Exception as exc:
|
|
logger.debug("browse-folders: could not load well-known dirs: %s", exc)
|
|
|
|
return BrowseFoldersResponse(
|
|
current = str(target),
|
|
parent = parent,
|
|
entries = entries,
|
|
suggestions = suggestions,
|
|
truncated = truncated,
|
|
model_files_here = _count_model_files(target),
|
|
)
|
|
|
|
|
|
def _looks_like_mlx_repo(model_id: str) -> bool:
|
|
"""Name heuristic for unloaded models (mirrors the -GGUF suffix check);
|
|
tokenized so MLX only matches as a whole name segment."""
|
|
if model_id.lower().startswith("mlx-community/"):
|
|
return True
|
|
tail = model_id.split("/")[-1]
|
|
return "MLX" in _re.split(r"[-_.]", tail.upper())
|
|
|
|
|
|
@router.get("/list")
|
|
async def list_models(current_subject: str = Depends(get_current_subject)):
|
|
"""List available models: default plus currently loaded."""
|
|
try:
|
|
inference_backend = get_inference_backend()
|
|
|
|
default_models = inference_backend.default_models
|
|
|
|
loaded_models = []
|
|
for model_name, model_data in inference_backend.models.items():
|
|
_is_vision = model_data.get("is_vision", False)
|
|
_audio_type = model_data.get("audio_type")
|
|
model_info = ModelDetails(
|
|
id = model_name,
|
|
name = model_name.split("/")[-1] if "/" in model_name else model_name,
|
|
is_vision = _is_vision,
|
|
is_lora = model_data.get("is_lora", False),
|
|
is_mlx = model_data.get("is_mlx", False),
|
|
is_audio = model_data.get("is_audio", False),
|
|
audio_type = _audio_type,
|
|
has_audio_input = model_data.get("has_audio_input", False),
|
|
model_type = derive_model_type(_is_vision, _audio_type),
|
|
)
|
|
loaded_models.append(model_info)
|
|
|
|
# Include active GGUF model (loaded via llama-server).
|
|
from routes.inference import get_llama_cpp_backend
|
|
|
|
llama_backend = get_llama_cpp_backend()
|
|
if llama_backend.is_loaded and llama_backend.model_identifier:
|
|
loaded_models.append(
|
|
ModelDetails(
|
|
id = llama_backend.model_identifier,
|
|
name = llama_backend.model_identifier.split("/")[-1],
|
|
is_gguf = True,
|
|
is_vision = llama_backend.is_vision,
|
|
is_audio = getattr(llama_backend, "_is_audio", False),
|
|
audio_type = getattr(llama_backend, "_audio_type", None),
|
|
)
|
|
)
|
|
|
|
# Combine default and loaded; prefer loaded entries for duplicate
|
|
# ids so runtime flags survive.
|
|
all_models = []
|
|
seen_ids = set()
|
|
loaded_by_id = {model_info.id: model_info for model_info in loaded_models}
|
|
|
|
for model_id in default_models:
|
|
if model_id not in seen_ids:
|
|
model_info = loaded_by_id.get(model_id) or ModelDetails(
|
|
id = model_id,
|
|
name = model_id.split("/")[-1] if "/" in model_id else model_id,
|
|
is_gguf = model_id.upper().endswith("-GGUF"),
|
|
is_mlx = _looks_like_mlx_repo(model_id),
|
|
)
|
|
all_models.append(model_info)
|
|
seen_ids.add(model_id)
|
|
|
|
for model_info in loaded_models:
|
|
if model_info.id not in seen_ids:
|
|
all_models.append(model_info)
|
|
seen_ids.add(model_info.id)
|
|
|
|
return ModelListResponse(models = all_models, default_models = default_models)
|
|
|
|
except Exception as e:
|
|
raise log_and_http_error(
|
|
e,
|
|
500,
|
|
"Failed to list models",
|
|
event = "models.list_models_failed",
|
|
log = logger,
|
|
)
|
|
|
|
|
|
def _get_max_position_embeddings(config) -> Optional[int]:
|
|
"""Extract max_position_embeddings from a config, with text_config fallback."""
|
|
if hasattr(config, "max_position_embeddings"):
|
|
return config.max_position_embeddings
|
|
if hasattr(config, "text_config") and hasattr(config.text_config, "max_position_embeddings"):
|
|
return config.text_config.max_position_embeddings
|
|
return None
|
|
|
|
|
|
def _get_model_size_bytes(model_name: str, hf_token: Optional[str] = None) -> Optional[int]:
|
|
"""Total size of model weight files from HF Hub."""
|
|
try:
|
|
from huggingface_hub import HfApi
|
|
|
|
api = HfApi(token = hf_token)
|
|
info = api.repo_info(model_name, repo_type = "model", token = hf_token)
|
|
if not info.siblings:
|
|
return None
|
|
|
|
weight_exts = (".safetensors", ".bin", ".pt", ".pth", ".gguf")
|
|
total = 0
|
|
for sibling in info.siblings:
|
|
if sibling.rfilename and any(sibling.rfilename.endswith(ext) for ext in weight_exts):
|
|
if sibling.size is not None:
|
|
total += sibling.size
|
|
|
|
return total if total > 0 else None
|
|
except Exception as e:
|
|
logger.warning(f"Could not get model size for {model_name}: {e}")
|
|
return None
|
|
|
|
|
|
@router.get("/config/{model_name:path}")
|
|
async def get_model_config(
|
|
model_name: str,
|
|
hf_token: Optional[str] = Query(None),
|
|
current_subject: str = Depends(get_current_subject),
|
|
):
|
|
"""Get configuration for a specific model (wraps load_model_defaults)."""
|
|
try:
|
|
if not is_local_path(model_name):
|
|
resolved = resolve_cached_repo_id_case(model_name)
|
|
if resolved != model_name:
|
|
logger.info(
|
|
"Using cached repo_id casing '%s' for requested '%s'",
|
|
resolved,
|
|
model_name,
|
|
)
|
|
model_name = resolved
|
|
|
|
logger.info(f"Getting model config for: {model_name}")
|
|
from utils.models.model_config import detect_audio_type
|
|
|
|
config_dict = load_model_defaults(model_name)
|
|
|
|
# Detect capabilities (HF token for gated models).
|
|
is_vision = is_vision_model(model_name, hf_token = hf_token)
|
|
is_embedding = is_embedding_model(model_name, hf_token = hf_token)
|
|
audio_type = detect_audio_type(model_name, hf_token = hf_token)
|
|
|
|
is_lora = False
|
|
base_model = None
|
|
max_position_embeddings = None
|
|
try:
|
|
model_config = ModelConfig.from_identifier(model_name)
|
|
is_lora = model_config.is_lora
|
|
base_model = model_config.base_model if is_lora else None
|
|
max_position_embeddings = _get_max_position_embeddings(model_config)
|
|
except Exception:
|
|
pass
|
|
|
|
# Fallback: try AutoConfig directly.
|
|
if max_position_embeddings is None:
|
|
try:
|
|
from transformers import AutoConfig as _AutoConfig
|
|
|
|
_trust = model_name.lower().startswith("unsloth/")
|
|
_ac = _AutoConfig.from_pretrained(
|
|
model_name, trust_remote_code = _trust, token = hf_token
|
|
)
|
|
max_position_embeddings = _get_max_position_embeddings(_ac)
|
|
except Exception:
|
|
pass
|
|
|
|
logger.info(
|
|
f"Model config result for {model_name}: is_vision={is_vision}, is_embedding={is_embedding}, audio_type={audio_type}, is_lora={is_lora}, max_position_embeddings={max_position_embeddings}"
|
|
)
|
|
return ModelDetails(
|
|
id = model_name,
|
|
model_name = model_name,
|
|
config = config_dict,
|
|
is_vision = is_vision,
|
|
is_embedding = is_embedding,
|
|
is_lora = is_lora,
|
|
is_audio = audio_type is not None,
|
|
audio_type = audio_type,
|
|
has_audio_input = is_audio_input_type(audio_type),
|
|
model_type = derive_model_type(is_vision, audio_type, is_embedding),
|
|
base_model = base_model,
|
|
max_position_embeddings = max_position_embeddings,
|
|
model_size_bytes = _get_model_size_bytes(model_name, hf_token),
|
|
)
|
|
|
|
except Exception as e:
|
|
raise log_and_http_error(
|
|
e,
|
|
500,
|
|
"Failed to get model config",
|
|
event = "models.get_model_config_failed",
|
|
log = logger,
|
|
)
|
|
|
|
|
|
@router.get("/loras")
|
|
async def scan_loras(
|
|
outputs_dir: str = Query(
|
|
default = str(outputs_root()), description = "Directory to scan for LoRA adapters"
|
|
),
|
|
exports_dir: str = Query(
|
|
default = str(exports_root()), description = "Directory to scan for exported models"
|
|
),
|
|
current_subject: str = Depends(get_current_subject),
|
|
):
|
|
"""Scan for trained LoRA adapters and exported models.
|
|
|
|
Returns training outputs (outputs_dir) and exported models
|
|
(exports_dir) in one list, distinguished by the source field.
|
|
"""
|
|
try:
|
|
resolved_outputs_dir = str(resolve_output_dir(outputs_dir))
|
|
resolved_exports_dir = str(resolve_export_dir(exports_dir))
|
|
lora_list = []
|
|
|
|
trained_models = scan_trained_models(outputs_dir = resolved_outputs_dir)
|
|
for display_name, model_path, model_type in trained_models:
|
|
base_model = get_base_model_from_checkpoint(model_path)
|
|
lora_list.append(
|
|
LoRAInfo(
|
|
display_name = display_name,
|
|
adapter_path = model_path,
|
|
base_model = base_model,
|
|
source = "training",
|
|
export_type = model_type,
|
|
)
|
|
)
|
|
|
|
# Scan exported models (merged, LoRA, base — skips GGUF)
|
|
exported = scan_exported_models(exports_dir = resolved_exports_dir)
|
|
for display_name, model_path, export_type, base_model in exported:
|
|
lora_list.append(
|
|
LoRAInfo(
|
|
display_name = display_name,
|
|
adapter_path = model_path,
|
|
base_model = base_model,
|
|
source = "exported",
|
|
export_type = export_type,
|
|
)
|
|
)
|
|
|
|
return LoRAScanResponse(loras = lora_list, outputs_dir = resolved_outputs_dir)
|
|
|
|
except Exception as e:
|
|
raise log_and_http_error(
|
|
e,
|
|
500,
|
|
"Failed to scan LoRA adapters",
|
|
event = "models.scan_loras_failed",
|
|
log = logger,
|
|
)
|
|
|
|
|
|
def _is_path_under(path: Path, root: Path) -> bool:
|
|
try:
|
|
path.resolve().relative_to(root.resolve())
|
|
return True
|
|
except ValueError:
|
|
return False
|
|
|
|
|
|
def _is_path_under_lexically(path: Path, root: Path) -> bool:
|
|
"""Check containment without resolving the final path's symlink."""
|
|
try:
|
|
absolute_path = Path(os.path.abspath(str(path)))
|
|
absolute_root = Path(os.path.abspath(str(root)))
|
|
absolute_path.relative_to(absolute_root)
|
|
return True
|
|
except ValueError:
|
|
return False
|
|
|
|
|
|
def _loaded_model_matches_deleted_path(active_model: str, deleted_path: Path) -> bool:
|
|
try:
|
|
active = Path(active_model).expanduser().resolve()
|
|
target = deleted_path.resolve()
|
|
return active == target or (target.is_dir() and active.is_relative_to(target))
|
|
except (OSError, RuntimeError, ValueError) as e:
|
|
logger.debug(
|
|
"Could not resolve loaded/deleted model paths; falling back to string comparison: %s",
|
|
e,
|
|
)
|
|
active_lower = active_model.lower()
|
|
target_lower = str(deleted_path).lower()
|
|
return active_lower == target_lower or active_lower.startswith(f"{target_lower}{os.sep}")
|
|
|
|
|
|
def _loading_model_matches_deleted_path(loading_model: object, deleted_path: Path) -> bool:
|
|
if not loading_model:
|
|
return False
|
|
return _loaded_model_matches_deleted_path(str(loading_model), deleted_path)
|
|
|
|
|
|
def _prune_empty_parents(start: Path, stop_at: Path) -> None:
|
|
"""Remove empty ancestors of ``start`` up to (not including) ``stop_at``.
|
|
|
|
Used after deleting a checkpoint so the enclosing run dir doesn't
|
|
linger as an empty entry in scan results.
|
|
"""
|
|
try:
|
|
stop_resolved = stop_at.resolve()
|
|
except OSError:
|
|
return
|
|
parent = start.parent
|
|
while True:
|
|
try:
|
|
parent_resolved = parent.resolve()
|
|
except OSError:
|
|
return
|
|
if parent_resolved == stop_resolved:
|
|
return
|
|
try:
|
|
parent_resolved.relative_to(stop_resolved)
|
|
except ValueError:
|
|
return
|
|
try:
|
|
parent.rmdir()
|
|
except OSError:
|
|
return
|
|
parent = parent.parent
|
|
|
|
|
|
def _delete_gguf_variant_files(root: Path, variant: str) -> tuple[int, int]:
|
|
deleted_count = 0
|
|
deleted_bytes = 0
|
|
for path in root.rglob("*"):
|
|
if not path.is_file() or not _is_main_gguf_filename(path.name):
|
|
continue
|
|
if _extract_quant_label(path.name).lower() != variant.lower():
|
|
continue
|
|
try:
|
|
deleted_bytes += path.stat().st_size
|
|
except OSError:
|
|
pass
|
|
path.unlink()
|
|
deleted_count += 1
|
|
return deleted_count, deleted_bytes
|
|
|
|
|
|
@router.delete("/delete-finetuned")
|
|
async def delete_finetuned_model(
|
|
model_path: str = Body(...),
|
|
source: str = Body(...),
|
|
export_type: Optional[str] = Body(None),
|
|
gguf_variant: Optional[str] = Body(None),
|
|
current_subject: str = Depends(get_current_subject),
|
|
):
|
|
"""Delete a Studio-trained or exported model from disk.
|
|
|
|
Only paths under Studio's outputs/exports roots are accepted.
|
|
Exported GGUF entries can delete one quant variant at a time.
|
|
"""
|
|
if source not in {"training", "exported"}:
|
|
raise HTTPException(
|
|
status_code = 400,
|
|
detail = "Only trained or exported Studio models can be deleted",
|
|
)
|
|
|
|
if not model_path or not model_path.strip():
|
|
raise HTTPException(status_code = 400, detail = "model_path is required")
|
|
|
|
if export_type == "gguf" and not gguf_variant:
|
|
raise HTTPException(
|
|
status_code = 400,
|
|
detail = "gguf_variant is required when export_type is 'gguf'",
|
|
)
|
|
|
|
raw_path = Path(model_path).expanduser()
|
|
if source == "training":
|
|
target_path = raw_path
|
|
allowed_root = outputs_root()
|
|
else:
|
|
allowed_root = exports_root()
|
|
target_path = (
|
|
raw_path.parent
|
|
if export_type == "gguf" and raw_path.suffix.lower() == ".gguf"
|
|
else raw_path
|
|
)
|
|
|
|
allowed_root = allowed_root.resolve()
|
|
delete_path = Path(os.path.abspath(str(target_path)))
|
|
delete_path_is_symlink = delete_path.is_symlink()
|
|
|
|
if delete_path_is_symlink:
|
|
if not _is_path_under_lexically(delete_path, allowed_root):
|
|
raise HTTPException(
|
|
status_code = 400,
|
|
detail = "Model path is outside Studio storage",
|
|
)
|
|
if export_type == "gguf" and gguf_variant:
|
|
target_path = delete_path.resolve()
|
|
if not _is_path_under(target_path, allowed_root):
|
|
raise HTTPException(
|
|
status_code = 400,
|
|
detail = "Model path is outside Studio storage",
|
|
)
|
|
else:
|
|
target_path = delete_path
|
|
else:
|
|
target_path = target_path.resolve()
|
|
|
|
should_check_resolved_path = not delete_path_is_symlink or (
|
|
export_type == "gguf" and gguf_variant
|
|
)
|
|
if should_check_resolved_path and not _is_path_under(target_path, allowed_root):
|
|
raise HTTPException(
|
|
status_code = 400,
|
|
detail = "Model path is outside Studio storage",
|
|
)
|
|
if target_path == allowed_root:
|
|
raise HTTPException(
|
|
status_code = 400,
|
|
detail = "Refusing to delete storage root",
|
|
)
|
|
if not target_path.exists() and not target_path.is_symlink():
|
|
raise HTTPException(status_code = 404, detail = "Model not found on disk")
|
|
|
|
if source == "training":
|
|
try:
|
|
from core.training import get_training_backend
|
|
training_backend = get_training_backend()
|
|
if training_backend.is_training_active():
|
|
raise HTTPException(
|
|
status_code = 409,
|
|
detail = "Cannot delete trained models while training is running",
|
|
)
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.warning("Could not check training status before delete: %s", e)
|
|
raise HTTPException(
|
|
status_code = 500,
|
|
detail = "Could not verify training status before deleting",
|
|
) from e
|
|
|
|
try:
|
|
from routes.inference import get_llama_cpp_backend
|
|
|
|
llama_backend = get_llama_cpp_backend()
|
|
if (
|
|
llama_backend.is_active
|
|
and not llama_backend.is_loaded
|
|
and llama_backend.model_identifier
|
|
and _loaded_model_matches_deleted_path(
|
|
llama_backend.model_identifier,
|
|
target_path,
|
|
)
|
|
and (
|
|
not gguf_variant
|
|
or not llama_backend.hf_variant
|
|
or llama_backend.hf_variant.lower() == gguf_variant.lower()
|
|
)
|
|
):
|
|
raise HTTPException(
|
|
status_code = 409,
|
|
detail = "Cannot delete a model while it is loading",
|
|
)
|
|
if (
|
|
llama_backend.is_loaded
|
|
and llama_backend.model_identifier
|
|
and _loaded_model_matches_deleted_path(
|
|
llama_backend.model_identifier,
|
|
target_path,
|
|
)
|
|
and (
|
|
not gguf_variant
|
|
or not llama_backend.hf_variant
|
|
or llama_backend.hf_variant.lower() == gguf_variant.lower()
|
|
)
|
|
):
|
|
raise HTTPException(
|
|
status_code = 400,
|
|
detail = "Unload the model before deleting",
|
|
)
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.warning("Could not check llama.cpp loaded model before delete: %s", e)
|
|
raise HTTPException(
|
|
status_code = 503,
|
|
detail = "Could not verify model load status before deleting",
|
|
) from e
|
|
|
|
try:
|
|
inference_backend = get_inference_backend()
|
|
loading_models = getattr(inference_backend, "loading_models", set())
|
|
if any(
|
|
_loading_model_matches_deleted_path(loading_model, target_path)
|
|
for loading_model in loading_models
|
|
):
|
|
raise HTTPException(
|
|
status_code = 409,
|
|
detail = "Cannot delete a model while it is loading",
|
|
)
|
|
if inference_backend.active_model_name:
|
|
if _loaded_model_matches_deleted_path(
|
|
inference_backend.active_model_name,
|
|
target_path,
|
|
):
|
|
raise HTTPException(
|
|
status_code = 400,
|
|
detail = "Unload the model before deleting",
|
|
)
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.warning("Could not check inference backend loaded model before delete: %s", e)
|
|
raise HTTPException(
|
|
status_code = 503,
|
|
detail = "Could not verify model load status before deleting",
|
|
) from e
|
|
|
|
try:
|
|
if export_type == "gguf" and gguf_variant:
|
|
if not target_path.is_dir():
|
|
raise HTTPException(
|
|
status_code = 400,
|
|
detail = "GGUF variant deletion requires an export directory",
|
|
)
|
|
deleted_count, deleted_bytes = _delete_gguf_variant_files(
|
|
target_path,
|
|
gguf_variant,
|
|
)
|
|
if deleted_count == 0:
|
|
raise HTTPException(
|
|
status_code = 404,
|
|
detail = f"Variant {gguf_variant} not found on disk",
|
|
)
|
|
try:
|
|
if not any(target_path.iterdir()):
|
|
target_path.rmdir()
|
|
_prune_empty_parents(target_path, allowed_root)
|
|
except OSError:
|
|
pass
|
|
logger.info(
|
|
"Deleted %s GGUF file(s) for exported model at %s variant %s (%0.1f MB freed)",
|
|
deleted_count,
|
|
target_path,
|
|
gguf_variant,
|
|
deleted_bytes / (1024 * 1024),
|
|
)
|
|
return {
|
|
"status": "deleted",
|
|
"path": str(target_path),
|
|
"gguf_variant": gguf_variant,
|
|
}
|
|
|
|
if target_path.is_symlink() or target_path.is_file():
|
|
target_path.unlink()
|
|
else:
|
|
shutil.rmtree(target_path)
|
|
|
|
if target_path.exists() or target_path.is_symlink():
|
|
raise HTTPException(
|
|
status_code = 500,
|
|
detail = "Deletion incomplete; some files could not be removed",
|
|
)
|
|
|
|
_prune_empty_parents(target_path, allowed_root)
|
|
|
|
logger.info("Deleted fine-tuned model at %s", target_path)
|
|
return {"status": "deleted", "path": str(target_path)}
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error(
|
|
"Error deleting fine-tuned model %s: %s",
|
|
target_path,
|
|
e,
|
|
exc_info = True,
|
|
)
|
|
raise HTTPException(
|
|
status_code = 500,
|
|
detail = "Failed to delete fine-tuned model",
|
|
)
|
|
|
|
|
|
@router.get("/loras/{lora_path:path}/base-model", response_model = LoRABaseModelResponse)
|
|
async def get_lora_base_model(lora_path: str, current_subject: str = Depends(get_current_subject)):
|
|
"""
|
|
Get the base model for a LoRA adapter.
|
|
|
|
This endpoint wraps the backend get_base_model_from_lora function.
|
|
"""
|
|
try:
|
|
base_model = get_base_model_from_lora(lora_path)
|
|
|
|
if base_model is None:
|
|
raise HTTPException(
|
|
status_code = 404,
|
|
detail = f"Could not determine base model for LoRA: {lora_path}",
|
|
)
|
|
|
|
return LoRABaseModelResponse(
|
|
lora_path = lora_path,
|
|
base_model = base_model,
|
|
)
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
raise log_and_http_error(
|
|
e,
|
|
500,
|
|
"Failed to get base model",
|
|
event = "models.get_lora_base_model_failed",
|
|
log = logger,
|
|
)
|
|
|
|
|
|
@router.get("/check-vision/{model_name:path}", response_model = VisionCheckResponse)
|
|
async def check_vision_model(model_name: str, current_subject: str = Depends(get_current_subject)):
|
|
"""
|
|
Check if a model is a vision model.
|
|
|
|
This endpoint wraps the backend is_vision_model function.
|
|
"""
|
|
try:
|
|
logger.info(f"Checking if vision model: {model_name}")
|
|
is_vision = is_vision_model(model_name)
|
|
|
|
logger.info(f"Vision check result for {model_name}: is_vision={is_vision}")
|
|
return VisionCheckResponse(
|
|
model_name = model_name,
|
|
is_vision = is_vision,
|
|
)
|
|
|
|
except Exception as e:
|
|
raise log_and_http_error(
|
|
e,
|
|
500,
|
|
"Failed to check vision model",
|
|
event = "models.check_vision_model_failed",
|
|
log = logger,
|
|
)
|
|
|
|
|
|
@router.get("/check-embedding/{model_name:path}", response_model = EmbeddingCheckResponse)
|
|
async def check_embedding_model(
|
|
model_name: str,
|
|
hf_token: Optional[str] = Query(None),
|
|
current_subject: str = Depends(get_current_subject),
|
|
):
|
|
"""
|
|
Check if a model is an embedding model.
|
|
|
|
This endpoint wraps the backend is_embedding_model function.
|
|
"""
|
|
try:
|
|
logger.info(f"Checking if embedding model: {model_name}")
|
|
is_embedding = is_embedding_model(model_name, hf_token = hf_token)
|
|
|
|
logger.info(f"Embedding check result for {model_name}: is_embedding={is_embedding}")
|
|
return EmbeddingCheckResponse(
|
|
model_name = model_name,
|
|
is_embedding = is_embedding,
|
|
)
|
|
|
|
except Exception as e:
|
|
raise log_and_http_error(
|
|
e,
|
|
500,
|
|
"Failed to check embedding model",
|
|
event = "models.check_embedding_model_failed",
|
|
log = logger,
|
|
)
|
|
|
|
|
|
@router.get("/gguf-variants", response_model = GgufVariantsResponse)
|
|
async def get_gguf_variants(
|
|
repo_id: str = Query(
|
|
..., description = "HuggingFace repo ID (e.g. 'unsloth/gemma-3-4b-it-GGUF')"
|
|
),
|
|
hf_token: Optional[str] = Query(None, description = "HuggingFace token for private repos"),
|
|
current_subject: str = Depends(get_current_subject),
|
|
):
|
|
"""List GGUF quantization variants for a HF repo or local directory.
|
|
|
|
Returns all variants with file sizes, vision support, and the
|
|
recommended default.
|
|
"""
|
|
try:
|
|
from utils.models.model_config import is_local_path, list_local_gguf_variants
|
|
|
|
# Local directory path — scan filesystem.
|
|
if is_local_path(repo_id):
|
|
variants, has_vision = list_local_gguf_variants(repo_id)
|
|
|
|
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 = repo_id,
|
|
variants = [
|
|
GgufVariantDetail(
|
|
filename = v.filename,
|
|
quant = v.quant,
|
|
size_bytes = v.size_bytes,
|
|
downloaded = True, # all local variants are downloaded
|
|
)
|
|
for v in variants
|
|
],
|
|
has_vision = has_vision,
|
|
default_variant = default_variant,
|
|
)
|
|
|
|
# Remote HuggingFace repo — query HF API.
|
|
variants, has_vision = list_gguf_variants(repo_id, hf_token = hf_token)
|
|
|
|
filenames = [v.filename for v in variants]
|
|
best = _pick_best_gguf(filenames)
|
|
default_variant = _extract_quant_label(best) if best else None
|
|
|
|
# Per-snapshot so a split GGUF's shards must all sit in one snapshot;
|
|
# mmproj adapters are excluded so they can't inflate a quant's bytes.
|
|
cached_bytes_by_quant_per_snapshot: list[dict[str, int]] = []
|
|
try:
|
|
from huggingface_hub import constants as hf_constants
|
|
|
|
if not _is_valid_repo_id(repo_id):
|
|
raise ValueError(f"Invalid repo_id format: {repo_id}")
|
|
|
|
cache_dir = Path(hf_constants.HF_HUB_CACHE)
|
|
target = f"models--{repo_id.replace('/', '--')}".lower()
|
|
for entry in cache_dir.iterdir():
|
|
if entry.name.lower() == target:
|
|
snapshots = entry / "snapshots"
|
|
if snapshots.is_dir():
|
|
for snap in snapshots.iterdir():
|
|
by_quant: dict[str, int] = {}
|
|
for f in _iter_gguf_paths(snap):
|
|
if _is_mmproj_filename(f.name):
|
|
continue
|
|
try:
|
|
size = f.stat().st_size
|
|
except OSError:
|
|
continue # broken symlink / unreadable: skip
|
|
q = _extract_quant_label(f.name).lower()
|
|
by_quant[q] = by_quant.get(q, 0) + size
|
|
if by_quant:
|
|
cached_bytes_by_quant_per_snapshot.append(by_quant)
|
|
break
|
|
except Exception:
|
|
pass
|
|
|
|
def _is_fully_downloaded(variant) -> bool:
|
|
if variant.size_bytes == 0:
|
|
return False
|
|
# Complete within one snapshot (tolerance for symlink size jitter).
|
|
quant = variant.quant.lower()
|
|
return any(
|
|
by_quant.get(quant, 0) >= variant.size_bytes * 0.99
|
|
for by_quant in cached_bytes_by_quant_per_snapshot
|
|
)
|
|
|
|
return GgufVariantsResponse(
|
|
repo_id = repo_id,
|
|
variants = [
|
|
GgufVariantDetail(
|
|
filename = v.filename,
|
|
quant = v.quant,
|
|
size_bytes = v.size_bytes,
|
|
downloaded = _is_fully_downloaded(v),
|
|
)
|
|
for v in variants
|
|
],
|
|
has_vision = has_vision,
|
|
default_variant = default_variant,
|
|
)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error listing GGUF variants for '{repo_id}': {e}", exc_info = True)
|
|
raise HTTPException(
|
|
status_code = 500,
|
|
detail = "Failed to list GGUF variants",
|
|
)
|
|
|
|
|
|
@router.get("/gguf-download-progress")
|
|
async def get_gguf_download_progress(
|
|
repo_id: str = Query(..., description = "HuggingFace repo ID"),
|
|
variant: str = Query("", description = "Quantization variant (e.g. UD-TQ1_0)"),
|
|
expected_bytes: int = Query(0, description = "Expected total download size in bytes"),
|
|
current_subject: str = Depends(get_current_subject),
|
|
):
|
|
"""Download progress from cached GGUF files for a specific variant.
|
|
|
|
Tracks completed shards in snapshots and in-progress (.incomplete)
|
|
downloads in the blobs directory.
|
|
"""
|
|
try:
|
|
if not _is_valid_repo_id(repo_id):
|
|
return {
|
|
"downloaded_bytes": 0,
|
|
"expected_bytes": expected_bytes,
|
|
"progress": 0,
|
|
}
|
|
|
|
from huggingface_hub import constants as hf_constants
|
|
|
|
cache_dir = Path(hf_constants.HF_HUB_CACHE)
|
|
target = f"models--{repo_id.replace('/', '--')}".lower()
|
|
variant_lower = variant.lower().replace("-", "").replace("_", "")
|
|
downloaded_bytes = 0
|
|
in_progress_bytes = 0
|
|
for entry in cache_dir.iterdir():
|
|
if entry.name.lower() == target:
|
|
# Completed .gguf files for this variant in snapshots.
|
|
# Exclude mmproj so a vision adapter can't satisfy a same-label
|
|
# main variant (e.g. mmproj-F16 vs an F16 weight).
|
|
for f in _iter_gguf_paths(entry):
|
|
if _is_mmproj_filename(f.name):
|
|
continue
|
|
fname = f.name.lower().replace("-", "").replace("_", "")
|
|
if not variant_lower or variant_lower in fname:
|
|
try:
|
|
downloaded_bytes += f.stat().st_size
|
|
except OSError:
|
|
continue # broken symlink / unreadable: skip
|
|
# In-progress (.incomplete) downloads in blobs.
|
|
blobs_dir = entry / "blobs"
|
|
if blobs_dir.is_dir():
|
|
for f in blobs_dir.iterdir():
|
|
if f.is_file() and f.name.endswith(".incomplete"):
|
|
try:
|
|
in_progress_bytes += f.stat().st_size
|
|
except OSError:
|
|
continue
|
|
break
|
|
|
|
total_progress_bytes = downloaded_bytes + in_progress_bytes
|
|
progress = min(total_progress_bytes / expected_bytes, 0.99) if expected_bytes > 0 else 0
|
|
# Report 1.0 only when all bytes are in completed files.
|
|
if expected_bytes > 0 and downloaded_bytes >= expected_bytes:
|
|
progress = 1.0
|
|
return {
|
|
"downloaded_bytes": total_progress_bytes,
|
|
"expected_bytes": expected_bytes,
|
|
"progress": round(progress, 3),
|
|
}
|
|
except Exception:
|
|
return {"downloaded_bytes": 0, "expected_bytes": expected_bytes, "progress": 0}
|
|
|
|
|
|
def _resolve_hf_cache_realpath(repo_dir: Path) -> Optional[str]:
|
|
"""Pick the most useful on-disk path for a HF cache repo.
|
|
|
|
Prefers the most-recent snapshot dir (what ``from_pretrained`` uses),
|
|
falling back to the cache repo root. Returns the resolved realpath so
|
|
snapshot symlinks follow back to blobs/.
|
|
"""
|
|
try:
|
|
snapshots_dir = repo_dir / "snapshots"
|
|
if snapshots_dir.is_dir():
|
|
snaps = [s for s in snapshots_dir.iterdir() if s.is_dir()]
|
|
if snaps:
|
|
latest = max(snaps, key = lambda s: s.stat().st_mtime)
|
|
return str(latest.resolve())
|
|
return str(repo_dir.resolve())
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
@router.get("/download-progress")
|
|
async def get_download_progress(
|
|
repo_id: str = Query(..., description = "HuggingFace repo ID"),
|
|
current_subject: str = Depends(get_current_subject),
|
|
):
|
|
"""Return download progress for any HuggingFace model repo.
|
|
|
|
Checks the local HF cache for completed blobs and in-progress
|
|
(.incomplete) downloads. Gets the expected total size from the HF API
|
|
on the first call, then caches it for later polls. Also returns
|
|
``cache_path``: the realpath of the snapshot dir (or cache repo root
|
|
if no snapshot yet) so the UI can show where weights live on disk.
|
|
"""
|
|
_empty = {
|
|
"downloaded_bytes": 0,
|
|
"expected_bytes": 0,
|
|
"progress": 0,
|
|
"cache_path": None,
|
|
}
|
|
try:
|
|
if not _is_valid_repo_id(repo_id):
|
|
return _empty
|
|
|
|
from huggingface_hub import constants as hf_constants
|
|
|
|
cache_dir = Path(hf_constants.HF_HUB_CACHE)
|
|
target = f"models--{repo_id.replace('/', '--')}".lower()
|
|
completed_bytes = 0
|
|
in_progress_bytes = 0
|
|
cache_path: Optional[str] = None
|
|
|
|
for entry in cache_dir.iterdir():
|
|
if entry.name.lower() != target:
|
|
continue
|
|
cache_path = _resolve_hf_cache_realpath(entry)
|
|
blobs_dir = entry / "blobs"
|
|
if not blobs_dir.is_dir():
|
|
break
|
|
for f in blobs_dir.iterdir():
|
|
if not f.is_file():
|
|
continue
|
|
if f.name.endswith(".incomplete"):
|
|
in_progress_bytes += f.stat().st_size
|
|
else:
|
|
completed_bytes += f.stat().st_size
|
|
break
|
|
|
|
downloaded_bytes = completed_bytes + in_progress_bytes
|
|
if downloaded_bytes == 0:
|
|
return {**_empty, "cache_path": cache_path}
|
|
|
|
expected_bytes = _get_repo_size_cached(repo_id)
|
|
if expected_bytes <= 0:
|
|
# Total unknown; report bytes only, no percentage.
|
|
return {
|
|
"downloaded_bytes": downloaded_bytes,
|
|
"expected_bytes": 0,
|
|
"progress": 0,
|
|
"cache_path": cache_path,
|
|
}
|
|
|
|
# 95% threshold (blob dedup can skew completed_bytes). Do NOT
|
|
# treat "no .incomplete files" as done: HF downloads sequentially,
|
|
# so none exist between files even when far from finished.
|
|
if completed_bytes >= expected_bytes * 0.95:
|
|
progress = 1.0
|
|
else:
|
|
progress = min(downloaded_bytes / expected_bytes, 0.99)
|
|
return {
|
|
"downloaded_bytes": downloaded_bytes,
|
|
"expected_bytes": expected_bytes,
|
|
"progress": round(progress, 3),
|
|
"cache_path": cache_path,
|
|
}
|
|
except Exception as e:
|
|
logger.warning(f"Error checking download progress for {repo_id}: {e}")
|
|
return _empty
|
|
|
|
|
|
_repo_size_cache: dict[str, int] = {}
|
|
|
|
|
|
def _get_repo_size_cached(repo_id: str) -> int:
|
|
if repo_id in _repo_size_cache:
|
|
return _repo_size_cache[repo_id]
|
|
try:
|
|
from huggingface_hub import model_info as hf_model_info
|
|
|
|
info = hf_model_info(repo_id, token = None, files_metadata = True)
|
|
total = sum(s.size for s in info.siblings if s.size)
|
|
_repo_size_cache[repo_id] = total
|
|
return total
|
|
except Exception as e:
|
|
logger.warning(f"Failed to get repo size for {repo_id}: {e}")
|
|
return 0
|
|
|
|
|
|
def _all_hf_cache_scans():
|
|
"""scan_cache_dir for the active, legacy, and default HF caches.
|
|
|
|
Each probe is isolated: an unreadable auxiliary cache (permission denied,
|
|
broken symlink, OS-redirected ~/.cache) is skipped, not fatal, so the
|
|
Downloaded list never blanks out and downloads never leak into Recommended.
|
|
"""
|
|
from huggingface_hub import scan_cache_dir
|
|
from utils.paths import legacy_hf_cache_dir, hf_default_cache_dir
|
|
|
|
scans = []
|
|
# Guard the active cache too: degrade to "no downloads" instead of raising.
|
|
try:
|
|
scans.append(scan_cache_dir())
|
|
except Exception as exc:
|
|
logger.warning("Could not scan active HF cache: %s", exc)
|
|
|
|
seen: set[str] = set()
|
|
try:
|
|
# Resolve the active cache dir for dedup.
|
|
from huggingface_hub.constants import HF_HUB_CACHE
|
|
seen.add(str(Path(HF_HUB_CACHE).resolve()))
|
|
except Exception:
|
|
pass
|
|
|
|
for extra_fn in (legacy_hf_cache_dir, hf_default_cache_dir):
|
|
try:
|
|
extra = extra_fn()
|
|
# is_dir()/resolve() can raise on an inaccessible path; skip it.
|
|
if not extra.is_dir():
|
|
continue
|
|
resolved = str(extra.resolve())
|
|
if resolved in seen:
|
|
continue
|
|
seen.add(resolved)
|
|
scans.append(scan_cache_dir(cache_dir = str(extra)))
|
|
except Exception as exc:
|
|
logger.warning("Could not scan HF cache %s: %s", extra_fn.__name__, exc)
|
|
return scans
|
|
|
|
|
|
def _is_gguf_filename(name: str) -> bool:
|
|
return name.lower().endswith(".gguf")
|
|
|
|
|
|
def _is_mmproj_filename(name: str) -> bool:
|
|
"""Match GGUF vision-adapter (mmproj) files. Consistent with
|
|
``utils.models.model_config._is_mmproj``."""
|
|
return "mmproj" in name.lower()
|
|
|
|
|
|
def _is_main_gguf_filename(name: str) -> bool:
|
|
"""A GGUF file that is a primary weight, not an mmproj vision
|
|
adapter."""
|
|
return _is_gguf_filename(name) and not _is_mmproj_filename(name)
|
|
|
|
|
|
def _iter_gguf_paths(root: Path):
|
|
for path in root.rglob("*"):
|
|
if path.is_file() and _is_gguf_filename(path.name):
|
|
yield path
|
|
|
|
|
|
def _repo_gguf_size_bytes(repo_info) -> int:
|
|
"""Total on-disk size of primary GGUF weight files across all
|
|
revisions, excluding mmproj vision-adapter files.
|
|
|
|
Hugging Face hardlinks blobs shared between revisions, so this
|
|
deduplicates by blob path (or revision commit hash + filename as a
|
|
fallback) to avoid double-counting. Unknown sizes (``size_on_disk is
|
|
None``, e.g. a partial download) count as zero. mmproj files are
|
|
excluded so repos whose only ``.gguf`` artifact is a vision adapter
|
|
aren't classed as GGUF repos: the variant selector filters mmproj
|
|
out and would otherwise show zero pickable variants.
|
|
"""
|
|
unique_blobs: dict[str, int] = {}
|
|
for revision in repo_info.revisions:
|
|
rev_id = getattr(revision, "commit_hash", None) or str(id(revision))
|
|
for f in revision.files:
|
|
if _is_main_gguf_filename(f.file_name):
|
|
blob_path = getattr(f, "blob_path", None)
|
|
size = f.size_on_disk or 0
|
|
if blob_path:
|
|
unique_blobs[str(blob_path)] = size
|
|
else:
|
|
unique_blobs[f"{rev_id}:{f.file_name}"] = size
|
|
return sum(unique_blobs.values())
|
|
|
|
|
|
def _repo_has_gguf_files(repo_info) -> bool:
|
|
"""True when any revision in a cached repo has a primary GGUF weight
|
|
file. Repos whose only ``.gguf`` artifact is an mmproj vision adapter
|
|
are not treated as GGUF here."""
|
|
return _repo_gguf_size_bytes(repo_info) > 0
|
|
|
|
|
|
def _blob_mtime(f) -> float:
|
|
"""Blob modification time in epoch seconds (0.0 if unknown).
|
|
|
|
Prefers HF metadata ``blob_last_modified``, falls back to stat(); uses
|
|
only mtimes (portable across Windows, macOS, Linux), never path parsing.
|
|
"""
|
|
ts = getattr(f, "blob_last_modified", None)
|
|
if isinstance(ts, (int, float)) and ts > 0:
|
|
return float(ts)
|
|
blob_path = getattr(f, "blob_path", None)
|
|
if blob_path:
|
|
try:
|
|
return float(Path(blob_path).stat().st_mtime)
|
|
except OSError:
|
|
pass
|
|
return 0.0
|
|
|
|
|
|
def _repo_gguf_last_modified(repo_info) -> float:
|
|
"""Newest mtime among a repo's primary (non-mmproj) GGUF blobs.
|
|
|
|
Drives the Downloaded list's "last downloaded" ordering and groups a
|
|
multi-quant repo by its most recently downloaded quant.
|
|
"""
|
|
latest = 0.0
|
|
for revision in repo_info.revisions:
|
|
for f in revision.files:
|
|
if _is_main_gguf_filename(f.file_name):
|
|
latest = max(latest, _blob_mtime(f))
|
|
return latest
|
|
|
|
|
|
@router.get("/cached-gguf")
|
|
async def list_cached_gguf(current_subject: str = Depends(get_current_subject)):
|
|
"""List GGUF repos downloaded to HF cache, legacy Unsloth cache, and HF default cache."""
|
|
try:
|
|
cache_scans = _all_hf_cache_scans()
|
|
|
|
seen_lower: dict[str, dict] = {}
|
|
for hf_cache in cache_scans:
|
|
for repo_info in hf_cache.repos:
|
|
try:
|
|
if repo_info.repo_type != "model":
|
|
continue
|
|
repo_id = repo_info.repo_id
|
|
if _is_hidden_model(repo_id):
|
|
continue
|
|
total_size = _repo_gguf_size_bytes(repo_info)
|
|
if total_size == 0:
|
|
continue
|
|
key = repo_id.lower()
|
|
existing = seen_lower.get(key)
|
|
last_modified = _repo_gguf_last_modified(repo_info)
|
|
if existing is None or total_size > existing["size_bytes"]:
|
|
row = {
|
|
"repo_id": repo_id,
|
|
"size_bytes": total_size,
|
|
"cache_path": str(repo_info.repo_path),
|
|
}
|
|
# Keep the newest timestamp across duplicate caches;
|
|
# attach only when known so absent rows sort as oldest.
|
|
lm = max(last_modified, (existing or {}).get("last_modified", 0.0))
|
|
if lm > 0:
|
|
row["last_modified"] = lm
|
|
seen_lower[key] = row
|
|
elif last_modified > existing.get("last_modified", 0.0):
|
|
existing["last_modified"] = last_modified
|
|
except Exception as e:
|
|
repo_label = getattr(repo_info, "repo_id", "<unknown>")
|
|
logger.warning(f"Skipping cached GGUF repo {repo_label}: {e}")
|
|
continue
|
|
# Newest download first; stable repo_id tie-break for equal/missing mtimes.
|
|
cached = sorted(
|
|
seen_lower.values(),
|
|
key = lambda c: (-(c.get("last_modified") or 0.0), c["repo_id"].lower()),
|
|
)
|
|
return {"cached": cached}
|
|
except Exception as e:
|
|
logger.error(f"Error listing cached GGUF repos: {e}", exc_info = True)
|
|
return {"cached": []}
|
|
|
|
|
|
@router.get("/cached-models")
|
|
async def list_cached_models(current_subject: str = Depends(get_current_subject)):
|
|
"""List non-GGUF model repos downloaded to HF cache, legacy Unsloth cache, and HF default cache."""
|
|
_WEIGHT_EXTENSIONS = (".safetensors", ".bin")
|
|
|
|
try:
|
|
cache_scans = _all_hf_cache_scans()
|
|
|
|
seen_lower: dict[str, dict] = {}
|
|
for hf_cache in cache_scans:
|
|
for repo_info in hf_cache.repos:
|
|
try:
|
|
if repo_info.repo_type != "model":
|
|
continue
|
|
repo_id = repo_info.repo_id
|
|
if _is_hidden_model(repo_id):
|
|
continue
|
|
if _repo_has_gguf_files(repo_info):
|
|
continue
|
|
total_size = sum(
|
|
(f.size_on_disk or 0) for rev in repo_info.revisions for f in rev.files
|
|
)
|
|
if total_size == 0:
|
|
continue
|
|
has_weights = any(
|
|
f.file_name.endswith(_WEIGHT_EXTENSIONS)
|
|
for rev in repo_info.revisions
|
|
for f in rev.files
|
|
)
|
|
if not has_weights:
|
|
continue
|
|
last_modified = max(
|
|
(
|
|
_blob_mtime(f)
|
|
for rev in repo_info.revisions
|
|
for f in rev.files
|
|
if f.file_name.endswith(_WEIGHT_EXTENSIONS)
|
|
),
|
|
default = 0.0,
|
|
)
|
|
key = repo_id.lower()
|
|
existing = seen_lower.get(key)
|
|
if existing is None or total_size > existing["size_bytes"]:
|
|
row = {
|
|
"repo_id": repo_id,
|
|
"size_bytes": total_size,
|
|
}
|
|
# Keep the newest timestamp across duplicate caches;
|
|
# attach only when known so absent rows sort as oldest.
|
|
lm = max(last_modified, (existing or {}).get("last_modified", 0.0))
|
|
if lm > 0:
|
|
row["last_modified"] = lm
|
|
seen_lower[key] = row
|
|
elif last_modified > existing.get("last_modified", 0.0):
|
|
existing["last_modified"] = last_modified
|
|
except Exception as e:
|
|
repo_label = getattr(repo_info, "repo_id", "<unknown>")
|
|
logger.warning(f"Skipping cached model repo {repo_label}: {e}")
|
|
continue
|
|
# Newest download first; stable repo_id tie-break for equal/missing mtimes.
|
|
cached = sorted(
|
|
seen_lower.values(),
|
|
key = lambda c: (-(c.get("last_modified") or 0.0), c["repo_id"].lower()),
|
|
)
|
|
return {"cached": cached}
|
|
except Exception as e:
|
|
logger.error(f"Error listing cached models: {e}", exc_info = True)
|
|
return {"cached": []}
|
|
|
|
|
|
@router.delete("/delete-cached")
|
|
async def delete_cached_model(
|
|
repo_id: str = Body(...),
|
|
variant: Optional[str] = Body(None),
|
|
current_subject: str = Depends(get_current_subject),
|
|
):
|
|
"""Delete a cached model repo (or a specific GGUF variant) from the HF cache.
|
|
|
|
With *variant*, only GGUF files matching that quant label are removed
|
|
(e.g. ``UD-Q4_K_XL``); otherwise the whole repo is deleted. Refuses
|
|
if the model is currently loaded for inference.
|
|
"""
|
|
if not _is_valid_repo_id(repo_id):
|
|
raise HTTPException(status_code = 400, detail = "Invalid repo_id format")
|
|
|
|
# Refuse if the model is currently loaded.
|
|
try:
|
|
from routes.inference import get_llama_cpp_backend
|
|
llama_backend = get_llama_cpp_backend()
|
|
if llama_backend.is_loaded and llama_backend.model_identifier:
|
|
loaded_id = llama_backend.model_identifier.lower()
|
|
if loaded_id == repo_id.lower() or loaded_id.startswith(repo_id.lower()):
|
|
raise HTTPException(
|
|
status_code = 400,
|
|
detail = "Unload the model before deleting",
|
|
)
|
|
except HTTPException:
|
|
raise
|
|
except Exception:
|
|
pass
|
|
|
|
try:
|
|
inference_backend = get_inference_backend()
|
|
if inference_backend.active_model_name:
|
|
active = inference_backend.active_model_name.lower()
|
|
if active == repo_id.lower() or active.startswith(repo_id.lower()):
|
|
raise HTTPException(
|
|
status_code = 400,
|
|
detail = "Unload the model before deleting",
|
|
)
|
|
except HTTPException:
|
|
raise
|
|
except Exception:
|
|
pass
|
|
|
|
try:
|
|
cache_scans = _all_hf_cache_scans()
|
|
|
|
target_repo = None
|
|
for hf_cache in cache_scans:
|
|
for repo_info in hf_cache.repos:
|
|
if repo_info.repo_type != "model":
|
|
continue
|
|
if repo_info.repo_id.lower() == repo_id.lower():
|
|
target_repo = repo_info
|
|
break
|
|
if target_repo is not None:
|
|
break
|
|
|
|
if target_repo is None:
|
|
raise HTTPException(status_code = 404, detail = "Model not found in cache")
|
|
|
|
# ── Per-variant GGUF deletion ────────────────────────────
|
|
if variant:
|
|
deleted_bytes = 0
|
|
deleted_count = 0
|
|
for rev in target_repo.revisions:
|
|
for f in rev.files:
|
|
if not _is_gguf_filename(f.file_name):
|
|
continue
|
|
quant = _extract_quant_label(f.file_name)
|
|
if quant.lower() != variant.lower():
|
|
continue
|
|
# Delete the blob (data) and the snapshot symlink.
|
|
try:
|
|
blob = Path(f.blob_path)
|
|
snap = Path(f.file_path)
|
|
size = blob.stat().st_size if blob.exists() else 0
|
|
if snap.exists() or snap.is_symlink():
|
|
snap.unlink()
|
|
if blob.exists():
|
|
blob.unlink()
|
|
deleted_bytes += size
|
|
deleted_count += 1
|
|
except Exception as e:
|
|
logger.warning(f"Failed to delete {f.file_name}: {e}")
|
|
|
|
if deleted_count == 0:
|
|
raise HTTPException(
|
|
status_code = 404,
|
|
detail = f"Variant {variant} not found in cache for {repo_id}",
|
|
)
|
|
|
|
freed_mb = deleted_bytes / (1024 * 1024)
|
|
logger.info(
|
|
f"Deleted {deleted_count} file(s) for {repo_id} variant {variant}: "
|
|
f"{freed_mb:.1f} MB freed"
|
|
)
|
|
return {"status": "deleted", "repo_id": repo_id, "variant": variant}
|
|
|
|
# ── Full repo deletion ───────────────────────────────────
|
|
revision_hashes = [rev.commit_hash for rev in target_repo.revisions]
|
|
if not revision_hashes:
|
|
raise HTTPException(status_code = 404, detail = "No revisions found for model")
|
|
|
|
delete_strategy = hf_cache.delete_revisions(*revision_hashes)
|
|
logger.info(
|
|
f"Deleting cached model {repo_id}: "
|
|
f"{delete_strategy.expected_freed_size_str} will be freed"
|
|
)
|
|
delete_strategy.execute()
|
|
|
|
return {"status": "deleted", "repo_id": repo_id}
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"Error deleting cached model {repo_id}: {e}", exc_info = True)
|
|
raise HTTPException(
|
|
status_code = 500,
|
|
detail = "Failed to delete cached model",
|
|
)
|
|
|
|
|
|
@router.get("/checkpoints", response_model = CheckpointListResponse)
|
|
async def list_checkpoints(
|
|
outputs_dir: str = Query(
|
|
default = str(outputs_root()),
|
|
description = "Directory to scan for checkpoints",
|
|
),
|
|
current_subject: str = Depends(get_current_subject),
|
|
):
|
|
"""List checkpoints in the outputs directory.
|
|
|
|
Scans the outputs folder for training runs and their checkpoints.
|
|
"""
|
|
try:
|
|
resolved_outputs_dir = str(resolve_output_dir(outputs_dir))
|
|
raw_models = scan_checkpoints(outputs_dir = resolved_outputs_dir)
|
|
|
|
models = [
|
|
ModelCheckpoints(
|
|
name = model_name,
|
|
checkpoints = [
|
|
CheckpointInfo(display_name = display_name, path = path, loss = loss)
|
|
for display_name, path, loss in checkpoints
|
|
],
|
|
base_model = metadata.get("base_model"),
|
|
peft_type = metadata.get("peft_type"),
|
|
lora_rank = metadata.get("lora_rank"),
|
|
is_quantized = metadata.get("is_quantized", False),
|
|
)
|
|
for model_name, checkpoints, metadata in raw_models
|
|
]
|
|
|
|
return CheckpointListResponse(
|
|
outputs_dir = resolved_outputs_dir,
|
|
models = models,
|
|
)
|
|
except Exception as e:
|
|
raise log_and_http_error(
|
|
e,
|
|
500,
|
|
"Failed to list checkpoints",
|
|
event = "models.list_checkpoints_failed",
|
|
log = logger,
|
|
)
|