unsloth/studio/backend/utils/llama_cpp_update.py

# SPDX-License-Identifier: AGPL-3.0-only
# Copyright 2026-present the Unsloth AI Inc. team. All rights reserved. See /studio/LICENSE.AGPL-3.0

"""In-app llama.cpp prebuilt update: the *apply* half over utils.llama_cpp_freshness, running install_llama_prebuilt.py to download the newest bundle for this host and atomically swap it in. ``update_available`` (installed_tag != latest_tag) is laxer than freshness' ``stale`` (which also requires the install to be >= 3 days old); the UI shows the update affordance on update_available. The install is slow, so it runs on a daemon thread and callers poll get_update_status(). Everything fails open: a missing marker, offline GitHub or a source build reports update_available=False and never blocks the app. Mechanics (managed-root resolution, local-link detection, the resolve probe, the streamed installer run) live in utils.prebuilt.update_flow; this module keeps the llama policy and the job dict. whisper.cpp piggybacks on this single update item: status folds in a whisper sub-status (update_available becomes the union) and apply chains a whisper phase when whisper is behind (update_flow.run_chained_update, whisper_cpp_update.chained_phase_plan). audio.cpp rides along the same way as a third phase (audio_cpp_update.chained_phase_plan)."""

from __future__ import annotations

import os
import re
import subprocess
import sys
import threading
from contextlib import ExitStack
from pathlib import Path
from typing import Optional

import structlog

from utils.llama_cpp_freshness import (
    _INSTALL_MARKER_NAME,
    check_prebuilt_freshness,
    latest_published_release,
    latest_release_assets,
    parse_base_build,
    read_install_marker,
    reset_caches,
    update_download_size_bytes,
)
from utils.llama_cpp_changelog import (
    changelog_for_update,
    release_page_url,
    unavailable_reason,
)
from utils.prebuilt import update_flow as _flow
from utils.prebuilt.llama_backend import (
    REQUESTABLE_BACKENDS,
    environment_backend_override,
    marker_backend,
    marker_backend_request,
    normalize_backend,
)
from utils.update_status import update_checks_disabled

logger = structlog.get_logger(__name__)

DEFAULT_PUBLISHED_REPO = "unslothai/llama.cpp"
_INSTALL_TIMEOUT_SECONDS = 1800  # 30 min ceiling for download + build/validate
# install_llama_prebuilt.py EXIT_NO_SPACE: out of disk, retrying will not help.
_EXIT_NO_SPACE = 4
# A concrete backend selection could not be satisfied.
_EXIT_BACKEND_UNAVAILABLE = 5
# Prebuilt path failed; setup scripts source-build, but the in-app updater cannot.
_EXIT_FALLBACK = 2


class _LlamaPhaseError(RuntimeError):
    """A llama phase failed after possibly unloading the active runtime."""

    def __init__(self, message: str, *, reload_required: bool):
        super().__init__(message)
        self.reload_required = reload_required


_JOB_IDLE = _flow.JOB_IDLE
_JOB_RUNNING = _flow.JOB_RUNNING
_JOB_SUCCESS = _flow.JOB_SUCCESS
_JOB_ERROR = _flow.JOB_ERROR

_job_lock = threading.Lock()
# Covers the whole operation including release/backend resolution before the worker starts. _job_lock only protects the status dict and must never be held across network work.
_operation_lock = threading.Lock()
_job: dict = _flow.new_job()
_ALREADY_RUNNING_MESSAGE = "Another llama.cpp install is already running."

_utcnow = _flow.utcnow
_is_under = _flow.is_under
_is_external_link = _flow.is_external_link
_rocm_install_args = _flow.rocm_install_args


def _find_binary() -> Optional[str]:
    """Locate the active llama-server via the inference backend's own resolver, so update targets exactly what Unsloth runs. Lazy import keeps the heavy inference module off this import path."""
    try:
        from core.inference.llama_cpp import LlamaCppBackend
        return LlamaCppBackend._find_llama_server_binary()
    except Exception as exc:  # pragma: no cover - defensive
        logger.debug("llama update: binary discovery failed", error = str(exc))
        return None


def _install_dir_for(binary_path: Optional[str]) -> Optional[Path]:
    """The directory holding UNSLOTH_PREBUILT_INFO.json, i.e. the install root install_llama_prebuilt.py wrote and the one we re-install into."""
    return _flow.install_dir_for(binary_path, marker_name = _INSTALL_MARKER_NAME)


def _installer_script() -> Optional[Path]:
    """Locate install_llama_prebuilt.py (UNSLOTH_LLAMA_INSTALLER wins)."""
    return _flow.find_installer_script(
        env_var = "UNSLOTH_LLAMA_INSTALLER", script_name = "install_llama_prebuilt.py"
    )


# Markerless (source-build) installs have no UNSLOTH_PREBUILT_INFO.json, so we
# ask the installer whether an official prebuilt now exists for this host.
_resolve_memo: dict = {}


def _resolve_prebuilt_for_host(*, force_refresh: bool = False) -> Optional[dict]:
    """Run install_llama_prebuilt.py --resolve-prebuilt (no download) -> {prebuilt_available, repo, release_tag, llama_tag, asset, install_kind} or None. Fail-open: any error -> None so a source build never blocks the app."""
    return _flow.resolve_prebuilt_for_host(
        force_refresh = force_refresh,
        memo = _resolve_memo,
        installer_script = lambda: _installer_script(),
        log_message = "llama update: resolve-prebuilt failed",
    )


def _installed_build_number(binary: Optional[str]) -> Optional[int]:
    """Best-effort build number from ``llama-server --version``. Current llama.cpp reports a semantic version then ``build NNNN``; older binaries put the build directly after ``version:``. A build <= 1 (setup's ``--depth 1`` source build, #12798) is recovered from the checkout's bNNNN tag; None when that fails, treated as unknown (offer update)."""
    if not binary:
        return None
    try:
        proc = subprocess.run(
            [binary, "--version"],
            capture_output = True,
            text = True,
            encoding = "utf-8",
            errors = "replace",
            timeout = 20,
        )
    except Exception:  # pragma: no cover - defensive
        return None
    output = "\n".join((proc.stderr or "", proc.stdout or ""))
    m = re.search(r"version:[^\r\n]*\bbuild\s+(\d+)\b", output)
    if not m:
        m = re.search(r"version:\s*(\d+)\b(?!\.)", output)
    if m and int(m.group(1)) > 1:
        return int(m.group(1))
    commit = re.search(r"version:[^\r\n]*\bcommit\s+([0-9a-fA-F]{7,64})\b", output) or re.search(
        r"version:\s*\d+\s*\(([0-9a-fA-F]{7,64})\)", output
    )
    return _checkout_tag_build(binary, commit.group(1)) if commit else None


