unsloth/studio/backend/tests/test_llama_cpp_max_context_threshold.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

"""Tests for the ``max_context_length`` warning-threshold semantics.

The ctx slider in the chat settings sheet reads
``/api/inference/status.max_context_length`` to decide when to render the
"Exceeds estimated VRAM capacity. The model may use system RAM." warning:

    ctxDisplayValue > ggufMaxContextLength → show warning

When weights fit on some GPU subset, the threshold is the largest ctx that
fits fully in VRAM (the binary-search cap from ``_fit_context_to_vram``).
When weights exceed 90% of every GPU subset's free memory, the warning must
fire as soon as the user drags above what Auto itself selects (otherwise
loading e.g. MiniMax-M2.7 on a 97 GB GPU shows a slider up to 196608 with no
hint that any larger value triggers ``--fit on`` and degrades performance).

The threshold therefore tracks ``_AUTO_OFFLOAD_CTX`` and is not a literal.
Anchoring it below that constant is worse than having no warning: Auto's own
context then exceeds the ceiling Auto published, so every load in this branch
warns about itself while advising the user to leave it on Auto.

These tests pin both cases. No GPU probing, subprocess, or GGUF I/O.
Cross-platform: Linux, macOS, Windows, WSL.
"""

from __future__ import annotations

import sys
import types as _types
from pathlib import Path

import pytest

# Stub heavy / unavailable deps before importing the module under test.
# Same pattern as test_kv_cache_estimation.py.

_BACKEND_DIR = str(Path(__file__).resolve().parent.parent)
if _BACKEND_DIR not in sys.path:
    sys.path.insert(0, _BACKEND_DIR)

# loggers
_loggers_stub = _types.ModuleType("loggers")
_loggers_stub.get_logger = lambda name: __import__("logging").getLogger(name)
sys.modules.setdefault("loggers", _loggers_stub)

# structlog
_structlog_stub = _types.ModuleType("structlog")
sys.modules.setdefault("structlog", _structlog_stub)

# httpx
_httpx_stub = _types.ModuleType("httpx")
for _exc_name in (
    "ConnectError",
    "TimeoutException",
    "ReadTimeout",
    "ReadError",
    "RemoteProtocolError",
    "CloseError",
):
    setattr(_httpx_stub, _exc_name, type(_exc_name, (Exception,), {}))


class _FakeTimeout:
    def __init__(self, *a, **kw):
        pass


_httpx_stub.Timeout = _FakeTimeout
_httpx_stub.Client = type(
    "Client",
    (),
    {
        "__init__": lambda self, **kw: None,
        "__enter__": lambda self: self,
        "__exit__": lambda self, *a: None,
    },
)
# Only when the real library is absent. sys.modules holds what has been IMPORTED, not
# what is installed, so setdefault does not defer to a real httpx that nothing in this
# process has touched yet: the stub wins and shadows it for the whole session. This stub
# has no Response, and starlette.testclient reads httpx.Response at import, so every
# module collected afterwards that reaches fastapi.testclient or routes.inference dies.
try:
    import httpx  # noqa: F401
except ImportError:
    sys.modules.setdefault("httpx", _httpx_stub)

from core.inference.llama_cpp import (
    _AUTO_OFFLOAD_CTX,
    _CTX_FIT_VRAM_FRACTION,
    LlamaCppBackend,
)


# Helpers

GIB = 1024**3


def _make_backend(native_ctx = 131072):
    inst = LlamaCppBackend.__new__(LlamaCppBackend)
    inst._context_length = native_ctx
    inst._n_layers = 80
    inst._n_kv_heads = 8
    inst._n_heads = 64
    inst._embedding_length = 8192
    inst._kv_key_length = 128
    inst._kv_value_length = 128
    inst._kv_lora_rank = None
    inst._sliding_window = None
    inst._sliding_window_pattern = None
    inst._ssm_inner_size = None
    inst._full_attention_interval = None
    inst._key_length_mla = None
    inst._n_kv_heads_by_layer = None
    inst._kv_key_length_swa = None
    inst._kv_value_length_swa = None
    return inst


