Studio: queue local GGUF OpenAI-compatible requests before llama-server (#7047)

---------

Co-authored-by: oobabooga <112222186+oobabooga@users.noreply.github.com>

studio/backend/core/inference/llama_admission.py

@@ -0,0 +1,368 @@
+# SPDX-License-Identifier: AGPL-3.0-only
+# Copyright 2026-present the Unsloth AI Inc. team. All rights reserved.
+
+"""Admission control for local llama-server generation requests.
+
+The helpers in this module deliberately know nothing about FastAPI, SSE, or the
+OpenAI-compatible route shape. They only coordinate how many upstream generation
+requests may be active for one llama-server backend and provide a cancellable
+FIFO queue for excess requests.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import os
+import threading
+from collections import deque
+from dataclasses import dataclass
+from typing import Deque, Optional
+
+
+ADMISSION_CONTROL_ENV = "UNSLOTH_OPENAI_COMPAT_ADMISSION_CONTROL"
+ADMISSION_QUEUE_TIMEOUT_ENV = "UNSLOTH_OPENAI_COMPAT_ADMISSION_QUEUE_TIMEOUT"
+ADMISSION_KEEPALIVE_INTERVAL_ENV = "UNSLOTH_OPENAI_COMPAT_ADMISSION_KEEPALIVE_INTERVAL"
+ADMISSION_MAX_QUEUE_ENV = "UNSLOTH_OPENAI_COMPAT_ADMISSION_MAX_QUEUE"
+
+DEFAULT_ADMISSION_ENABLED = True
+DEFAULT_ADMISSION_QUEUE_TIMEOUT_S = None
+DEFAULT_ADMISSION_KEEPALIVE_INTERVAL_S = 5.0
+DEFAULT_ADMISSION_MAX_QUEUE = 64
+
+
+@dataclass(frozen = True)
+class LlamaAdmissionConfig:
+    enabled: bool = DEFAULT_ADMISSION_ENABLED
+    queue_timeout_s: Optional[float] = DEFAULT_ADMISSION_QUEUE_TIMEOUT_S
+    keepalive_interval_s: float = DEFAULT_ADMISSION_KEEPALIVE_INTERVAL_S
+    max_queue: Optional[int] = DEFAULT_ADMISSION_MAX_QUEUE
+
+
+@dataclass(frozen = True)
+class LlamaAdmissionSnapshot:
+    key: str
+    capacity: int
+    active: int
+    queued: int
+
+
+class LlamaAdmissionError(Exception):
+    def __init__(
+        self,
+        message: str,
+        *,
+        snapshot: Optional[LlamaAdmissionSnapshot] = None,
+    ):
+        super().__init__(message)
+        self.snapshot = snapshot
+
+
+class LlamaAdmissionQueueFull(LlamaAdmissionError):
+    pass
+
+
+class LlamaAdmissionTimeout(LlamaAdmissionError):
+    pass
+
+
+class LlamaAdmissionCancelled(LlamaAdmissionError):
+    pass
+
+
+def _bool_env(name: str, default: bool) -> bool:
+    value = os.environ.get(name)
+    if value is None or not value.strip():
+        return default
+    value = value.strip().lower()
+    if value in {"1", "true", "yes", "on"}:
+        return True
+    if value in {"0", "false", "no", "off"}:
+        return False
+    return default
+
+
+def _optional_positive_float_env(name: str, default: Optional[float]) -> Optional[float]:
+    value = os.environ.get(name)
+    if value is None or not value.strip():
+        return default
+    try:
+        parsed = float(value.strip())
+    except ValueError:
+        return default
+    return parsed if parsed > 0 else None
+
+
+def _positive_float_env(name: str, default: float) -> float:
+    value = os.environ.get(name)
+    if value is None or not value.strip():
+        return default
+    try:
+        parsed = float(value.strip())
+    except ValueError:
+        return default
+    return parsed if parsed > 0 else default
+
+
+def _optional_positive_int_env(name: str, default: Optional[int]) -> Optional[int]:
+    value = os.environ.get(name)
+    if value is None or not value.strip():
+        return default
+    try:
+        parsed = int(value.strip())
+    except ValueError:
+        return default
+    return parsed if parsed > 0 else None
+
+
+def llama_admission_config_from_env() -> LlamaAdmissionConfig:
+    return LlamaAdmissionConfig(
+        enabled = _bool_env(ADMISSION_CONTROL_ENV, DEFAULT_ADMISSION_ENABLED),
+        queue_timeout_s = _optional_positive_float_env(
+            ADMISSION_QUEUE_TIMEOUT_ENV,
+            DEFAULT_ADMISSION_QUEUE_TIMEOUT_S,
+        ),
+        keepalive_interval_s = _positive_float_env(
+            ADMISSION_KEEPALIVE_INTERVAL_ENV,
+            DEFAULT_ADMISSION_KEEPALIVE_INTERVAL_S,
+        ),
+        max_queue = _optional_positive_int_env(
+            ADMISSION_MAX_QUEUE_ENV,
+            DEFAULT_ADMISSION_MAX_QUEUE,
+        ),
+    )
+
+
+@dataclass
+class _Waiter:
+    loop: asyncio.AbstractEventLoop
+    future: asyncio.Future
+    cancelled: bool = False
+    granted_lease: Optional["LlamaAdmissionLease"] = None
+
+
+class LlamaAdmissionLease:
+    def __init__(self, queue: Optional["LlamaAdmissionQueue"]):
+        self._queue = queue
+        self._released = False
+        self._release_lock = threading.Lock()
+
+    def release(self) -> None:
+        queue = None
+        with self._release_lock:
+            if self._released:
+                return
+            self._released = True
+            queue = self._queue
+        if queue is not None:
+            queue.release()
+
+    async def __aenter__(self) -> "LlamaAdmissionLease":
+        return self
+
+    async def __aexit__(self, *_args) -> None:
+        self.release()
+
+
+class LlamaAdmissionReservation:
+    def __init__(
+        self,
+        *,
+        queue: Optional["LlamaAdmissionQueue"],
+        lease: Optional[LlamaAdmissionLease] = None,
+        waiter: Optional[_Waiter] = None,
+        snapshot: Optional[LlamaAdmissionSnapshot] = None,
+    ):
+        self._queue = queue
+        self._lease = lease
+        self._waiter = waiter
+        self.snapshot = snapshot
+
+    @property
+    def is_cancelled(self) -> bool:
+        return self._lease is None and self._waiter is None
+
+    def lease_nowait(self) -> Optional[LlamaAdmissionLease]:
+        if self._lease is not None:
+            return self._lease
+        if self._waiter is None or not self._waiter.future.done():
+            return None
+        if self._waiter.future.cancelled():
+            self._waiter.cancelled = True
+            self._waiter = None
+            return None
+        self._lease = self._waiter.future.result()
+        self._waiter = None
+        return self._lease
+
+    async def wait(self, timeout_s: float) -> Optional[LlamaAdmissionLease]:
+        lease = self.lease_nowait()
+        if lease is not None:
+            return lease
+        if self._waiter is None:
+            return None
+        waiter = self._waiter
+        try:
+            await asyncio.wait_for(asyncio.shield(waiter.future), timeout = timeout_s)
+        except asyncio.CancelledError:
+            if waiter.future.cancelled():
+                waiter.cancelled = True
+                if self._waiter is waiter:
+                    self._waiter = None
+                return None
+            raise
+        return self.lease_nowait()
+
+    def cancel(self) -> None:
+        lease = self.lease_nowait()
+        if lease is not None:
+            lease.release()
+            self._lease = None
+            return
+        if self._queue is not None and self._waiter is not None:
+            self._queue.cancel(self._waiter)
+        self._waiter = None
+
+    def snapshot_now(self) -> Optional[LlamaAdmissionSnapshot]:
+        if self._queue is None:
+            return self.snapshot
+        return self._queue.snapshot()
+
+
+class LlamaAdmissionQueue:
+    def __init__(self, key: str):
+        self.key = key
+        self._lock = threading.Lock()
+        self._active = 0
+        self._capacity = 1
+        self._waiters: Deque[_Waiter] = deque()
+
+    def reserve(self, *, capacity: int, config: LlamaAdmissionConfig) -> LlamaAdmissionReservation:
+        capacity = max(1, int(capacity or 1))
+        if not config.enabled:
+            return LlamaAdmissionReservation(
+                queue = None,
+                lease = LlamaAdmissionLease(None),
+                snapshot = LlamaAdmissionSnapshot(self.key, capacity, 0, 0),
+            )
+
+        loop = asyncio.get_running_loop()
+        with self._lock:
+            self._capacity = capacity
+            self._prune_waiters_locked()
+            self._grant_waiters_locked()
+            if self._active < self._capacity and not self._waiters:
+                self._active += 1
+                return LlamaAdmissionReservation(
+                    queue = self,
+                    lease = LlamaAdmissionLease(self),
+                    snapshot = self._snapshot_locked(),
+                )
+            if config.max_queue is not None and len(self._waiters) >= config.max_queue:
+                raise LlamaAdmissionQueueFull(
+                    "llama-server generation queue is full",
+                    snapshot = self._snapshot_locked(),
+                )
+            waiter = _Waiter(
+                loop = loop,
+                future = loop.create_future(),
+            )
+            self._waiters.append(waiter)
+            return LlamaAdmissionReservation(
+                queue = self,
+                waiter = waiter,
+                snapshot = self._snapshot_locked(),
+            )
+
+    def release(self) -> None:
+        with self._lock:
+            if self._active > 0:
+                self._active -= 1
+            self._grant_waiters_locked()
+
+    def cancel(self, waiter: _Waiter) -> None:
+        lease_to_release = None
+        with self._lock:
+            waiter.cancelled = True
+            try:
+                self._waiters.remove(waiter)
+            except ValueError:
+                pass
+            if waiter.granted_lease is not None:
+                lease_to_release = waiter.granted_lease
+                waiter.granted_lease = None
+            if not waiter.future.done():
+                waiter.loop.call_soon_threadsafe(waiter.future.cancel)
+        if lease_to_release is not None:
+            lease_to_release.release()
+
+    def snapshot(self) -> LlamaAdmissionSnapshot:
+        with self._lock:
+            self._prune_waiters_locked()
+            return self._snapshot_locked()
+
+    def is_idle(self) -> bool:
+        with self._lock:
+            self._prune_waiters_locked()
+            return self._active == 0 and not self._waiters
+
+    def _grant_waiters_locked(self) -> None:
+        self._prune_waiters_locked()
+        while self._waiters and self._active < self._capacity:
+            waiter = self._waiters.popleft()
+            if waiter.cancelled or waiter.future.done():
+                continue
+            self._active += 1
+            lease = LlamaAdmissionLease(self)
+            waiter.granted_lease = lease
+            waiter.loop.call_soon_threadsafe(self._deliver_lease, waiter, lease)
+
+    def _deliver_lease(self, waiter: _Waiter, lease: LlamaAdmissionLease) -> None:
+        if waiter.cancelled or waiter.future.done():
+            waiter.granted_lease = None
+            if not waiter.future.done():
+                waiter.future.cancel()
+            lease.release()
+            return
+        try:
+            waiter.future.set_result(lease)
+            waiter.granted_lease = None
+        except asyncio.InvalidStateError:
+            waiter.granted_lease = None
+            lease.release()
+
+    def _prune_waiters_locked(self) -> None:
+        self._waiters = deque(
+            waiter for waiter in self._waiters if not waiter.cancelled and not waiter.future.done()
+        )
+
+    def _snapshot_locked(self) -> LlamaAdmissionSnapshot:
+        return LlamaAdmissionSnapshot(
+            key = self.key,
+            capacity = self._capacity,
+            active = self._active,
+            queued = len(self._waiters),
+        )
+
+
+_QUEUES_LOCK = threading.Lock()
+_QUEUES: dict[str, LlamaAdmissionQueue] = {}
+
+
+def get_llama_admission_queue(key: str) -> LlamaAdmissionQueue:
+    with _QUEUES_LOCK:
+        queue = _QUEUES.get(key)
+        if queue is None:
+            queue = LlamaAdmissionQueue(key)
+            _QUEUES[key] = queue
+            # base_url carries a fresh ephemeral port on every model load, so
+            # each load registers a new key. Drop the now-idle queues from prior
+            # loads so the registry can't grow without bound on a long-running
+            # server. Queues with in-flight requests are kept until they drain.
+            for stale_key in [k for k in _QUEUES if k != key and _QUEUES[k].is_idle()]:
+                del _QUEUES[stale_key]
+        return queue
+
+
+def reset_llama_admission_queues() -> None:
+    with _QUEUES_LOCK:
+        _QUEUES.clear()

studio/backend/core/inference/llama_cpp.py

@@ -1541,6 +1541,7 @@ class LlamaCppBackend:
         self._context_length: Optional[int] = None
         self._effective_context_length: Optional[int] = None
         self._max_context_length: Optional[int] = None
+        self._effective_parallel_slots: int = 1
         self._chat_template: Optional[str] = None
         self._chat_template_override: Optional[str] = None
         self._supports_reasoning: bool = False
@@ -1726,6 +1727,15 @@ class LlamaCppBackend:
         """Return the effective context length the server is running at."""
         return self._effective_context_length or self._context_length
 
+    @property
+    def effective_parallel_slots(self) -> int:
+        """Return the serving-slot count the active llama-server actually uses."""
+        try:
+            slots = int(getattr(self, "_effective_parallel_slots", 1))
+        except (TypeError, ValueError):
+            slots = 1
+        return max(1, slots)
+
     @property
     def max_context_length(self) -> Optional[int]:
         """Return the largest context that fits on this hardware at load time.
@@ -1742,6 +1752,16 @@ class LlamaCppBackend:
         """Return the model's native context length from GGUF metadata."""
         return self._context_length
 
+    def _commit_effective_parallel_slots(self, n_parallel: int) -> None:
+        try:
+            slots = int(n_parallel)
+        except (TypeError, ValueError):
+            slots = 1
+        self._effective_parallel_slots = max(1, slots)
+
+    def _reset_effective_parallel_slots(self) -> None:
+        self._effective_parallel_slots = 1
+
     @staticmethod
     def _read_rss_bytes(pid: int) -> Optional[int]:
         """Resident set size of ``pid`` in bytes, from /proc/<pid>/status (Linux).
@@ -7187,6 +7207,7 @@ class LlamaCppBackend:
                         )
 
                 self._healthy = True
+                self._commit_effective_parallel_slots(n_parallel)
 
                 # Commit caller intent only after _healthy=True so a failed start
                 # can't poison the next inheritance check. None keeps prior, []
@@ -7724,6 +7745,7 @@ class LlamaCppBackend:
             self._context_length = None
             self._effective_context_length = None
             self._max_context_length = None
+            self._reset_effective_parallel_slots()
             self._chat_template = None
             self._chat_template_override = None
             self._supports_reasoning = False
@@ -7779,6 +7801,7 @@ class LlamaCppBackend:
         # Stop the watchdog before a deliberate kill so a planned reload/unload
         # isn't seen as a crash; a real crash never routes through here.
         self._stop_mtp_crash_watchdog()
+        self._reset_effective_parallel_slots()
         if self._process is None:
             return
         try:

studio/backend/tests/test_llama_admission.py

@@ -0,0 +1,320 @@
+# SPDX-License-Identifier: AGPL-3.0-only
+# Copyright 2026-present the Unsloth AI Inc. team. All rights reserved.
+
+import asyncio
+import os
+import sys
+import threading
+
+import pytest
+
+_backend = os.path.join(os.path.dirname(__file__), "..")
+sys.path.insert(0, _backend)
+
+from core.inference import llama_admission
+from core.inference.llama_admission import (
+    ADMISSION_CONTROL_ENV,
+    ADMISSION_KEEPALIVE_INTERVAL_ENV,
+    ADMISSION_MAX_QUEUE_ENV,
+    ADMISSION_QUEUE_TIMEOUT_ENV,
+    DEFAULT_ADMISSION_KEEPALIVE_INTERVAL_S,
+    DEFAULT_ADMISSION_MAX_QUEUE,
+    DEFAULT_ADMISSION_QUEUE_TIMEOUT_S,
+    LlamaAdmissionConfig,
+    LlamaAdmissionQueueFull,
+    get_llama_admission_queue,
+    llama_admission_config_from_env,
+    reset_llama_admission_queues,
+)
+
+
+@pytest.fixture(autouse = True)
+def _reset_queues():
+    reset_llama_admission_queues()
+    yield
+    reset_llama_admission_queues()
+
+
+def test_admission_config_defaults(monkeypatch):
+    for name in (
+        ADMISSION_CONTROL_ENV,
+        ADMISSION_QUEUE_TIMEOUT_ENV,
+        ADMISSION_KEEPALIVE_INTERVAL_ENV,
+        ADMISSION_MAX_QUEUE_ENV,
+    ):
+        monkeypatch.delenv(name, raising = False)
+
+    config = llama_admission_config_from_env()
+
+    assert config.enabled is True
+    assert config.queue_timeout_s == DEFAULT_ADMISSION_QUEUE_TIMEOUT_S
+    assert config.keepalive_interval_s == DEFAULT_ADMISSION_KEEPALIVE_INTERVAL_S
+    assert config.max_queue == DEFAULT_ADMISSION_MAX_QUEUE
+
+
+def test_admission_config_env_overrides(monkeypatch):
+    monkeypatch.setenv(ADMISSION_CONTROL_ENV, "off")
+    monkeypatch.setenv(ADMISSION_QUEUE_TIMEOUT_ENV, "0")
+    monkeypatch.setenv(ADMISSION_KEEPALIVE_INTERVAL_ENV, "0.25")
+    monkeypatch.setenv(ADMISSION_MAX_QUEUE_ENV, "0")
+
+    config = llama_admission_config_from_env()
+
+    assert config.enabled is False
+    assert config.queue_timeout_s is None
+    assert config.keepalive_interval_s == 0.25
+    assert config.max_queue is None
+
+
+def test_admission_config_positive_queue_timeout_env(monkeypatch):
+    monkeypatch.setenv(ADMISSION_QUEUE_TIMEOUT_ENV, "600")
+
+    config = llama_admission_config_from_env()
+
+    assert config.queue_timeout_s == 600.0
+
+
+def test_fifo_capacity_one_grants_next_waiter_on_release():
+    async def _run():
+        queue = get_llama_admission_queue("http://llama.test")
+        config = LlamaAdmissionConfig()
+
+        first = queue.reserve(capacity = 1, config = config)
+        second = queue.reserve(capacity = 1, config = config)
+        third = queue.reserve(capacity = 1, config = config)
+
+        first_lease = first.lease_nowait()
+        assert first_lease is not None
+        assert second.lease_nowait() is None
+        assert third.lease_nowait() is None
+        assert queue.snapshot().queued == 2
+
+        first_lease.release()
+        second_lease = await second.wait(0.1)
+        assert second_lease is not None
+        assert third.lease_nowait() is None
+
+        second_lease.release()
+        third_lease = await third.wait(0.1)
+        assert third_lease is not None
+        third_lease.release()
+
+        snapshot = queue.snapshot()
+        assert snapshot.active == 0
+        assert snapshot.queued == 0
+
+    asyncio.run(_run())
+
+
+def test_queue_full_rejects_excess_waiter():
+    async def _run():
+        queue = get_llama_admission_queue("http://llama.test")
+        config = LlamaAdmissionConfig(max_queue = 1)
+
+        first = queue.reserve(capacity = 1, config = config)
+        queued = queue.reserve(capacity = 1, config = config)
+
+        assert first.lease_nowait() is not None
+        assert queued.lease_nowait() is None
+        with pytest.raises(LlamaAdmissionQueueFull):
+            queue.reserve(capacity = 1, config = config)
+
+    asyncio.run(_run())
+
+
+def test_disabled_admission_bypasses_active_slot_limit():
+    async def _run():
+        queue = get_llama_admission_queue("http://llama.test")
+        config = LlamaAdmissionConfig(enabled = False)
+
+        first = queue.reserve(capacity = 1, config = config)
+        second = queue.reserve(capacity = 1, config = config)
+
+        assert first.lease_nowait() is not None
+        assert second.lease_nowait() is not None
+        assert queue.snapshot().active == 0
+        assert queue.snapshot().queued == 0
+
+    asyncio.run(_run())
+
+
+def test_cancelling_promoted_waiter_releases_slot():
+    async def _run():
+        queue = get_llama_admission_queue("http://llama.test")
+        config = LlamaAdmissionConfig()
+
+        first = queue.reserve(capacity = 1, config = config)
+        second = queue.reserve(capacity = 1, config = config)
+        first_lease = first.lease_nowait()
+
+        first_lease.release()
+        await asyncio.sleep(0)
+        second.cancel()
+
+        snapshot = queue.snapshot()
+        assert snapshot.active == 0
+        assert snapshot.queued == 0
+
+    asyncio.run(_run())
+
+
+def test_cancelling_promoted_waiter_before_delivery_releases_slot():
+    async def _run():
+        queue = get_llama_admission_queue("http://llama.test")
+        config = LlamaAdmissionConfig()
+
+        first = queue.reserve(capacity = 1, config = config)
+        second = queue.reserve(capacity = 1, config = config)
+        first_lease = first.lease_nowait()
+
+        first_lease.release()
+        second.cancel()
+
+        snapshot = queue.snapshot()
+        assert snapshot.active == 0
+        assert snapshot.queued == 0
+
+    asyncio.run(_run())
+
+
+def test_external_waiter_future_cancel_invalidates_reservation():
+    async def _run():
+        queue = get_llama_admission_queue("http://llama.test")
+        config = LlamaAdmissionConfig()
+
+        first = queue.reserve(capacity = 1, config = config)
+        second = queue.reserve(capacity = 1, config = config)
+        first_lease = first.lease_nowait()
+        assert first_lease is not None
+        assert second._waiter is not None
+
+        second._waiter.future.cancel()
+
+        assert second.lease_nowait() is None
+        assert second.is_cancelled is True
+        assert await second.wait(0.01) is None
+
+        first_lease.release()
+        snapshot = queue.snapshot()
+        assert snapshot.active == 0
+        assert snapshot.queued == 0
+
+    asyncio.run(_run())
+
+
+def test_wait_returns_none_when_waiter_future_cancelled_during_wait():
+    async def _run():
+        queue = get_llama_admission_queue("http://llama.test")
+        config = LlamaAdmissionConfig()
+
+        first = queue.reserve(capacity = 1, config = config)
+        second = queue.reserve(capacity = 1, config = config)
+        first_lease = first.lease_nowait()
+        assert first_lease is not None
+        assert second._waiter is not None
+
+        wait_task = asyncio.create_task(second.wait(1.0))
+        await asyncio.sleep(0)
+        second._waiter.future.cancel()
+
+        assert await asyncio.wait_for(wait_task, timeout = 0.1) is None
+        assert second.is_cancelled is True
+
+        first_lease.release()
+        snapshot = queue.snapshot()
+        assert snapshot.active == 0
+        assert snapshot.queued == 0
+
+    asyncio.run(_run())
+
+
+def test_capacity_increase_promotes_existing_waiter_fifo():
+    async def _run():
+        queue = get_llama_admission_queue("http://llama.test")
+        config = LlamaAdmissionConfig()
+
+        first = queue.reserve(capacity = 1, config = config)
+        second = queue.reserve(capacity = 1, config = config)
+
+        first_lease = first.lease_nowait()
+        assert first_lease is not None
+        assert second.lease_nowait() is None
+        assert queue.snapshot().active == 1
+        assert queue.snapshot().queued == 1
+
+        third = queue.reserve(capacity = 2, config = config)
+
+        second_lease = await second.wait(0.1)
+        assert second_lease is not None
+        assert third.lease_nowait() is None
+
+        snapshot = queue.snapshot()
+        assert snapshot.capacity == 2
+        assert snapshot.active == 2
+        assert snapshot.queued == 1
+
+        first_lease.release()
+        third_lease = await third.wait(0.1)
+        assert third_lease is not None
+
+        second_lease.release()
+        third_lease.release()
+        snapshot = queue.snapshot()
+        assert snapshot.active == 0
+        assert snapshot.queued == 0
+
+    asyncio.run(_run())
+
+
+def test_lease_release_is_idempotent_under_concurrent_calls():
+    async def _run():
+        queue = get_llama_admission_queue("http://llama.test")
+        config = LlamaAdmissionConfig()
+
+        reservation = queue.reserve(capacity = 1, config = config)
+        lease = reservation.lease_nowait()
+        assert lease is not None
+
+        threads = [threading.Thread(target = lease.release) for _ in range(16)]
+        for thread in threads:
+            thread.start()
+        for thread in threads:
+            thread.join()
+
+        snapshot = queue.snapshot()
+        assert snapshot.active == 0
+        assert snapshot.queued == 0
+
+    asyncio.run(_run())
+
+
+def test_new_key_evicts_idle_prior_load_queues():
+    # Each model load carries a fresh ephemeral port, so a new base_url key must
+    # not leave the drained queues from earlier loads accumulating forever.
+    get_llama_admission_queue("http://127.0.0.1:1001")
+    get_llama_admission_queue("http://127.0.0.1:1002")
+    assert set(llama_admission._QUEUES) == {"http://127.0.0.1:1002"}
+
+    get_llama_admission_queue("http://127.0.0.1:1003")
+    assert set(llama_admission._QUEUES) == {"http://127.0.0.1:1003"}
+
+
+def test_new_key_retains_in_flight_prior_load_queue():
+    config = LlamaAdmissionConfig()
+    busy = get_llama_admission_queue("http://127.0.0.1:2001")
+
+    async def _run():
+        reservation = busy.reserve(capacity = 1, config = config)
+        lease = reservation.lease_nowait()
+        assert lease is not None
+
+        # A new load must not drop a queue that still has an in-flight request.
+        get_llama_admission_queue("http://127.0.0.1:2002")
+        assert set(llama_admission._QUEUES) == {"http://127.0.0.1:2001", "http://127.0.0.1:2002"}
+
+        # Once it drains, the next load reclaims it.
+        lease.release()
+        get_llama_admission_queue("http://127.0.0.1:2003")
+        assert set(llama_admission._QUEUES) == {"http://127.0.0.1:2003"}
+
+    asyncio.run(_run())

studio/backend/tests/test_llama_cpp_effective_parallel_slots.py

@@ -0,0 +1,53 @@
+# SPDX-License-Identifier: AGPL-3.0-only
+# Copyright 2026-present the Unsloth AI Inc. team. All rights reserved.
+
+import os
+import sys
+
+import pytest
+
+_backend = os.path.join(os.path.dirname(__file__), "..")
+sys.path.insert(0, _backend)
+
+from core.inference import llama_cpp as llama_cpp_module
+from core.inference.llama_cpp import LlamaCppBackend
+
+
+@pytest.fixture
+def backend(monkeypatch):
+    monkeypatch.setattr(LlamaCppBackend, "_kill_orphaned_servers", lambda self: 0)
+    monkeypatch.setattr(llama_cpp_module.atexit, "register", lambda *_args, **_kwargs: None)
+    return LlamaCppBackend()
+
+
+def test_effective_parallel_slots_initial_value_is_one(backend):
+    assert backend.effective_parallel_slots == 1
+
+
+def test_effective_parallel_slots_commit_uses_final_positive_parallel(backend):
+    backend._commit_effective_parallel_slots(3)
+
+    assert backend.effective_parallel_slots == 3
+
+
+@pytest.mark.parametrize("value", [None, 0, -2, "not-an-int"])
+def test_effective_parallel_slots_commit_invalid_value_falls_back_to_one(backend, value):
+    backend._commit_effective_parallel_slots(value)
+
+    assert backend.effective_parallel_slots == 1
+
+
+def test_effective_parallel_slots_reset_returns_to_one(backend):
+    backend._commit_effective_parallel_slots(4)
+
+    backend._reset_effective_parallel_slots()
+
+    assert backend.effective_parallel_slots == 1
+
+
+def test_effective_parallel_slots_unload_resets_to_one(backend):
+    backend._commit_effective_parallel_slots(4)
+
+    backend.unload_model()
+
+    assert backend.effective_parallel_slots == 1

tests/studio/test_stream_cancel_registration_timing.py

@@ -17,7 +17,7 @@ from pathlib import Path
 
 
 SOURCE_PATH = Path(__file__).resolve().parents[2] / "studio" / "backend" / "routes" / "inference.py"
-SRC = SOURCE_PATH.read_text()
+SRC = SOURCE_PATH.read_text(encoding = "utf-8")
 _TREE = ast.parse(SRC)
 
 
@@ -166,16 +166,20 @@ def test_chat_completions_streams_avoid_starlette_task_group():
 
 
 def test_openai_passthrough_stream_avoids_starlette_task_group():
-    top = _async_function("_openai_passthrough_stream")
+    functions = [
+        _async_function("_openai_passthrough_stream"),
+        _async_function("_openai_passthrough_stream_admitted"),
+    ]
     legacy_calls = []
     same_task_calls = 0
-    for sub in ast.walk(top):
-        if not (isinstance(sub, ast.Call) and isinstance(sub.func, ast.Name)):
-            continue
-        if sub.func.id == "StreamingResponse":
-            legacy_calls.append(sub.lineno)
-        if sub.func.id == "_SameTaskStreamingResponse":
-            same_task_calls += 1
+    for fn in functions:
+        for sub in ast.walk(fn):
+            if not (isinstance(sub, ast.Call) and isinstance(sub.func, ast.Name)):
+                continue
+            if sub.func.id == "StreamingResponse":
+                legacy_calls.append(sub.lineno)
+            if sub.func.id == "_SameTaskStreamingResponse":
+                same_task_calls += 1
     assert not legacy_calls, (
         "OpenAI passthrough streams must use _SameTaskStreamingResponse, "
         "not Starlette's legacy task-group StreamingResponse. Lines: "
@@ -197,7 +201,7 @@ def test_direct_llama_server_streams_install_disconnect_watcher():
         "openai_completions",
         "_responses_stream",
         "_anthropic_passthrough_stream",
-        "_openai_passthrough_stream",
+        "_openai_passthrough_stream_admitted",
     }
     missing = [
         name