def _checkout_tag_build(binary: str, reported_commit: str) -> Optional[int]:
    """Build number from the ``bNNNN`` tag naming the checkout's HEAD, from git files only. FETCH_HEAD is the only record setup.ps1's ``fetch --depth 1 origin bNNNN`` + ``checkout -B`` leaves. The binary's commit must prefix HEAD, so a checkout moved after the build is not misread. None on any mismatch."""
    # A loose ref replaces its packed entry, as in git.
    tags: dict[str, set[str]] = {}
    packed: dict[str, str] = {}
    try:
        git_dir = next(
            (p / ".git" for p in list(Path(binary).parents)[:5] if (p / ".git").exists()), None
        )
        if git_dir is None or not git_dir.is_dir():
            return None
        head = (git_dir / "HEAD").read_text(encoding = "utf-8").strip()
        packed_refs = git_dir / "packed-refs"
        if packed_refs.is_file():
            last_tag = None
            for line in packed_refs.read_text(encoding = "utf-8").splitlines():
                if line.startswith("^"):
                    if last_tag:
                        tags[last_tag].add(line[1:].strip())
                    continue
                parts = line.split()
                last_tag = None
                if len(parts) == 2 and not line.startswith("#"):
                    packed[parts[1]] = parts[0]
                    if parts[1].startswith("refs/tags/"):
                        last_tag = parts[1][len("refs/tags/") :]
                        tags[last_tag] = {parts[0]}
        if head.startswith("ref:"):
            ref = head[4:].strip()
            loose = git_dir / ref
            head = (
                loose.read_text(encoding = "utf-8").strip()
                if loose.is_file()
                else packed.get(ref, "")
            )
        tags_dir = git_dir / "refs" / "tags"
        if tags_dir.is_dir():
            for f in tags_dir.iterdir():
                if f.is_file():
                    tags[f.name] = {f.read_text(encoding = "utf-8").strip()}
        fetch_head = git_dir / "FETCH_HEAD"
        if fetch_head.is_file():
            for line in fetch_head.read_text(encoding = "utf-8").splitlines():
                fm = re.match(r"([0-9a-fA-F]{40,64})\t[^\t]*\ttag '([^']+)'", line)
                if fm:
                    tags.setdefault(fm.group(2), set()).add(fm.group(1))
    except (OSError, UnicodeDecodeError, ValueError):
        return None
    head = head.lower()
    if not re.fullmatch(r"[0-9a-f]{40,64}", head):
        return None
    if not head.startswith(reported_commit.lower()):
        return None
    builds = [
        int(tm.group(1))
        for name, shas in tags.items()
        if head in {sha.lower() for sha in shas} and (tm := re.fullmatch(r"b([0-9]{1,9})", name))
    ]
    builds = [n for n in builds if n > 1]
    return max(builds) if builds else None


def get_installed_llama_version() -> Optional[str]:
    """Display string for the active llama.cpp install (e.g. 'b9585' or 'b9601-mix-a0e2906'), or None. Prefers the marker's release_tag, the full unsloth release identity the update banner compares as installed (#6219), so a mix build reads back in full rather than collapsing to its base; the marker's bare ``tag`` is only the upstream build (no '-mix-<commit>'), so it is the fallback, and ``b<build>`` from ``llama-server --version`` is the last resort for markerless builds. Reads the local marker and at most runs --version, no network or freshness work, so it is safe on latency-sensitive paths like the About panel."""
    binary = _find_binary()
    marker = read_install_marker(binary)
    if marker:
        tag = marker.get("release_tag") or marker.get("tag")
        if tag:
            return tag
    # Markerless/source build: the fallback execs ``llama-server --version``. Skip it while an update is swapping the tree, since on Windows that exec can make the installer's os.replace fail; the panel just omits the row.
    with _job_lock:
        job_running = _job["state"] == _JOB_RUNNING
    if job_running:
        return None
    n = _installed_build_number(binary)
    return f"b{n}" if n is not None else None


def _llama_install_root(binary: Optional[str]) -> Optional[Path]:
    """The Unsloth-managed llama.cpp root the active binary lives under, or None when unmanaged (see update_flow.managed_install_root)."""
    return _flow.managed_install_root(
        binary,
        marker_root = _install_dir_for(binary),
        server_path_var = "LLAMA_SERVER_PATH",
        cpp_path_var = "UNSLOTH_LLAMA_CPP_PATH",
        dir_name = "llama.cpp",
    )


def _source_build_status(binary: str, *, force_refresh: bool) -> Optional[dict]:
    """Update status for a markerless (source-build) install: offer the official prebuilt when one exists for this host and is newer. None -> caller falls through to the no-marker default (unsupported)."""
    res = _resolve_prebuilt_for_host(force_refresh = force_refresh)
    if not res or not res.get("prebuilt_available"):
        return None
    # llama_tag is the upstream bNNNN base whose numeric part matches --version's build field; release_tag is the full tag, either a same-base mix (bNNNN-mix-<sha>) or a fork wrapper (e.g. v1.0). Compare the numeric base against llama_tag.
    base_tag = res.get("llama_tag") or res.get("release_tag")
    release_tag = res.get("release_tag")
    if not base_tag:
        return None
    # No resolvable install root (e.g. a pinned LLAMA_SERVER_PATH we cannot manage) means an apply would not take effect, so do not offer.
    if _llama_install_root(binary) is None:
        return None
    installed_build = _installed_build_number(binary)
    latest_build = parse_base_build(base_tag)
    # A same-base mix adds patches the bare base lacks, so it is newer even at an unchanged build number; the bNNNN anchor keeps a fork wrapper tag from being read as a mix.
    latest_is_mix = (
        isinstance(release_tag, str)
        and latest_build is not None
        and parse_base_build(release_tag) == latest_build
        and release_tag.strip() != f"b{latest_build}"
    )
    if installed_build is None or latest_build is None:
        # Unknown installed/latest version (the involuntary source-build case): treat as behind so we still offer the prebuilt.
        update_available = True
    elif installed_build < latest_build:
        update_available = True
    elif installed_build == latest_build:
        # Same upstream base: offer the extra-patch mix, never a bare rebuild.
        update_available = latest_is_mix
    else:
        # Source build newer than the latest prebuilt: downgrade guard.
        update_available = False
    # Display the mix tag when that's what makes it newer; otherwise the base.
    latest = release_tag if latest_is_mix else base_tag
    # Size of the resolved prebuilt, so source builds show it like the marker path. Fails open to None (offline / asset absent).
    update_size_bytes = None
    if update_available:
        asset_name = res.get("asset")
        if isinstance(asset_name, str) and asset_name:
            try:
                assets = latest_release_assets(res.get("repo"), force_refresh = force_refresh)
                if assets:
                    update_size_bytes = assets.get(asset_name)
            except Exception as exc:  # pragma: no cover - network defensive
                logger.debug("llama update: source-build size lookup failed", error = str(exc))
    with _job_lock:
        job = dict(_job)
    return {
        "supported": True,
        "update_available": update_available,
        "stale": False,
        "installed_tag": (f"b{installed_build}" if installed_build else None),
        "latest_tag": latest,
        "published_repo": res.get("repo"),
        "installed_at_utc": None,
        "age_days": None,
        "source_build": True,
        "update_size_bytes": update_size_bytes,
        "job": job,
    }


def _active_install_is_local_link(binary: Optional[str]) -> bool:
    """True when the active llama-server resolves through a --with-llama-cpp-dir local link at the canonical llama.cpp directory (update_flow.active_install_is_local_link)."""
    return _flow.active_install_is_local_link(binary, dir_name = "llama.cpp")


def _studio_custom_path_active() -> bool:
    """True when Settings, rather than the managed installer, owns the runtime."""
    try:
        from utils.llama_cpp_path_settings import custom_llama_cpp_path_source
        return custom_llama_cpp_path_source() == "studio"
    except Exception:
        return False


def _local_link_status() -> dict:
    """Status payload for a local-link install: unmanaged, no update offered."""
    return _flow.local_link_status(_job, _job_lock)


def _whisper_chain_status(
    *, force_refresh: bool = False, paired_llama_will_update: bool = False
) -> Optional[dict]:
    """Whisper's piggyback plan for the combined update item (whisper_cpp_update.chained_phase_plan). None disables the piggyback: fail-open so whisper can never break the llama status or apply."""
    try:
        from utils import whisper_cpp_update
        return whisper_cpp_update.chained_phase_plan(
            force_refresh = force_refresh,
            paired_llama_will_update = paired_llama_will_update,
        )
    except Exception as exc:  # pragma: no cover - defensive
        logger.debug("llama update: whisper piggyback probe failed", error = str(exc))
        return None


