path: root/app/backends/audiocpp/models.py

"""Model install state: what is on disk, what is missing, how to fetch it."""

import json
import os
import shutil
import sys
import tempfile
from pathlib import Path
from typing import Dict, List, Optional, Set, Tuple

from backends import common
from . import catalog as _catalog

# No-output watchdog for model downloads: huggingface_hub streams steady
# byte progress, so 5 minutes of silence means the transfer wedged. The
# runner kills it and reports exit 124 (see run_console_subprocess).
DOWNLOAD_STALL_TIMEOUT = 300

# Spec sanitizing lives with the catalog (the wizard's catalog view applies
# the same in-memory repair; the download path materializes it through its
# sanitized specs copy). The alias keeps this module's historical name.
_sanitize_model_spec = _catalog.sanitize_model_spec

def _installed_display_names(audiocpp_dir: Path,
                             model_entries: Optional[List[dict]],
                             install_guidance: List[Tuple[str, str]]
                             ) -> Set[str]:
    """Display names from INSTALL_GUIDANCE whose model files are on disk.

    MODEL_ENTRIES and INSTALL_GUIDANCE are built in lockstep by
    ``_build_entries`` (one guidance pair per entry), so the pairs resolve
    positionally: each entry's ``path`` is checked against the checkout
    file-precisely (see ``_entry_present``). Returns an empty set
    when ENTRIES is None or does not line up with the guidance (no
    filtering — every model counts as not installed).
    """
    if model_entries is None or len(model_entries) != len(install_guidance):
        return set()
    packages_by_dir = _catalog_packages_by_dir(audiocpp_dir)
    installed: Set[str] = set()
    for entry, (name, _install_id) in zip(model_entries, install_guidance):
        rel = entry.get("path")
        if not isinstance(rel, str) or not rel:
            continue
        if _entry_present(audiocpp_dir, rel, packages_by_dir):
            installed.add(name)
    return installed


def _split_pending_and_installed(
        install_guidance: List[Tuple[str, str]],
        installed_names: Set[str]) -> Tuple[List[Tuple[str, str]], List[str]]:
    """Partition guidance into (pending installs, installed display names).

    PENDING keeps only models whose display name is not INSTALLED_NAMES,
    de-duped by install id (the same package may host several entries) in
    first-occurrence order. INSTALLED lists each installed display name
    once, also in first-occurrence order.
    """
    seen: Set[str] = set()
    pending: List[Tuple[str, str]] = []
    noted: List[str] = []
    for name, install_id in install_guidance:
        if name in installed_names:
            if name not in noted:
                noted.append(name)
            continue
        if install_id in seen:
            continue
        seen.add(install_id)
        pending.append((name, install_id))
    return pending, noted


def _merge_companions(audiocpp_dir: Path,
                      pending: List[Tuple[str, str]],
                      companions: Optional[List[Tuple[str, str]]]
                      ) -> List[Tuple[str, str]]:
    """Fold COMPANIONS into PENDING, skipping ones already on disk.

    Each companion is a (display name, install id) pair for a package a
    hosted model requires but the TTS catalog never offers (MioCodec for
    MioTTS). Companions already pending (same install id) or already
    installed (their package's files present, checked file-precisely) are
    dropped; the installed ones are reported so the user knows the
    requirement is satisfied. Companion ids with no matching spec package
    are kept (the install command will report what is wrong).
    """
    merged = list(pending)
    seen = {install_id for _name, install_id in pending}
    for name, install_id in companions or []:
        if install_id in seen:
            continue
        package = _companion_package(audiocpp_dir, install_id)
        if package is not None \
                and _package_files_present(audiocpp_dir, package):
            print(f"[OK] {name} is already installed.")
            continue
        seen.add(install_id)
        merged.append((name, install_id))
    return merged