def _compute_max_available_ctx(
    native_ctx,
    model_gib,
    gpus,
    kv_per_token_bytes = 325_000,
):
    """Run load_model's ceiling-probe block and return the final
    ``max_available_ctx`` the backend would assign to ``_max_context_length``.
    """
    inst = _make_backend(native_ctx = native_ctx)
    model_size = int(model_gib * GIB)

    inst._estimate_kv_cache_bytes = (
        lambda n, _t = None, **_kw: 0 if n <= 0 else n * kv_per_token_bytes
    )
    inst._can_estimate_kv = lambda: True

    context_length = inst._context_length
    effective_ctx = context_length
    max_available_ctx = context_length

    cache_type_kv = None
    native_ctx_for_cap = context_length

    ranked_for_cap = sorted(gpus, key = lambda g: g[1], reverse = True)
    best_cap = 0
    for n_gpus in range(1, len(ranked_for_cap) + 1):
        subset = ranked_for_cap[:n_gpus]
        pool_mib = sum(free for _, free in subset)
        capped = inst._fit_context_to_vram(
            native_ctx_for_cap,
            pool_mib,
            model_size,
            cache_type_kv,
        )
        kv = inst._estimate_kv_cache_bytes(capped, cache_type_kv)
        total_mib = (model_size + kv) / (1024 * 1024)
        if total_mib <= pool_mib * _CTX_FIT_VRAM_FRACTION:
            best_cap = max(best_cap, capped)
    if best_cap > 0:
        max_available_ctx = best_cap
    else:
        max_available_ctx = min(_AUTO_OFFLOAD_CTX, native_ctx_for_cap)

    return max_available_ctx


# Weights exceed every GPU subset's VRAM  (MiniMax-M2.7-like)


class TestMaxContextLengthForWeightsExceedVRAM:
    """UI ``max_context_length`` must fall back to the Auto offload context so
    the warning fires as soon as the user drags above what Auto selects.
    """

    def test_minimax_like(self):
        """131 GB weights, single 97 GB GPU, native ctx 196608."""
        got = _compute_max_available_ctx(
            native_ctx = 196608,
            model_gib = 131,
            gpus = [(0, 97_000)],
        )
        assert got == _AUTO_OFFLOAD_CTX

    def test_multi_gpu_all_subsets_fail(self):
        """400 GB weights across a 4x80 GB pool (320 GB total, still too small)."""
        got = _compute_max_available_ctx(
            native_ctx = 131072,
            model_gib = 400,
            gpus = [(0, 80_000), (1, 80_000), (2, 80_000), (3, 80_000)],
        )
        assert got == _AUTO_OFFLOAD_CTX

    def test_native_below_fallback_is_preserved(self):
        """If native ctx is itself below the fallback, don't advertise a larger
        value than the model supports."""
        got = _compute_max_available_ctx(
            native_ctx = 2048,
            model_gib = 200,
            gpus = [(0, 80_000)],
        )
        assert got == 2048


# Fittable models (regression guard)


class TestMaxContextLengthForFittableModels:
    """The existing best-cap behaviour must be unchanged."""

    def test_small_model_fits_easily(self):
        """8 GB model on 24 GB GPU: should auto-pick a large ctx."""
        got = _compute_max_available_ctx(
            native_ctx = 131072,
            model_gib = 8,
            gpus = [(0, 24_000)],
            kv_per_token_bytes = 8192,
        )
        assert got > _AUTO_OFFLOAD_CTX
        assert got <= 131072

    def test_medium_model_multi_gpu(self):
        """60 GB model split across 2 GPUs: picks a fitting ctx."""
        got = _compute_max_available_ctx(
            native_ctx = 131072,
            model_gib = 60,
            gpus = [(0, 40_000), (1, 40_000)],
            kv_per_token_bytes = 8192,
        )
        assert got > _AUTO_OFFLOAD_CTX

    def test_tiny_model_on_huge_gpu_near_native(self):
        """2 GB model, 80 GB GPU, negligible KV: should approach native."""
        got = _compute_max_available_ctx(
            native_ctx = 131072,
            model_gib = 2,
            gpus = [(0, 80_000)],
            kv_per_token_bytes = 64,
        )
        assert got >= 131072 - 256  # rounded to 256 boundary


# Property plumbing


class TestMaxContextLengthProperty:
    def test_falls_back_to_native_when_unset(self):
        inst = _make_backend(native_ctx = 131072)
        inst._max_context_length = None
        assert inst.max_context_length == 131072

    def test_returns_stored_value_when_set(self):
        inst = _make_backend(native_ctx = 131072)
        inst._max_context_length = 4096
        assert inst.max_context_length == 4096