def _merge_whisper_status(status: dict, *, force_refresh: bool = False) -> dict:
    """Fold the whisper sub-status into the llama payload: the llama update item is the single UI surface, so update_available becomes the union (llama behind OR whisper behind) while llama_update_available keeps the llama-only flag. All pre-existing top-level fields are preserved."""
    status["llama_update_available"] = bool(status.get("update_available"))
    plan = _whisper_chain_status(
        force_refresh = force_refresh,
        paired_llama_will_update = status["llama_update_available"],
    )
    if plan is None:
        status["whisper"] = None
        status["update_component"] = "llama" if status["llama_update_available"] else None
        return status
    sub = plan.get("status") or {}
    status["whisper"] = {
        "update_available": bool(plan.get("update_available")),
        "installed_tag": sub.get("installed_tag"),
        "latest_tag": sub.get("latest_tag"),
        "update_size_bytes": sub.get("update_size_bytes"),
        "skip_reason": plan.get("skip_reason"),
    }
    whisper_update_available = bool(plan.get("update_available"))
    if whisper_update_available:
        status["update_available"] = True
    status["update_component"] = (
        "llama"
        if status["llama_update_available"]
        else "whisper"
        if whisper_update_available
        else None
    )
    return status


def _audio_chain_status(*, force_refresh: bool = False) -> Optional[dict]:
    """audio.cpp's plan for the combined update item. None disables it: fail-open like the whisper probe."""
    try:
        from utils import audio_cpp_update
        return audio_cpp_update.chained_phase_plan(force_refresh = force_refresh)
    except Exception as exc:  # pragma: no cover - defensive
        logger.debug("llama update: audio.cpp probe failed", error = str(exc))
        return None


def _merge_audio_status(status: dict, *, force_refresh: bool = False) -> dict:
    """Fold the audio.cpp sub-status in after whisper's: it joins the update_available union and names the component only when neither llama.cpp nor whisper.cpp is pending."""
    plan = _audio_chain_status(force_refresh = force_refresh)
    if plan is None:
        status["audio"] = None
        return status
    sub = plan.get("status") or {}
    audio_update_available = bool(plan.get("update_available"))
    status["audio"] = {
        "update_available": audio_update_available,
        "installed_tag": sub.get("installed_tag"),
        "latest_tag": sub.get("latest_tag"),
        "update_size_bytes": sub.get("update_size_bytes"),
        "skip_reason": plan.get("skip_reason"),
    }
    if audio_update_available:
        status["update_available"] = True
        if status.get("update_component") is None:
            status["update_component"] = "audio"
    return status


def get_update_status(*, force_refresh: bool = False) -> dict:
    """Report whether an update is available plus the job state. This is the single main update item: llama.cpp drives it and the whisper and audio.cpp phases are folded in (_merge_whisper_status, _merge_audio_status). force_refresh bypasses the 24h release cache for an explicit "check now"."""
    status = _llama_only_status(force_refresh = force_refresh)
    status = _merge_whisper_status(status, force_refresh = force_refresh)
    return _merge_audio_status(status, force_refresh = force_refresh)


def get_update_changelog(
    *,
    force_refresh: bool = False,
    installed_tag: Optional[str] = None,
    latest_tag: Optional[str] = None,
) -> dict:
    """Only carried changes added after the installed llama.cpp release. Separate from the polled status endpoint so opening the banner, not every 3s poll, pays for the two exact release lookups. installed_tag/latest_tag name the pair the caller is displaying."""
    empty = {
        "matched": False,
        "installed_tag": None,
        "latest_tag": None,
        "changes": [],
        "total_changes": 0,
        "truncated": False,
        "release_url": None,
        "error": None,
    }
    binary = _find_binary()
    if _studio_custom_path_active() or _active_install_is_local_link(binary):
        return empty
    marker = read_install_marker(binary)
    if not marker:
        return empty
    repo = marker.get("published_repo") or DEFAULT_PUBLISHED_REPO
    installed_full = marker.get("release_tag") or marker.get("tag")
    # force_refresh reaches only the exact release lookups. Refreshing the latest pointer would retarget a release the banner has not adopted, which it rejects.
    freshness = check_prebuilt_freshness(binary)
    latest = freshness.get("latest_tag")
    empty["installed_tag"] = freshness.get("installed_tag")
    empty["latest_tag"] = latest
    if not freshness.get("behind") or not installed_full or not latest:
        return empty
    # Answer about the caller's pair, not whatever the shared latest-release memo now holds: any other surface's forced status check advances it process-wide and the frontend rejects a mismatched answer, Retry included. Only while the install still matches.
    if latest_tag and installed_tag and installed_tag == empty["installed_tag"]:
        latest = latest_tag
        empty["latest_tag"] = latest
    # Offer the release page even when the comparison fails; GitHub can still show it.
    empty["release_url"] = release_page_url(repo, latest)
    try:
        result = changelog_for_update(
            repo,
            installed_full,
            latest,
            force_refresh = force_refresh,
        )
    except Exception as exc:  # pragma: no cover - defensive
        logger.debug("llama changelog comparison failed", error = str(exc))
        result = None
    if result is None:
        try:
            empty["error"] = unavailable_reason(repo, installed_full, latest)
        except Exception as exc:  # pragma: no cover - defensive
            logger.debug("llama changelog reason lookup failed", error = str(exc))
            empty["error"] = "release_notes_unavailable"
        return empty
    return {**empty, **result, "matched": True}


def _llama_only_status(
    *, force_refresh: bool = False, allow_source_probe_while_running: bool = False
) -> dict:
    """The llama.cpp half of get_update_status (no whisper sub-status)."""
    binary = _find_binary()
    # A path selected in Settings is user-managed even if its folder happens to contain an Unsloth prebuilt marker. Never offer to replace that tree.
    if _studio_custom_path_active():
        return _local_link_status()
    # A --with-llama-cpp-dir local link is the user's own tree; never offer to replace it. Bail before any network/freshness work.
    if _active_install_is_local_link(binary):
        return _local_link_status()
    marker = read_install_marker(binary)
    checks_disabled = update_checks_disabled()

    with _job_lock:
        job_running = _job["state"] == _JOB_RUNNING

    # No marker = source build / custom path: offer the official prebuilt if one now exists for this host (this is why macOS source builds showed no button). Skipped while the updater swaps the tree, since each 3s poll would exec the half-replaced binary (on Windows that can make the installer's os.replace fail) and the poller only consumes job progress.
    if (
        marker is None
        and binary is not None
        and not checks_disabled
        and (not job_running or allow_source_probe_while_running)
    ):
        src = _source_build_status(binary, force_refresh = force_refresh)
        if src is not None:
            return src

    repo = (marker or {}).get("published_repo") or DEFAULT_PUBLISHED_REPO

    if force_refresh and repo and not checks_disabled:
        try:
            latest_published_release(repo, force_refresh = True)
        except Exception as exc:  # pragma: no cover - network defensive
            logger.debug("llama update: force refresh failed", error = str(exc))

    freshness = check_prebuilt_freshness(binary)
    installed = freshness.get("installed_tag")
    latest = freshness.get("latest_tag")
    # `behind` compares the full release identity with a base-build guard, so a lagging /releases/latest or a mix-tagged latest cannot show a false update (llama_cpp_freshness.is_behind).
    update_available = bool(freshness.get("has_marker") and freshness.get("behind"))

    # Size of the prebuilt Update would download, for the banner. Only when an update is offered; fails open to None.
    update_size_bytes = None
    if update_available:
        try:
            update_size_bytes = update_download_size_bytes(
                marker,
                latest,
                freshness.get("published_repo") or repo,
                force_refresh = force_refresh,
            )
        except Exception as exc:  # pragma: no cover - network defensive
            logger.debug("llama update: size lookup failed", error = str(exc))

    # An automatic install whose detection now resolves elsewhere; nothing else surfaces it. Skipped while a job runs, and when an update is offered: that update re-detects.
    to_backend = (
        None
        if job_running or update_available or checks_disabled
        else _pending_backend_migration(binary, marker, force_refresh = force_refresh)
    )

    with _job_lock:
        job = dict(_job)

    return {
        "supported": bool(freshness.get("has_marker")),
        "update_available": update_available,
        "stale": bool(freshness.get("stale")),
        "installed_tag": installed,
        "latest_tag": latest,
        "published_repo": freshness.get("published_repo") or repo,
        "installed_at_utc": freshness.get("installed_at_utc"),
        "age_days": freshness.get("age_days"),
        "source_build": False,
        "update_size_bytes": update_size_bytes,
        "backend_migration_available": to_backend is not None,
        "from_backend": marker_backend(marker) if to_backend else None,
        "to_backend": to_backend,
        "job": job,
    }