def _install_models(audiocpp_dir: Path,
                    install_guidance: List[Tuple[str, str]],
                    download: bool, emit=None, cancel=None,
                    model_entries: Optional[List[dict]] = None,
                    companions: Optional[List[Tuple[str, str]]] = None) -> int:
    """Report and optionally run the model install commands.

    When MODEL_ENTRIES (built in lockstep with INSTALL_GUIDANCE by
    ``_build_entries``) is given, models already on disk are reported as
    installed and never re-downloaded or printed as commands; when every
    selected model is present nothing runs at all. The remaining models
    get one ``python <manager> install <id>`` command each (de-duped by
    install id). COMPANIONS carries (display name, install id) pairs for
    packages a hosted model needs but the TTS catalog does not offer
    (MioCodec for MioTTS): they join the pending list, de-duped against
    it and skipped when their package's files are already on disk — so a
    configure run that only adds the companion still downloads it. When
    DOWNLOAD is True each command is run in the audio.cpp checkout via
    ``subprocess`` so the models are downloaded automatically; a failing
    install is reported as a warning and does not abort the remaining
    downloads. When DOWNLOAD is False (or the model manager is missing)
    the commands are only printed after a note that setup downloads
    them automatically — copy-pasteable for a manual install.

    With EMIT given (the in-TUI task view) each download streams its output
    to EMIT and — when the checkout's ``model_manager_v2.py`` supports it —
    runs with ``--progress --cancel-file`` so the view can show a real byte
    progress bar and cancel gracefully. CANCEL aborts a running download.

    When the checkout's model specs carry a broken ``strip_prefix`` (a dot
    prefix over repo-root files, or a single-GGUF package nested under a
    repository directory with no prefix — the manager installs files the
    server cannot load) and the manager supports ``--specs-dir``, the
    installs run against a sanitized temporary copy of the specs (see
    ``_prepare_specs_dir``); the checkout itself is left untouched. Specs
    that cannot be repaired confidently are left as-is: those installs
    fail, are reported as warnings, and the remaining downloads continue.

    Returns 0 when every command succeeded (or nothing needed running),
    130 when cancelled, 1 when any download failed.
    """
    if not install_guidance and not companions:
        return 0
    manager = audiocpp_dir / "tools" / "model_manager_v2.py"
    installed_names = _installed_display_names(
        audiocpp_dir, model_entries, install_guidance)
    pending, installed_noted = _split_pending_and_installed(
        install_guidance, installed_names)
    pending = _merge_companions(audiocpp_dir, pending, companions)

    supports_progress = emit is not None and _manager_supports_progress(manager)

    if download and not manager.is_file():
        print(f"[WARNING] {manager} not found; printing the install commands "
              "instead of running them")
        download = False

    for name in installed_noted:
        print(f"[OK] {name} is already installed.")
    if not pending:
        print("[OK] All selected models are already installed.")
        return 0

    if not download:
        print("[INFO] Models are downloaded automatically by this tool's "
              "setup — to download them manually instead, run:")
        for _, install_id in pending:
            print(f"python {manager} install {install_id}")
        return 0

    failed = False
    specs_dir: Optional[Path] = None
    if _manager_supports_flag(manager, "--specs-dir"):
        try:
            specs_dir = _prepare_specs_dir(audiocpp_dir)
        except OSError as exc:
            print(f"[WARNING] Could not prepare sanitized model specs: {exc}")
            specs_dir = None
        if specs_dir is not None:
            print(f"[INFO] The checkout's model specs carry a broken "
                  "strip_prefix; installing from a sanitized copy "
                  f"({specs_dir})")
    specs_args = (["--specs-dir", str(specs_dir)]
                  if specs_dir is not None else [])
    try:
        for _, install_id in pending:
            print(f"[INFO] Downloading {install_id}...")
            argv = [sys.executable, str(manager)] + specs_args + [
                "install", install_id]
            cancel_file: Optional[Path] = None
            on_cancel = None
            if supports_progress:
                fd, cancel_path = tempfile.mkstemp(
                    prefix="audiocpp_cancel_", suffix=".cancel")
                os.close(fd)
                cancel_file = Path(cancel_path)
                cancel_file.unlink()  # absent = not cancelled
                argv += ["--progress", "--cancel-file", str(cancel_file)]
                on_cancel = cancel_file.touch
            try:
                rc = common.run_console_subprocess(
                    argv, cwd=str(audiocpp_dir), emit=emit, cancel=cancel,
                    on_cancel=on_cancel,
                    stall_timeout=(DOWNLOAD_STALL_TIMEOUT
                                   if supports_progress else None))
            except OSError as exc:
                print(f"[WARNING] Could not run python {manager} install "
                      f"{install_id}: {exc}")
                rc = 1
            finally:
                if cancel_file is not None:
                    try:
                        cancel_file.unlink()
                    except OSError:
                        pass
            if rc == 130 or (cancel is not None and cancel.is_set()):
                return 130
            if rc != 0:
                failed = True
                print(f"[WARNING] install {install_id} exited with code "
                      f"{rc}; the model may need to be downloaded "
                      "by hand")
    finally:
        if specs_dir is not None:
            shutil.rmtree(specs_dir, ignore_errors=True)
    return 1 if failed else 0