_backends_memo: dict = {}


def _resolve_backends_for_host(
    install_dir: Optional[Path],
    *,
    force_refresh: bool = False,
    published_repo: Optional[str] = None,
    rocm_gfx: Optional[str] = None,
) -> Optional[dict]:
    """Ask the installer which backends it could install here. None on any failure. ``rocm_gfx`` replays the marker's arch as _run_llama_phase does: this subprocess re-probes from scratch, so a host whose arch only Windows setup could infer resolves "auto" to CPU. The REMEMBERED spelling, so a live probe still wins."""
    args: list[str] = []
    if install_dir is not None:
        args.extend(("--install-dir", str(install_dir)))
    if published_repo:
        args.extend(("--published-repo", published_repo))
    return _flow.resolve_prebuilt_for_host(
        force_refresh = force_refresh,
        memo = _backends_memo,
        installer_script = lambda: _installer_script(),
        log_message = "llama backend: resolve-backends failed",
        mode = ("--resolve-backends", "latest"),
        extra_args = tuple(args),
        extra_env = {"UNSLOTH_ROCM_GFX_REMEMBERED": rocm_gfx} if rocm_gfx else None,
    )


def _switch_support(binary: Optional[str], marker: Optional[dict]) -> Optional[str]:
    """Return why this install cannot be switched, or None if it can."""
    if _studio_custom_path_active():
        return "custom_path"
    if binary is None:
        return "not_installed"
    if _active_install_is_local_link(binary):
        return "local_link"
    if marker is None:
        return "source_build"
    if _install_dir_for(binary) is None:
        return "no_install_dir"
    return None


def _env_backend_override() -> Optional[str]:
    """Return the environment override that outranks the picker."""
    return environment_backend_override(
        os.environ.get("UNSLOTH_LLAMA_CPP_BACKEND"),
        os.environ.get("UNSLOTH_FORCE_VULKAN"),
    )


def _backend_options(resolved: Optional[dict], assets: Optional[dict] = None) -> list[dict]:
    """Normalize a resolver payload into the options the picker and the switch planner both read, so both judge a request against the same list."""
    options = []
    for entry in (resolved or {}).get("backends") or []:
        backend = normalize_backend(entry.get("backend"))
        if backend is None:
            continue
        asset = entry.get("asset")
        options.append(
            {
                "backend": backend,
                "available": bool(entry.get("available")),
                "unavailable_reason": entry.get("reason"),
                "resolved_backend": normalize_backend(entry.get("resolved_backend")),
                "release_tag": entry.get("release_tag"),
                "download_size_bytes": (assets or {}).get(asset) if asset else None,
            }
        )
    return options


def _selection_applied(
    backend_request: str, installed_backend: Optional[str], options: list[dict]
) -> bool:
    """Whether the recorded choice still describes the installed backend. A concrete choice used to be applied by definition, since the installer recorded it only on an install that honoured it; it now PRESERVES a request the install could not honour (#11143: a Vulkan choice that landed the ROCm bundle used to be erased to "auto", which destroyed the setting and made every later update re-detect), so a concrete choice is applied only when it is the backend that actually landed. Unknown installed backend contradicts nothing and stays applied. ``auto`` is the one that drifts (a GPU or driver appearing after an automatic CPU install makes detection resolve elsewhere) and re-applying it is offered exactly then."""
    if backend_request != "auto":
        return installed_backend is None or installed_backend == backend_request
    auto = next((option for option in options if option["backend"] == "auto"), None)
    if not auto or not auto.get("available"):
        return True
    resolved = auto.get("resolved_backend")
    return resolved is None or installed_backend is None or resolved == installed_backend


def _remembered_rocm_gfx(marker: Optional[dict]) -> Optional[str]:
    """The arch to replay when re-resolving, falling back to the installed asset's name. The marker carries rocm_gfx only when the install resolved one, and a host whose probe never named an arch is the one that needs the replay. The bundle it runs names the family, the record setup.sh reads too."""
    recorded = (marker or {}).get("rocm_gfx")
    if recorded:
        return recorded
    args = _flow.rocm_install_args((marker or {}).get("asset"))
    return args[1] if len(args) == 2 and args[0] == "--rocm-gfx" else None


def _pending_backend_migration(
    binary: Optional[str],
    marker: Optional[dict],
    *,
    force_refresh: bool = False,
) -> Optional[str]:
    """The backend a re-applied AUTOMATIC selection would install, when it differs: the one drift an install can develop untouched, since "auto" recorded what detection chose then and a default can move later (AMD integrated GPUs now route to Vulkan). Not computed for a concrete recorded choice, a decision by hand this must never offer to undo, which also makes the offer self-limiting since picking a backend in Settings ends the drift. None on anything unresolved. Shares _backends_memo with get_backend_status, so this costs a subprocess a day."""
    if marker is None or _switch_support(binary, marker) is not None:
        return None
    _override = _env_backend_override()
    if _override is not None and _override != "auto":
        # The environment owns the backend, so an offer could not be applied. "auto" is the exception: it asks for the detection the migration re-applies.
        return None
    if marker_backend_request(marker) != "auto":
        return None
    repo = marker.get("published_repo") or DEFAULT_PUBLISHED_REPO
    resolved = _resolve_backends_for_host(
        _install_dir_for(binary),
        force_refresh = force_refresh,
        published_repo = repo,
        rocm_gfx = _remembered_rocm_gfx(marker),
    )
    if not resolved:
        return None
    options = _backend_options(resolved)
    if _selection_applied("auto", marker_backend(marker), options):
        return None
    auto = next((option for option in options if option["backend"] == "auto"), None)
    target = (auto or {}).get("resolved_backend")
    if target == "cpu" and marker_backend(marker) not in (None, "cpu"):
        # An empty probe and a host that really lost its GPU look identical from here, and moving a working GPU install onto CPU is the costly side of that. Settings can.
        return None
    return target


def get_backend_status(*, force_refresh: bool = False) -> dict:
    """Return the installed backend and host-compatible alternatives."""
    binary = _find_binary()
    marker = read_install_marker(binary)
    unsupported = _switch_support(binary, marker)
    with _job_lock:
        job = dict(_job)
    status: dict = {
        "supported": unsupported is None,
        "reason": unsupported,
        "env_backend": _env_backend_override(),
        "backend": marker_backend(marker),
        "backend_request": marker_backend_request(marker),
        # Only the resolver can tell that an automatic choice has drifted, so an unresolved status must not invite an apply.
        "selection_applied": True,
        "installed_tag": (marker or {}).get("release_tag") or (marker or {}).get("tag"),
        "options": [],
        "job": job,
    }
    if unsupported is not None:
        return status
    if update_checks_disabled():
        status["supported"] = False
        status["reason"] = "update_checks_disabled"
        return status
    if job["state"] == _JOB_RUNNING:
        return status
    repo = (marker or {}).get("published_repo") or DEFAULT_PUBLISHED_REPO
    resolved = _resolve_backends_for_host(
        _install_dir_for(binary),
        force_refresh = force_refresh,
        published_repo = repo,
        # The same recovery the update-status path uses, or the picker describes a different host: no arch here reads an AMD box with no probes as CPU-only, so Settings offers an Automatic that installs something else.
        rocm_gfx = _remembered_rocm_gfx(marker),
    )
    if not resolved:
        status["reason"] = "unresolved"
        status["supported"] = False
        return status
    assets = None
    try:
        assets = latest_release_assets(repo, force_refresh = force_refresh)
    except Exception as exc:  # pragma: no cover - network defensive
        logger.debug("llama backend: asset size lookup failed", error = str(exc))
    options = _backend_options(resolved, assets)
    status["options"] = options
    status["selection_applied"] = _selection_applied(
        status["backend_request"], status["backend"], options
    )
    return status