def _manager_supports_flag(manager: Path, flag: str) -> bool:
    """True when MANAGER's (model_manager_v2.py's) source mentions FLAG.

    The checkout is downloaded, so an older copy may lack a relatively
    recent flag; probing the script source once is cheaper than failing an
    install with an unknown option. A false positive (the string appears
    outside argparse) is caught when the subprocess reports the error.
    """
    try:
        text = manager.read_text(encoding="utf-8", errors="ignore")
    except OSError:
        return False
    return flag in text


def _manager_supports_progress(manager: Path) -> bool:
    """True when MANAGER (model_manager_v2.py) supports --progress output.

    The ``--progress``/``--cancel-file`` flags are relatively recent; an
    older audio.cpp checkout may not have them, so probe the script source
    once instead of failing the download with an unknown flag.
    """
    return (_manager_supports_flag(manager, "AUDIOCPP_PROGRESS")
            and _manager_supports_flag(manager, "--cancel-file"))


def _sanitize_model_spec(spec: dict) -> bool:
    """Repair dot ``strip_prefix`` packages in SPEC, in place.

    A package's ``strip_prefix`` is stripped from the front of every file
    path to get the local layout, so it only works when every file is
    listed under that prefix (``<prefix>/<file>``). A dot prefix ("." or
    "./") is meant for files written ``./<file>``; when the package instead
    lists repo-root files bare (``model.gguf``), the manager rejects the
    whole package ("file path does not start with strip_prefix '.': ...")
    and nothing can be downloaded. Root-level files need no prefix at all
    (upstream specs like minimax_music3.json store ""), so dropping the dot
    prefix is the safe repair. Prefixes naming a real directory are left
    alone — the correct remote paths cannot be guessed. Returns True when
    SPEC changed.
    """
    changed = False
    for package in spec.get("packages") or []:
        if not isinstance(package, dict):
            continue
        prefix = str(package.get("strip_prefix") or "").rstrip("/")
        if prefix not in (".", ".."):
            continue
        files = package.get("files")
        if not isinstance(files, list) or not files:
            continue
        if all(isinstance(item, str) and item.startswith(prefix + "/")
               for item in files):
            continue
        package["strip_prefix"] = ""
        changed = True
    return changed