def _run_llama_phase(
    install_dir: Path,
    repo: str,
    asset: Optional[str],
    script: Path,
    pin_release_tag: Optional[str],
    set_progress,
    llama_backend: Optional[str] = None,
    rocm_gfx: Optional[str] = None,
    backend_request: Optional[str] = None,
    migration_target: Optional[str] = None,
) -> dict:
    """The llama phase of a chained update: put the backend into a maintenance state, run the installer for the latest prebuilt, then refresh caches so the next load uses the new build. Returns {to_tag, reload_required, message}; raises on failure. pin_release_tag pins the installer to that exact published release instead of letting it re-resolve "latest" (see start_update)."""
    backend = None
    marked = []
    model_was_active = False
    mtmd_guard = ExitStack()
    # The installer exits 0 for a transient failure it answered by keeping the tree, so success no longer implies a new release. Read as the post-install check reads it.
    prior_marker = read_install_marker(_find_binary())
    prior_tag = (prior_marker or {}).get("release_tag") or (prior_marker or {}).get("tag")
    try:
        try:
            from routes.inference import get_llama_cpp_backend
            backend = get_llama_cpp_backend()
        except Exception as exc:
            logger.debug(
                "llama update: backend unavailable, skipping load coordination", error = str(exc)
            )
            backend = None

        if backend is not None:
            try:
                with backend._serial_load_lock:
                    backend._llama_update_in_progress = True
                    if getattr(backend, "is_active", False):
                        model_was_active = True
                        backend.unload_model()
            except Exception as exc:
                logger.debug("llama update: load coordination failed", error = str(exc))
        from core.inference import model_slots

        # Each kept model's in-flight load drains under its own lock, as the primary's did above.
        for slot in list(model_slots.slots):
            with slot.llama._serial_load_lock:
                slot.llama._llama_update_in_progress = True
            marked.append(slot.llama)
        try:
            if model_slots.unload_llama_slots(strict = True):
                model_was_active = True
        except RuntimeError:
            # A kept server that survived would run from, or lock, the tree being replaced.
            model_was_active = True
            raise

        # The mtmd dictation sidecar serves Qwen3-ASR from this same llama-server out of this same tree, so a live one locks the exe on Windows and a concurrent load would start against a half-swapped install.
        model_was_active = _block_mtmd_sidecar(mtmd_guard) or model_was_active
        model_was_active = _block_prompt_upsampler(mtmd_guard) or model_was_active

        cmd = [
            sys.executable,
            str(script),
            "--install-dir",
            str(install_dir),
            "--llama-tag",
            "latest",
            "--published-repo",
            repo,
        ]
        if pin_release_tag:
            cmd.extend(["--published-release-tag", pin_release_tag])
        cmd.extend(_rocm_install_args(asset))
        if backend_request is not None:
            cmd.extend(["--llama-backend", backend_request])
        logger.info("llama update: installing", cmd = " ".join(cmd))
        env = dict(os.environ, UNSLOTH_PROGRESS_PERCENT_STEP = "5")
        if backend_request is not None:
            env.pop("UNSLOTH_FORCE_VULKAN", None)
            env["UNSLOTH_LLAMA_CPP_BACKEND"] = backend_request
        # A Vulkan asset name carries no arch, so the marker is the only record of the gfx an automatic AMD route used. Automatic only: elsewhere the asset has the arch and replaying it would assert ROCm on a host whose AMD GPU is gone. Advisory even then, applied only if this host's own probe finds none.
        if rocm_gfx and llama_backend == "auto":
            env["UNSLOTH_ROCM_GFX_REMEMBERED"] = rocm_gfx
        _flow.stream_installer(
            cmd,
            env,
            set_progress = set_progress,
            timeout_seconds = _INSTALL_TIMEOUT_SECONDS,
        )

        # Drop stale caches so the banner re-checks the swapped marker. If GitHub is offline, latest stays unknown and the banner fails open.
        reset_caches(drop_disk = True)
        _backends_memo.clear()
        try:
            latest_published_release(repo, force_refresh = True)
        except Exception as exc:  # pragma: no cover - network defensive
            logger.debug("llama update: post-install freshness refresh failed", error = str(exc))
        new_marker = read_install_marker(_find_binary())
        new_tag = (new_marker or {}).get("release_tag") or (new_marker or {}).get("tag")
        new_backend = marker_backend(new_marker)
        new_backend_request = marker_backend_request(new_marker)

        new_repo = (new_marker or {}).get("published_repo")
        if pin_release_tag and backend_request is not None:
            if new_repo != repo or new_tag != pin_release_tag:
                raise RuntimeError(
                    "backend switch must preserve "
                    f"{repo}@{pin_release_tag}, but installer produced "
                    f"{new_repo or 'an unknown repository'}@{new_tag or 'an unknown release'}"
                )
        elif pin_release_tag and new_tag and new_repo == repo and new_tag != pin_release_tag:
            raise RuntimeError(f"pinned release {pin_release_tag} but installer produced {new_tag}")

        if backend_request is not None:
            if new_backend is None:
                raise RuntimeError(
                    f"requested {backend_request} but the installed backend is unknown"
                )
            if new_backend_request != backend_request:
                raise RuntimeError(
                    f"requested {backend_request} but the installer recorded "
                    f"{new_backend_request or 'an unknown selection'}"
                )
            if backend_request != "auto" and new_backend != backend_request:
                raise RuntimeError(
                    f"requested {backend_request} but the installer produced "
                    f"{new_backend or 'an unknown backend'}"
                )

        kept_existing = backend_request is None and new_tag is not None and new_tag == prior_tag
        # A migration asks for "auto", so it can land back where it started (the ROCm fallback behind the preference), which would read as applied and be re-offered.
        migration_kept = bool(migration_target) and new_backend != migration_target
        logger.info(
            "llama update: success",
            to_tag = new_tag,
            backend = new_backend,
            kept_existing = kept_existing,
            migration_applied = None if not migration_target else not migration_kept,
        )
        reload_hint = " Reload your model to use it." if model_was_active else ""
        if migration_kept:
            message = (
                f"llama.cpp could not be moved to {migration_target} right now, so the "
                f"existing {new_backend or 'installed'} build was kept. Try again later."
            )
        elif backend_request is not None:
            message = f"llama.cpp is now running on {new_backend or backend_request}.{reload_hint}"
        elif kept_existing:
            # The phase only runs when a newer release was offered, so naming the kept release as current would send the user looking for a fix it does not have.
            message = (
                f"llama.cpp could not be updated right now, so the existing {new_tag} "
                "install was kept. Try again later."
            )
        else:
            message = f"Updated llama.cpp to {new_tag}.{reload_hint}"
        return {
            "to_tag": new_tag,
            "backend": new_backend,
            "reload_required": model_was_active,
            "kept_existing": kept_existing,
            "migration_applied": None if not migration_target else not migration_kept,
            "message": message,
        }
    except _flow.InstallerExit as exc:
        if exc.returncode == _EXIT_NO_SPACE:
            logger.warning("llama update: out of disk space")
            raise _LlamaPhaseError(
                "Not enough disk space to install llama.cpp. Free up space or point "
                "UNSLOTH_STUDIO_HOME/TMPDIR at a larger volume, then retry.",
                reload_required = model_was_active,
            ) from exc
        if exc.returncode == _EXIT_BACKEND_UNAVAILABLE:
            failed_backend = backend_request or _env_backend_override()
            # "requested" reads as the user's choice; an automatic update has none.
            backend_label = (
                f"{failed_backend} "
                if failed_backend is not None and failed_backend != "auto"
                else ""
            )
            # The reason, as every sibling branch does: otherwise the record is "null".
            logger.warning(
                "llama update: backend unavailable",
                backend = failed_backend,
                error = str(exc),
            )
            raise _LlamaPhaseError(
                f"Could not install a {backend_label}llama.cpp build on this machine. "
                "The installed backend was kept.",
                reload_required = model_was_active,
            ) from exc
        if exc.returncode == _EXIT_FALLBACK:
            message = str(exc)
            # Same predicate the formatter uses: exit 2 also carries failures like a rate-limited huggingface.co validation-model fetch, which GH_TOKEN cannot fix.
            if _flow.is_github_rate_limit_text(message):
                token_present = _flow.github_token_present(env)
                logger.warning("llama update: GitHub rate limit", authenticated = token_present)
                advice = (
                    "Wait for the limit to reset and try again."
                    if token_present
                    else "Set GH_TOKEN or GITHUB_TOKEN in your environment and try again."
                )
                raise _LlamaPhaseError(
                    "Could not update llama.cpp: GitHub is rate-limiting release downloads. "
                    f"{advice}",
                    reload_required = model_was_active,
                ) from exc
            logger.warning("llama update: prebuilt fallback", error = message)
            detail = message.split(": ", 1)[-1] if ": " in message else message
            raise _LlamaPhaseError(
                f"Could not update llama.cpp from the prebuilt bundle. {detail}",
                reload_required = model_was_active,
            ) from exc
        logger.warning("llama update: failed", error = str(exc))
        raise _LlamaPhaseError(str(exc), reload_required = model_was_active) from exc
    except Exception as exc:
        logger.warning("llama update: failed", error = str(exc))
        if isinstance(exc, _LlamaPhaseError):
            raise
        if model_was_active:
            raise _LlamaPhaseError(str(exc), reload_required = True) from exc
        raise
    finally:
        mtmd_guard.close()
        # A kept slot serving a non-GGUF model stays loaded through the update; its llama backend
        # must not stay refused once the update is done.
        for llama in (backend, *marked):
            if llama is not None:
                try:
                    llama._llama_update_in_progress = False
                except Exception:  # pragma: no cover - defensive
                    pass


def _block_mtmd_sidecar(stack: ExitStack) -> bool:
    """Hold the mtmd sidecar's maintenance guard for the install, if it exists. Unlike whisper.cpp this is not fail-closed: llama.cpp updates predate this sidecar and must keep working where dictation cannot even be imported. Returns whether a warm dictation server had to be unloaded."""
    try:
        from core.inference.stt_mtmd_sidecar import get_mtmd_stt_sidecar
        return stack.enter_context(get_mtmd_stt_sidecar().update_maintenance())
    except Exception as exc:  # noqa: BLE001 - the update proceeds without it
        logger.debug("llama update: mtmd coordination failed", error = str(exc))
        return False


def _block_prompt_upsampler(stack: ExitStack) -> bool:
    """The FLUX.3 prompt upsampler runs the same llama-server; hold its maintenance guard for the install, only
    if Images ever loaded the module. Not fail-closed, like the mtmd guard."""
    module = sys.modules.get("core.inference.flux3_prompt_upsampler")
    if module is None:
        return False
    try:
        return stack.enter_context(module.get_sidecar().update_maintenance())
    except Exception as exc:  # noqa: BLE001 - the update proceeds without it
        logger.debug("llama update: prompt upsampler coordination failed", error = str(exc))
        return False


# Combined-job progress split when every phase runs (the llama bundle dwarfs the others); normalized to 0..1 over the phases that run.
_LLAMA_PHASE_WEIGHT = 0.7
_WHISPER_PHASE_WEIGHT = 0.3
_AUDIO_PHASE_WEIGHT = 0.3


def _plan_llama_phase(backend_request: Optional[str] = None) -> dict:
    """Plan the llama phase for an update or backend switch."""
    binary = _find_binary()
    if _studio_custom_path_active():
        return {
            "skip_reason": "custom_path",
            "refusal": {
                "started": False,
                "reason": "custom_path",
                "message": (
                    "llama.cpp is using the custom folder selected in Settings; "
                    "Unsloth won't replace it. Update that build yourself or restore "
                    "the bundled runtime first."
                ),
            },
        }
    # Refuse to update a --with-llama-cpp-dir local link: installing a prebuilt here would write through the link into the user's own checkout (or fail) and silently drop the link.
    if _active_install_is_local_link(binary):
        return {
            "skip_reason": "local_link",
            "refusal": {
                "started": False,
                "reason": "local_link",
                "message": (
                    "llama.cpp is a local directory linked with --with-llama-cpp-dir; "
                    "Unsloth won't replace it. Update your own llama.cpp checkout instead."
                ),
            },
        }
    marker = read_install_marker(binary)
    script = _installer_script()
    if script is None:
        return {
            "skip_reason": "installer_missing",
            "refusal": {
                "started": False,
                "reason": "installer_missing",
                "message": "install_llama_prebuilt.py could not be located.",
            },
        }

    if marker:
        # Mirror the detection guard: a direct POST or a stale banner must not start an install when the latest is not actually newer (force a fresh check so a stale 24h cache cannot wrongly block a real update either). A switch replaces the backend at the same release.
        migration = False
        migration_target = None
        status = (
            {}
            if backend_request is not None
            else _llama_only_status(
                force_refresh = True,
                allow_source_probe_while_running = True,
            )
        )
        if backend_request is None and not status.get("update_available"):
            # Nothing newer to install, but "auto" can resolve elsewhere now. Re-asking for "auto" keeps it automatic, so rocm_gfx and the crash recovery survive.
            migration_target = _pending_backend_migration(binary, marker)
            if migration_target is None:
                return {
                    "skip_reason": "up_to_date",
                    "refusal": {
                        "started": False,
                        "reason": "up_to_date",
                        "message": (
                            "The installed llama.cpp build is already at the latest prebuilt."
                        ),
                    },
                }
            backend_request = "auto"
            migration = True
        install_dir = _install_dir_for(binary)
        repo = marker.get("published_repo") or DEFAULT_PUBLISHED_REPO
        from_tag = marker.get("tag") or marker.get("release_tag")
        asset = marker.get("asset")
        llama_backend = marker.get("llama_backend")
        # Recovered, not read: the status path offers the migration through _remembered_rocm_gfx, so an apply reading the bare field would re-resolve without the arch on exactly the host the recovery exists for, and refuse the offer it just made as already_selected.
        rocm_gfx = _remembered_rocm_gfx(marker)
        # Install exactly the release the banner offered: the installer's own "latest" is commit-date ordered and can lag the published_at pick above, reinstalling the current build in a loop (the #6219 class). Not on macOS, which needs the older-release walk-back a pin disables; elsewhere an unusable latest fails the job loudly (retryable) instead of walking back. Switch only the backend: slim whisper bundles require this exact release.
        installed_release_tag = marker.get("release_tag") or marker.get("tag")
        wanted_tag = (
            installed_release_tag if backend_request is not None else status.get("latest_tag")
        )
        pin_release_tag = None if sys.platform == "darwin" else wanted_tag
    elif backend_request is not None:
        return {
            "skip_reason": "not_prebuilt",
            "refusal": {
                "started": False,
                "reason": "not_prebuilt",
                "message": (
                    "This llama.cpp install was not made from an Unsloth prebuilt, so "
                    "its backend cannot be switched from here."
                ),
            },
        }
    else:
        # Source build / custom path: only proceed when the same detection logic would offer the update (prebuilt exists, install is behind, root is manageable), so a direct POST cannot downgrade a newer source build.
        src = _source_build_status(binary, force_refresh = True) if binary else None
        if src is None:
            return {
                "skip_reason": "no_prebuilt_available",
                "refusal": {
                    "started": False,
                    "reason": "no_prebuilt_available",
                    "message": (
                        "No official llama.cpp prebuilt is available for this host, "
                        "so the source build cannot be swapped automatically."
                    ),
                },
            }
        if not src.get("update_available"):
            return {
                "skip_reason": "up_to_date",
                "refusal": {
                    "started": False,
                    "reason": "up_to_date",
                    "message": (
                        "The installed llama.cpp build is already at or newer than the "
                        "latest prebuilt."
                    ),
                },
            }
        res = _resolve_prebuilt_for_host()
        install_dir = _llama_install_root(binary)
        repo = (res or {}).get("repo") or DEFAULT_PUBLISHED_REPO
        from_tag = None
        asset = (res or {}).get("asset")
        llama_backend = None
        rocm_gfx = None
        # No pin: source-build detection resolves via --resolve-prebuilt latest, the same resolver the unpinned apply uses, so the two already agree.
        pin_release_tag = None
        migration = False
        migration_target = None

    if install_dir is None:
        return {
            "skip_reason": "no_install_dir",
            "refusal": {
                "started": False,
                "reason": "no_install_dir",
                "message": "Could not determine the llama.cpp install directory.",
            },
        }
    return {
        "spec": {
            "install_dir": install_dir,
            "repo": repo,
            "asset": asset,
            "script": script,
            "pin_release_tag": pin_release_tag,
            "from_tag": from_tag,
            "llama_backend": llama_backend,
            "rocm_gfx": rocm_gfx,
            "backend_request": backend_request,
            # So the caller keeps presenting this as the update it is, not the switch its backend_request would make it look like.
            "migration": migration,
            "migration_target": migration_target,
        }
    }