def _prepare_specs_dir(audiocpp_dir: Path) -> Optional[Path]:
    """Return a temp specs dir with dot ``strip_prefix`` entries repaired.

    audio.cpp's model manager accepts ``--specs-dir``, so a checkout whose
    specs carry a broken ``strip_prefix`` can be installed from a sanitized
    copy without modifying the checkout. Every ``model_specs/*.json`` is
    copied; the ones needing a repair are rewritten via
    ``_sanitize_model_spec`` (specs that fail to parse are copied verbatim
    so the manager reports them exactly as it would upstream). Returns None
    when no spec needed a repair (or the specs directory is missing or
    unreadable) — the caller then runs against the checkout's own specs.
    The caller owns the returned directory and removes it when the installs
    are done.
    """
    specs_dir = audiocpp_dir / "model_specs"
    try:
        spec_paths = sorted(specs_dir.glob("*.json"))
    except OSError:
        return None
    if not spec_paths:
        return None
    payloads: List[Tuple[str, str]] = []
    sanitized = False
    for spec_path in spec_paths:
        try:
            text = spec_path.read_text(encoding="utf-8")
        except OSError:
            return None
        try:
            spec = json.loads(text)
        except ValueError:
            payloads.append((spec_path.name, text))
            continue
        if isinstance(spec, dict) and _sanitize_model_spec(spec):
            sanitized = True
            text = json.dumps(spec, indent=2, ensure_ascii=False) + "\n"
        payloads.append((spec_path.name, text))
    if not sanitized:
        return None
    staging = Path(tempfile.mkdtemp(prefix="audiocpp_specs_"))
    try:
        for name, payload in payloads:
            (staging / name).write_text(payload, encoding="utf-8")
    except OSError:
        shutil.rmtree(staging, ignore_errors=True)
        raise
    return staging


def download_applicable(audiocpp_dir: Path, model_entries: List[dict],
                        companions: Optional[List[Tuple[str, str]]] = None
                        ) -> bool:
    """True when the wizard's "download models automatically?" row applies.

    The audio.cpp model manager must be present (otherwise the install
    commands can only be printed), and at least one selected model must be
    missing from disk (see ``_all_models_present``), or one COMPANION
    package a hosted model needs (MioCodec for MioTTS) must be — so an
    already-configured checkout is not asked to re-download models it
    already has.
    """
    manager = audiocpp_dir / "tools" / "model_manager_v2.py"
    if not manager.is_file():
        return False
    if not _all_models_present(audiocpp_dir, model_entries):
        return True
    for _name, install_id in companions or []:
        package = _companion_package(audiocpp_dir, install_id)
        if package is None \
                or not _package_files_present(audiocpp_dir, package):
            return True
    return False


def _build_tree_families(catalog: List[dict]) -> List[dict]:
    """Shape the catalog into the checkbox_tree widget's family list."""
    families: List[dict] = []
    for entry in catalog:
        tasks = set(entry["tasks"])
        # Clone-only families (e.g. Chatterbox) cannot synthesize without
        # a reference voice, so they do not advertise plain "tts".
        clone_only = _catalog.is_clone_only_family(entry["family"], tasks)
        capabilities = []
        if not clone_only:
            capabilities.append("tts")
        if "clone" in tasks or clone_only:
            capabilities.append("cloning")
        if "design" in tasks:
            capabilities.append("design")
        name = entry["display_name"]
        options = []
        for opt in _catalog.package_dir_options(entry):
            options.append({
                "key": opt["target_directory"],
                "label": opt["install_id"],
                "recommended": opt["recommended"],
            })
        families.append({
            "label": name,
            "detail": ", ".join(capabilities),
            "options": options,
        })
    return families


def _model_path_present(path: Path) -> bool:
    """True when a server.json model path holds actual model files.

    A present path is either a file (a single-model package) or a non-empty
    directory (the usual GGUF package target directory; an empty one means a
    download that never ran or was cleaned up halfway).
    """
    try:
        if path.is_file():
            return True
        if path.is_dir():
            return any(path.iterdir())
    except OSError:
        return False
    return False