def start_update() -> dict:
    """Kick off a background update job chaining the llama phase with whisper and audio.cpp phases that run only when their component is behind; each phase no-ops cleanly when its component is current or unmanaged. Idempotent: a second call while one is running returns the in-flight job."""
    if update_checks_disabled():
        with _job_lock:
            job = dict(_job)
        return {
            "started": False,
            "reason": "update_checks_disabled",
            "message": (
                "Update checks are disabled (UNSLOTH_DISABLE_UPDATE_CHECK=1). "
                "Unset it and restart Unsloth to update llama.cpp from here."
            ),
            "job": job,
        }
    return _start_llama_job()


def start_backend_switch(backend: str) -> dict:
    """Install the llama.cpp bundle for ``backend`` and record it as the choice. Deliberately the same job, lock and phase machinery as an update: both replace the same tree while the same server is unloaded, which is what makes a switch and an update mutually exclusive instead of two writers racing over one install. Callers poll get_update_status() exactly as for an update."""
    normalized = normalize_backend(backend)
    if normalized is None or normalized not in REQUESTABLE_BACKENDS:
        with _job_lock:
            job = dict(_job)
        return {
            "started": False,
            "reason": "unknown_backend",
            "message": f"{backend!r} is not a llama.cpp backend Unsloth can install.",
            "job": job,
        }
    env_backend = _env_backend_override()
    if env_backend is not None:
        with _job_lock:
            job = dict(_job)
        return {
            "started": False,
            "reason": "environment_override",
            "message": (
                f"llama.cpp is controlled by the {env_backend} environment override. "
                "Unset it and restart Unsloth before switching backends here."
            ),
            "job": job,
        }

    return _start_llama_job(backend_request = normalized)


def _update_can_move_the_backend(llama_will_run: bool) -> bool:
    """Whether an update that names no backend could still land on a different one. Detection re-runs, so a default can move under it; only an automatic install is exposed. Read from the marker rather than resolved, since resolving costs a subprocess at planning time and over-answering is cheap: the repair phase does nothing when the backend did not move."""
    if not llama_will_run:
        return False
    return marker_backend_request(read_install_marker(_find_binary())) == "auto"


def _repair_pairing_plan_or_empty(llama_will_run: bool, llama_skip_reason: Optional[str]) -> dict:
    """The slim re-pair plan, or {} when there is nothing to re-pair against."""
    try:
        from utils import whisper_cpp_update
        if not llama_will_run:
            if llama_skip_reason != "already_selected":
                return {}
            if not whisper_cpp_update.slim_pairing_is_stale():
                return {}
        return whisper_cpp_update.repair_pairing_plan()
    except Exception as exc:  # pragma: no cover - defensive
        logger.debug("llama switch: whisper repair probe failed", error = str(exc))
        return {}


def _whisper_phase_plan(
    backend_request: Optional[str],
    *,
    llama_will_run: bool,
    llama_skip_reason: Optional[str] = None,
    migration: bool = False,
) -> dict:
    """Whisper's half of the chained job: catch up on releases for an update, or re-pair with the new backend for a switch. A switch that installs nothing leaves llama's ggml where it was, so there is nothing to re-pair and a refused switch stays refused instead of becoming a whisper-only job. The exception is what a failed re-pair leaves behind: llama runs first and records the new backend, so a retryable whisper failure ends with llama on the requested backend and dictation still hardlinked to the old runtime, making a retry ``already_selected`` and stranding the user unless they switch away and back. So allow a repair-only job for that one refusal, and only while the pairing is genuinely stale."""
    # A migration is switch-shaped but update-behaved, so whisper needs the chained plan: the repair-only branch has no phase for a self-contained install and drops updates.
    if backend_request is None or migration:
        chained = (
            _whisper_chain_status(force_refresh = True, paired_llama_will_update = llama_will_run) or {}
        )
        if chained.get("phase") is not None:
            return chained
        if backend_request is None:
            if not _update_can_move_the_backend(llama_will_run):
                return chained
            # Whisper has nothing to catch up on, so the chained answer stands unless a re-pair is owed.
            repair = _repair_pairing_plan_or_empty(llama_will_run, llama_skip_reason)
            return repair if repair.get("phase") is not None else chained
    return _repair_pairing_plan_or_empty(llama_will_run, llama_skip_reason)


def _claim_operation(backend_request: Optional[str]) -> bool:
    """Reserve the operation and publish its planning state atomically."""
    with _job_lock:
        if not _operation_lock.acquire(blocking = False):
            return False
        if _job["state"] == _JOB_RUNNING:
            _operation_lock.release()
            return False
        _job.update(_flow.new_job())
        _job.update(
            state = _JOB_RUNNING,
            operation = "switch" if backend_request is not None else "update",
            requested_backend = backend_request,
            message = (
                f"Checking the {backend_request} llama.cpp build..."
                if backend_request is not None
                else "Checking for llama.cpp updates..."
            ),
            progress = 0.0,
            started_at = _utcnow(),
        )
    return True


def _finish_planning_refusal(reason: str, message: str) -> dict:
    """Publish a terminal result for an operation that stopped during planning."""
    succeeded = reason in {"up_to_date", "already_selected"}
    with _job_lock:
        _job.update(
            state = _JOB_SUCCESS if succeeded else _JOB_ERROR,
            message = message,
            error = None if succeeded else message,
            progress = 1.0 if succeeded else None,
            finished_at = _utcnow(),
        )
        job = dict(_job)
    return {"started": False, "reason": reason, "message": message, "job": job}