def _package_files_present(audiocpp_dir: Path, package: dict) -> bool:
    """True when PACKAGE's files are on disk at their strip_prefix-stripped paths.

    Mirrors model_manager_v2.py's own ``package_is_installed`` check, which
    is what decides a re-install. Used instead of the plain "directory is
    non-empty" check so a stale install from a broken spec layout (e.g. the
    GLM/OuteTTS GGUFs nested under ``Text to audio (TTS)/`` before the spec
    sanitizer existed) counts as missing and gets re-downloaded correctly.
    """
    files = package.get("files") or []
    if not files:
        return False
    prefix = str(package.get("strip_prefix") or "").rstrip("/")
    base = audiocpp_dir / "models" / str(package.get("target_directory") or "")
    for item in files:
        if not isinstance(item, str):
            return False
        local = item
        if prefix:
            if not local.startswith(prefix + "/"):
                return False
            local = local[len(prefix) + 1:]
        try:
            if not (base / local).is_file():
                return False
        except OSError:
            return False
    return True


def _entry_directory_key(rel: str) -> str:
    """The ``<target_directory>`` a server.json model path belongs to.

    ``models/<dir>`` entries map to ``<dir>``; entries hosted from a file
    inside a package directory (``models/<dir>/<gguf>``, the multi-GGUF
    hosting convention) map to ``<dir>`` as well, so old directory-style
    and new file-style entries of the same package compare equal.
    """
    stripped = rel[len("models/"):] if rel.startswith("models/") else rel
    return stripped.split("/", 1)[0]


def _catalog_packages_by_dir(audiocpp_dir: Path) -> Dict[str, List[dict]]:
    """Map each catalog package target directory to its packages.

    The catalog's packages carry the sanitized strip prefixes (see
    ``load_model_catalog``), so presence checks see the same layout the
    model manager will install. Families whose specs cannot be read yield
    no entries and callers fall back to the plain path check.
    """
    try:
        entries = _catalog.load_model_catalog(audiocpp_dir)
    except (NotADirectoryError, OSError):
        return {}
    by_dir: Dict[str, List[dict]] = {}
    for entry in entries:
        for package in entry.get("packages") or []:
            if not isinstance(package, dict):
                continue
            directory = str(package.get("target_directory") or entry["family"])
            by_dir.setdefault(directory, []).append(package)
    return by_dir


def _companion_package(audiocpp_dir: Path, install_id: str) -> Optional[dict]:
    """The spec package with INSTALL_ID across every model spec.

    Companion packages live in specs the TTS catalog filters out
    (miocodec is an audio_tools family with no text-synthesis task), so
    the lookup scans all model_specs/*.json with the sanitizer applied,
    matching the sanitized layout the download will use.
    """
    try:
        spec_paths = sorted((audiocpp_dir / "model_specs").glob("*.json"))
    except OSError:
        return None
    for spec_path in spec_paths:
        try:
            spec = json.loads(spec_path.read_text(encoding="utf-8"))
        except (OSError, ValueError):
            continue
        if not isinstance(spec, dict):
            continue
        _catalog.sanitize_model_spec(spec)
        for package in spec.get("packages") or []:
            if isinstance(package, dict) \
                    and str(package.get("id") or "") == install_id:
                return package
    return None


def _entry_present(audiocpp_dir: Path, rel: str,
                   packages_by_dir: Dict[str, List[dict]]) -> bool:
    """Whether the model entry path REL holds its package's files.

    When REL's directory matches a catalog package, the check is
    file-precise (every package file at its stripped path). Otherwise the
    plain path check applies: the entry may host a package the local
    specs do not describe (an older checkout, a custom path), and "the
    configured path exists" is then the best available signal.
    """
    directory = _entry_directory_key(rel)
    candidates = packages_by_dir.get(directory)
    if candidates is not None:
        return any(_package_files_present(audiocpp_dir, package)
                   for package in candidates)
    path = Path(rel) if Path(rel).is_absolute() else audiocpp_dir / rel
    return _model_path_present(path)


def _all_models_present(audiocpp_dir: Path, model_entries: List[dict]) -> bool:
    """True when every selected model entry's files are on disk.

    Presence is file-precise against the catalog packages (see
    ``_entry_present``): a directory that merely exists — e.g. holding a
    stale install from a since-repaired spec layout — counts as missing so
    the next download replaces it with the correct layout. An empty
    selection is treated as not-present.
    """
    if not model_entries:
        return False
    packages_by_dir = _catalog_packages_by_dir(audiocpp_dir)
    for entry in model_entries:
        rel = entry.get("path")
        if not isinstance(rel, str) or not rel:
            return False
        if not _entry_present(audiocpp_dir, rel, packages_by_dir):
            return False
    return True


def _server_entries_by_presence(server_json: Path, present: bool) -> List[dict]:
    """The server.json model entries whose on-disk presence matches PRESENT.

    Presence is file-precise against the catalog packages when the
    server.json's directory hosts the model_specs (the usual checkout
    layout); otherwise the plain path check applies. Each returned entry
    carries the entry ``id`` and ``rel`` (the configured path string),
    resolved exactly like the server resolves them (relative against the
    server.json's directory; absolute paths honored).
    """
    try:
        data = json.loads(server_json.read_text(encoding="utf-8"))
    except (OSError, ValueError):
        return []
    if not isinstance(data, dict):
        return []
    base = server_json.parent
    packages_by_dir = _catalog_packages_by_dir(base)
    matching: List[dict] = []
    for entry in data.get("models") or []:
        if not isinstance(entry, dict):
            continue
        rel = entry.get("path")
        if not isinstance(rel, str) or not rel:
            continue
        if _entry_present(base, rel, packages_by_dir) != present:
            continue
        matching.append({"id": str(entry.get("id") or rel), "rel": rel})
    return matching


def missing_model_entries(server_json: Path) -> List[dict]:
    """Return the server.json model entries whose files are not on disk.

    Used by ``detect`` to warn that a conversion would fail until the
    models are installed (see ``_server_entries_by_presence``).
    """
    return _server_entries_by_presence(server_json, present=False)


def installed_model_entries(server_json: Path) -> List[dict]:
    """Return the server.json model entries whose files ARE on disk.

    The complement of ``missing_model_entries`` (see
    ``_server_entries_by_presence``). Used by the wizard's "Delete unused
    models?" step to find already-downloaded models that were unselected.
    """
    return _server_entries_by_presence(server_json, present=True)


def _install_id_by_path(audiocpp_dir: Path) -> Dict[str, str]:
    """Map ``models/<target_directory>`` -> catalog install id.

    The catalog package that installs a model is derived from the
    ``default_path`` of each TTS family; an entry whose path matches no
    catalog package has no install id.
    """
    by_path: Dict[str, str] = {}
    try:
        for entry in _catalog.load_model_catalog(audiocpp_dir):
            by_path[entry["default_path"]] = entry["install_id"]
    except (NotADirectoryError, OSError):
        pass
    return by_path


def missing_model_install_guidance(audiocpp_dir: Path,
                                   missing: List[dict]) -> List[Tuple[str, str]]:
    """Map MISSING model entries to (display name, install id) pairs.

    The install id is derived from each entry's configured path via the
    catalog (see ``_install_id_by_path``); entries whose path matches no
    catalog package are skipped (there is no ``model_manager_v2.py install``
    command for them). Feeds ``_install_models`` for the "Download Missing
    Models" action.
    """
    by_path = _install_id_by_path(audiocpp_dir)
    guidance: List[Tuple[str, str]] = []
    for item in missing:
        install_id = by_path.get(item["rel"])
        if install_id:
            guidance.append((item["id"], install_id))
    return guidance