def _run_claimed_job(phases: list[dict]) -> None:
    """Run a planned job and always release its full-operation reservation."""
    try:
        _flow.run_chained_update(phases, job = _job, job_lock = _job_lock)
    except Exception as exc:  # pragma: no cover - phase failures are handled inside the runner
        logger.exception("llama update: job runner failed", error = str(exc))
        with _job_lock:
            _job.update(
                state = _JOB_ERROR,
                message = "llama.cpp install failed.",
                error = str(exc),
                finished_at = _utcnow(),
            )
    finally:
        _operation_lock.release()


def _start_llama_job(backend_request: Optional[str] = None) -> dict:
    """Shared body of start_update() and start_backend_switch()."""
    if not _claim_operation(backend_request):
        with _job_lock:
            return {
                "started": False,
                "reason": "already_running",
                "message": _ALREADY_RUNNING_MESSAGE,
                "job": dict(_job),
            }

    handed_to_worker = False
    migration = False
    try:
        # The operation reservation starts before any marker read or resolver call, so a second update/switch cannot change the install between this plan and the worker that applies it.
        llama_plan = _plan_llama_phase(backend_request)
        llama_spec = llama_plan.get("spec")
        if llama_spec is not None and llama_spec.get("migration"):
            # The planner turned an up-to-date update into a re-application of "auto"; adopt its request, and keep the claimed operation the offered "update".
            migration = True
            backend_request = llama_spec["backend_request"]
            with _job_lock:
                _job.update(requested_backend = backend_request)
        if backend_request is not None and llama_spec is not None:
            resolved = _resolve_backends_for_host(
                llama_spec["install_dir"],
                force_refresh = True,
                published_repo = llama_spec["repo"],
                # Same replay the status path uses: without it this re-probe resolves "auto" to CPU and the offered migration refuses as already_selected.
                rocm_gfx = llama_spec.get("rocm_gfx"),
            )
            if not resolved:
                return _finish_planning_refusal(
                    "unresolved",
                    "Could not verify the available llama.cpp backends. Try again online.",
                )
            options = _backend_options(resolved)
            option = next((o for o in options if o["backend"] == backend_request), None)
            if (
                option is None
                or not option["available"]
                or (backend_request != "auto" and option["resolved_backend"] != backend_request)
            ):
                return _finish_planning_refusal(
                    "backend_unavailable",
                    f"No {backend_request} llama.cpp build is available for this machine.",
                )
            marker = read_install_marker(_find_binary())
            if marker_backend_request(marker) == backend_request and _selection_applied(
                backend_request, marker_backend(marker), options
            ):
                llama_plan = {
                    "skip_reason": "already_selected",
                    "refusal": {
                        "started": False,
                        "reason": "already_selected",
                        "message": f"llama.cpp is already set to {backend_request}.",
                    },
                }
                llama_spec = None

        whisper_plan = _whisper_phase_plan(
            backend_request,
            llama_will_run = llama_spec is not None,
            llama_skip_reason = llama_plan.get("skip_reason"),
            migration = migration,
        )
        whisper_spec = (whisper_plan or {}).get("phase")
        # audio.cpp ships its own ggml, so a backend switch never touches it.
        audio_plan = (_audio_chain_status() or {}) if backend_request is None or migration else {}
        audio_spec = audio_plan.get("phase")
        if llama_spec is None and whisper_spec is None and audio_spec is None:
            # Nothing to run: answer with the llama refusal so the existing reasons (local_link / up_to_date / already_selected / ...) keep their meaning.
            refusal = llama_plan["refusal"]
            return _finish_planning_refusal(refusal["reason"], refusal["message"])

        whisper_run = None
        if whisper_spec is not None:
            from utils import whisper_cpp_update as _whisper
            whisper_run = (
                (lambda set_progress: _whisper.run_repair_phase(whisper_spec, set_progress))
                if whisper_spec.get("repair")
                else (
                    lambda set_progress: _whisper.run_chained_phase_after_llama(
                        whisper_spec, set_progress
                    )
                )
                if llama_spec is not None
                else (lambda set_progress: _whisper.run_chained_phase(whisper_spec, set_progress))
            )

        audio_run = None
        if audio_spec is not None:
            from utils import audio_cpp_update as _audio
            audio_run = lambda set_progress: _audio.run_chained_phase(audio_spec, set_progress)

        phases = [
            {
                "name": "llama",
                "weight": _LLAMA_PHASE_WEIGHT,
                "failure_message": (
                    f"Could not switch llama.cpp to {backend_request}."
                    if backend_request is not None and not migration
                    else "llama.cpp update failed."
                ),
                "skip_reason": llama_plan.get("skip_reason"),
                "run": (
                    (
                        lambda set_progress: _run_llama_phase(
                            llama_spec["install_dir"],
                            llama_spec["repo"],
                            llama_spec["asset"],
                            llama_spec["script"],
                            llama_spec["pin_release_tag"],
                            set_progress,
                            llama_backend = llama_spec.get("llama_backend"),
                            rocm_gfx = llama_spec.get("rocm_gfx"),
                            backend_request = llama_spec.get("backend_request"),
                            migration_target = llama_spec.get("migration_target"),
                        )
                    )
                    if llama_spec
                    else None
                ),
            },
            {
                "name": "whisper",
                "weight": _WHISPER_PHASE_WEIGHT,
                "failure_message": (
                    "whisper.cpp could not be re-paired with the new backend."
                    if backend_request is not None
                    else "whisper.cpp update failed."
                ),
                # The sidecar reload is whisper-internal; it must not trip the job-level reload flag the chat frontend resyncs on.
                "affects_job_reload": False,
                "skip_reason": (whisper_plan or {}).get("skip_reason") or "unavailable",
                "run": whisper_run,
            },
            {
                "name": "audio",
                "weight": _AUDIO_PHASE_WEIGHT,
                "failure_message": "audio.cpp update failed.",
                # Unloading a main-slot audio.cpp model is a server model change the UI must resync on.
                "affects_job_reload": True,
                "reports_job_tag": False,
                "skip_reason": audio_plan.get("skip_reason") or "unavailable",
                "run": audio_run,
            },
        ]
        running = " + ".join(
            name
            for name, spec in (
                ("llama.cpp", llama_spec),
                ("whisper.cpp", whisper_spec),
                ("audio.cpp", audio_spec),
            )
            if spec
        )
        if backend_request is not None and llama_spec is not None:
            starting_message = f"Installing the {backend_request} llama.cpp build..."
        elif backend_request is not None:
            starting_message = "Re-pairing whisper.cpp with llama.cpp..."
        else:
            starting_message = f"Downloading and installing the latest {running} prebuilt..."

        with _job_lock:
            _job.update(
                message = starting_message,
                from_tag = (llama_spec or {}).get("from_tag"),
            )
            job_snapshot = dict(_job)

        thread = threading.Thread(
            target = _run_claimed_job,
            args = (phases,),
            name = "llama-cpp-backend-switch" if backend_request else "llama-cpp-update",
            daemon = True,
        )
        try:
            thread.start()
        except Exception as exc:  # pragma: no cover - interpreter resource failure
            with _job_lock:
                _job.update(
                    state = _JOB_ERROR,
                    message = "Could not start the llama.cpp install worker.",
                    error = str(exc),
                    finished_at = _utcnow(),
                )
                failed_job = dict(_job)
            return {
                "started": False,
                "reason": "worker_start_failed",
                "message": "Could not start the llama.cpp install worker.",
                "job": failed_job,
            }
        handed_to_worker = True
        return {"started": True, "reason": None, "job": job_snapshot}
    except Exception as exc:
        with _job_lock:
            _job.update(
                state = _JOB_ERROR,
                message = "Could not plan the llama.cpp install.",
                error = str(exc),
                finished_at = _utcnow(),
            )
        raise
    finally:
        if not handed_to_worker:
            _operation_lock.release()


def _reset_job_for_tests() -> None:
    """Test-only: return the job tracker to idle."""
    _flow.reset_job(_job, _job_lock)