def model_install_hints(audiocpp_dir: Path,
                        missing: List[dict]) -> List[str]:
    """Remediation lines for MISSING model entries (see missing_model_entries).

    Maps each entry's configured path back to the catalog package that
    installs it (``models/<target_directory>`` -> install id) so the line
    carries the exact ``model_manager_v2.py install`` command; entries whose
    directory matches no catalog package just name the path.
    """
    by_path = _install_id_by_path(audiocpp_dir)
    hints: List[str] = []
    for item in missing:
        install_id = by_path.get(item["rel"])
        hint = f"model not downloaded: {item['id']} ({item['rel']})"
        if install_id:
            hint += (f" — install with: python tools/model_manager_v2.py "
                     f"install {install_id}")
        hints.append(hint)
    return hints


def install_models(audiocpp_dir: Path,
                   guidance: List[Tuple[str, str]],
                   emit=None, cancel=None) -> int:
    """Download the (display name, install id) models via the helper script.

    Runs ``model_manager_v2.py install`` for each de-duped install id in the
    checkout, streaming to the console (or to EMIT, the in-TUI task view); a
    failing install is reported as a warning and does not abort the rest.
    Returns 0 when every download succeeded, 130 when cancelled, 1 when any
    failed. Used by the hub's "Download Missing Models" action (see
    ``missing_model_install_guidance`` for the mapping).
    """
    return _install_models(audiocpp_dir, guidance, download=True,
                           emit=emit, cancel=cancel)


def hand_install_guidance(audiocpp_dir: Path,
                          missing: List[dict]) -> str:
    """Explain how to install MISSING model entries by hand.

    Returns a multi-line message listing each missing model's id and the
    path its files must be placed in (``rel``, resolved against the
    checkout). Used when the missing models cannot be mapped to
    a ``model_manager_v2.py install`` command, so the user still knows what
    to download and where to put it.
    """
    lines = [
        "None of the missing models map to a model_manager_v2.py install "
        "command.",
        "Download them by hand and place the files at these paths:",
    ]
    for item in missing:
        lines.append(f"  {item['id']}  ->  {item['rel']}")
    lines.append(f"(paths are relative to {audiocpp_dir})")
    return "\n".join(lines)


def unused_installed_entries(server_json: Path,
                             new_paths: Set[str]) -> List[dict]:
    """Return installed server.json entries whose package is not in NEW_PATHS.

    The already-downloaded models (see ``installed_model_entries``) that the
    new selection does not host any more — the candidates for the wizard's
    "Delete unused models?" prompt. Entries are compared by their package
    directory key (``_entry_directory_key``), so an entry re-hosted from
    ``models/<dir>`` to ``models/<dir>/<gguf>`` (the multi-GGUF convention)
    is not offered for deletion when the new config still hosts that
    directory. Entries whose files are not on disk are never listed (there
    is nothing to delete).
    """
    new_keys = {_entry_directory_key(path) for path in new_paths}
    return [entry for entry in installed_model_entries(server_json)
            if _entry_directory_key(entry["rel"]) not in new_keys]


def delete_model_files(server_json: Path, entries: List[dict]) -> int:
    """Remove the on-disk model files for ENTRIES ({id, rel}) from disk.

    Each entry's ``rel`` is resolved exactly like the server resolves it
    (relative against ``server_json``'s directory; absolute paths honored),
    then removed as a directory tree or a single file. Missing entries are
    ignored. Returns the number of paths removed. Used by the wizard's
    "Delete unused models?" step — the regenerated server.json already only
    lists the kept models, so no entry cleanup is needed here.
    """
    base = server_json.parent
    removed = 0
    for item in entries:
        rel = item.get("rel")
        if not isinstance(rel, str) or not rel:
            continue
        path = Path(rel) if Path(rel).is_absolute() else base / rel
        try:
            if not path.exists():
                continue
            if path.is_dir():
                shutil.rmtree(path, ignore_errors=True)
            else:
                path.unlink()
        except OSError as exc:
            print(f"[WARNING] Could not remove {path}: {exc}")
            continue
        print(f"[OK] Removed unused model {path}")
        removed += 1
    return removed