From 619e2256d38c5470363f0a2d0e240d2a516252e2 Mon Sep 17 00:00:00 2001
From: Hogne <227774406+hognek@users.noreply.github.com>
Date: Wed, 8 Jul 2026 00:19:51 +0200
Subject: [PATCH 1/2] feat(models): atomic VRAM check-and-reserve before model
 load (TOCTOU fix)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Adds VramReservationManager — an asyncio.Lock-protected VRAM reservation
system that atomically checks free VRAM (via nvidia-smi) and reserves the
required amount before a model load begins.  Concurrent callers see the
in-flight reservation and back off, preventing the TOCTOU race where two
model pulls both see enough free VRAM and both try to load, causing OOM.

Integration points:
- POST /api/models/pull  — reserve before ollama pull, release after
- POST /api/models/download (rkllama path) — reserve before installer,
  release in the background task's finally block
- GET  /api/models/vram-reservations — stats endpoint for monitoring

New files:
- tinyagentos/vram_reservation.py — atomic reserve/release manager
- tests/test_vram_reservation.py — 15 tests including concurrent safety

Wired as app.state.vram_reservation in the FastAPI lifespan alongside
the cluster manager and resource scheduler.

Issue: #1706
---
 tests/test_vram_reservation.py  | 238 ++++++++++++++++++++++++++++++++
 tinyagentos/app.py              |   7 +
 tinyagentos/routes/models.py    | 115 +++++++++++----
 tinyagentos/vram_reservation.py | 186 +++++++++++++++++++++++++
 4 files changed, 522 insertions(+), 24 deletions(-)
 create mode 100644 tests/test_vram_reservation.py
 create mode 100644 tinyagentos/vram_reservation.py

diff --git a/tests/test_vram_reservation.py b/tests/test_vram_reservation.py
new file mode 100644
index 000000000..0054b3837
--- /dev/null
+++ b/tests/test_vram_reservation.py
@@ -0,0 +1,238 @@
+"""Tests for atomic VRAM reservation (taOS #1706 TOCTOU fix).
+
+Verifies that:
+- Concurrent reserve calls cannot over-commit VRAM.
+- reserve() returns None when there's not enough VRAM.
+- release() correctly returns VRAM to the pool.
+- release() is idempotent.
+- Zero-VRAM reservations always succeed.
+- stats() reflects the current state.
+"""
+
+import asyncio
+
+import pytest
+
+from tinyagentos.vram_reservation import VramReservationManager
+
+
+# ── test helpers ────────────────────────────────────────────────────
+
+
+def _manager_with_probe(free_mb: int, total_mb: int) -> VramReservationManager:
+    """Return a manager whose _probe_vram returns fixed values."""
+    mgr = VramReservationManager()
+
+    def _fake_probe() -> tuple[int, int]:
+        return free_mb, total_mb
+
+    mgr._probe_vram = staticmethod(_fake_probe)  # type: ignore[method-assign]
+    return mgr
+
+
+# ── basic reserve / release ─────────────────────────────────────────
+
+
+class TestReserveRelease:
+    """Happy-path reserve and release."""
+
+    @pytest.mark.asyncio
+    async def test_reserve_when_vram_available(self):
+        """reserve() succeeds when enough VRAM is free."""
+        mgr = _manager_with_probe(8192, 8192)
+        reservation = await mgr.reserve(4096, caller="test")
+        assert reservation is not None
+        assert reservation.vram_mb == 4096
+        assert reservation.caller == "test"
+        assert mgr.reserved_vram_mb == 4096
+        assert mgr.pending_count == 1
+
+    @pytest.mark.asyncio
+    async def test_reserve_denied_when_insufficient(self):
+        """reserve() returns None when VRAM is too low."""
+        mgr = _manager_with_probe(2048, 8192)
+        reservation = await mgr.reserve(4096, caller="test")
+        assert reservation is None
+        assert mgr.reserved_vram_mb == 0
+        assert mgr.pending_count == 0
+
+    @pytest.mark.asyncio
+    async def test_release_returns_vram(self):
+        """release() returns reserved VRAM to the pool."""
+        mgr = _manager_with_probe(8192, 8192)
+        res = await mgr.reserve(4096, caller="test")
+        assert res is not None
+
+        released = mgr.release(res.reservation_id)
+        assert released is True
+        assert mgr.reserved_vram_mb == 0
+        assert mgr.pending_count == 0
+
+    @pytest.mark.asyncio
+    async def test_release_idempotent(self):
+        """release() is safe to call multiple times."""
+        mgr = _manager_with_probe(8192, 8192)
+        res = await mgr.reserve(4096, caller="test")
+        assert res is not None
+
+        assert mgr.release(res.reservation_id) is True
+        assert mgr.release(res.reservation_id) is False  # already released
+        assert mgr.release("non-existent") is False
+        assert mgr.reserved_vram_mb == 0
+
+    @pytest.mark.asyncio
+    async def test_multiple_reservations_serial(self):
+        """Serial reservations accumulate correctly."""
+        mgr = _manager_with_probe(8192, 8192)
+
+        r1 = await mgr.reserve(2000, caller="a")
+        r2 = await mgr.reserve(3000, caller="b")
+        assert r1 is not None
+        assert r2 is not None
+        assert mgr.reserved_vram_mb == 5000
+        assert mgr.pending_count == 2
+
+        mgr.release(r1.reservation_id)
+        assert mgr.reserved_vram_mb == 3000
+
+        mgr.release(r2.reservation_id)
+        assert mgr.reserved_vram_mb == 0
+
+    @pytest.mark.asyncio
+    async def test_reserve_at_boundary(self):
+        """Reserving exactly the free amount succeeds."""
+        mgr = _manager_with_probe(4096, 8192)
+        res = await mgr.reserve(4096, caller="test")
+        assert res is not None
+        assert mgr.reserved_vram_mb == 4096
+
+    @pytest.mark.asyncio
+    async def test_reserve_one_more_than_free_fails(self):
+        """Reserving free+1 fails."""
+        mgr = _manager_with_probe(4096, 8192)
+        res = await mgr.reserve(4097, caller="test")
+        assert res is None
+
+
+# ── concurrent safety ───────────────────────────────────────────────
+
+
+class TestConcurrentSafety:
+    """Two concurrent reserve calls cannot over-commit."""
+
+    @pytest.mark.asyncio
+    async def test_concurrent_reserve_only_one_admitted(self):
+        """With 8192 MiB free and each needing 5000 MiB, only one admitted."""
+        mgr = _manager_with_probe(8192, 8192)
+        results: list[bool] = []
+
+        async def try_reserve(caller: str):
+            res = await mgr.reserve(5000, caller=caller)
+            results.append(res is not None)
+
+        await asyncio.gather(try_reserve("a"), try_reserve("b"))
+
+        admitted = sum(results)
+        assert admitted == 1, f"Expected 1 admission, got {admitted}: {results}"
+        assert mgr.reserved_vram_mb == 5000
+
+    @pytest.mark.asyncio
+    async def test_concurrent_vram_accounting(self):
+        """After two serial reserves filling VRAM, a third fails."""
+        mgr = _manager_with_probe(8192, 8192)
+
+        r1 = await mgr.reserve(4000, caller="a")
+        r2 = await mgr.reserve(4000, caller="b")
+        assert r1 is not None
+        assert r2 is not None
+
+        # Third should fail — only 192 MiB effective left.
+        r3 = await mgr.reserve(4000, caller="c")
+        assert r3 is None
+        assert mgr.reserved_vram_mb == 8000
+
+
+# ── zero / negative VRAM ────────────────────────────────────────────
+
+
+class TestZeroVram:
+    """Zero-VRAM reservations always succeed as lightweight markers."""
+
+    @pytest.mark.asyncio
+    async def test_zero_vram_reservation(self):
+        mgr = _manager_with_probe(0, 0)
+        res = await mgr.reserve(0, caller="noop")
+        assert res is not None
+        assert res.vram_mb == 0
+        assert mgr.reserved_vram_mb == 0
+        assert mgr.pending_count == 0
+
+    @pytest.mark.asyncio
+    async def test_negative_vram_reservation(self):
+        mgr = _manager_with_probe(8192, 8192)
+        res = await mgr.reserve(-1, caller="test")
+        assert res is not None
+        assert res.vram_mb == 0
+        assert mgr.reserved_vram_mb == 0
+        assert mgr.pending_count == 0
+
+
+# ── available_vram / stats ──────────────────────────────────────────
+
+
+class TestAvailableVram:
+    """available_vram() and stats() reflect current state."""
+
+    @pytest.mark.asyncio
+    async def test_available_vram_no_reservations(self):
+        mgr = _manager_with_probe(8192, 16384)
+        free, total = mgr.available_vram()
+        assert free == 8192
+        assert total == 16384
+
+    @pytest.mark.asyncio
+    async def test_available_vram_with_reservations(self):
+        mgr = _manager_with_probe(8192, 16384)
+        res = await mgr.reserve(3000, caller="test")
+        assert res is not None
+
+        free, total = mgr.available_vram()
+        assert free == 5192  # 8192 - 3000
+        assert total == 16384
+
+    @pytest.mark.asyncio
+    async def test_stats(self):
+        mgr = _manager_with_probe(8192, 16384)
+
+        s = mgr.stats()
+        assert s["free_vram_mb"] == 8192
+        assert s["total_vram_mb"] == 16384
+        assert s["reserved_vram_mb"] == 0
+        assert s["effective_free_mb"] == 8192
+        assert s["pending_reservations"] == 0
+
+        res = await mgr.reserve(4096, caller="test")
+        assert res is not None
+
+        s = mgr.stats()
+        assert s["reserved_vram_mb"] == 4096
+        assert s["effective_free_mb"] == 4096
+        assert s["pending_reservations"] == 1
+
+
+# ── real probe (integration smoke test) ─────────────────────────────
+
+
+class TestRealProbe:
+    """Smoke-test with the real nvidia-smi probe (if available)."""
+
+    @pytest.mark.asyncio
+    async def test_reserve_with_real_probe(self):
+        """Reserving 1 MiB should always succeed on a real GPU."""
+        mgr = VramReservationManager()
+        res = await mgr.reserve(1, caller="smoke-test")
+        # On a system with nvidia-smi and some free VRAM, this should succeed.
+        # On a system without nvidia-smi, the probe returns (0, 0) so
+        # this will fail — that's fine, the test is a best-effort smoke.
+        if res is not None:
+            mgr.release(res.reservation_id)
diff --git a/tinyagentos/app.py b/tinyagentos/app.py
index 617bd719a..406ed5d3d 100644
--- a/tinyagentos/app.py
+++ b/tinyagentos/app.py
@@ -43,6 +43,7 @@ async def get_response(self, path, scope):
 from tinyagentos.backend_fallback import BackendFallback
 from tinyagentos.capabilities import CapabilityChecker
 from tinyagentos.cluster.manager import ClusterManager
+from tinyagentos.vram_reservation import VramReservationManager
 from tinyagentos.cluster.router import TaskRouter
 from tinyagentos.config import auto_register_from_manifest, load_config, save_config, save_config_locked
 from tinyagentos.lifecycle_manager import LifecycleManager
@@ -1140,6 +1141,12 @@ async def _reload_llm_proxy_on_catalog_change() -> None:
         except Exception:
             logger.exception("resource scheduler failed to build — routes will use static config")
             app.state.resource_scheduler = None
+
+        # VRAM reservation manager — atomic check-and-reserve so two
+        # concurrent model loads cannot both pass a VRAM check before
+        # either consumes physical VRAM (TOCTOU race, taOS #1706).
+        app.state.vram_reservation = VramReservationManager()
+
         # Detect and set container runtime
         from tinyagentos.containers.backend import configure_container_runtime
         configure_container_runtime(config)
diff --git a/tinyagentos/routes/models.py b/tinyagentos/routes/models.py
index 407248414..c7c8bf6f0 100644
--- a/tinyagentos/routes/models.py
+++ b/tinyagentos/routes/models.py
@@ -33,6 +33,7 @@ class DeleteRequest(BaseModel):
 
 class PullRequest(BaseModel):
     model_name: str
+    required_vram_mb: int = 0
 
 
 router = APIRouter()
@@ -367,36 +368,64 @@ async def download_model(request: Request, body: DownloadRequest):
         installer = RkllamaInstaller(rkllama_url=resolve_rkllama_url(None))
         catalog = getattr(request.app.state, "backend_catalog", None)
 
-        async def _install_and_record(on_progress):
-            result = await installer.install(
-                body.app_id, {}, variant=variant, on_progress=on_progress
+        # Atomic VRAM check-and-reserve (TOCTOU fix, taOS #1706).
+        # rkllama pulls the model into its backend, which loads it
+        # into VRAM — two concurrent pulls could both pass the VRAM
+        # check and OOM.
+        vram_mgr = getattr(request.app.state, "vram_reservation", None)
+        reservation = None
+        estimated_vram = int(variant.get("min_ram_mb", 0) or 0)  # heuristic
+        if vram_mgr is not None and estimated_vram > 0:
+            reservation = await vram_mgr.reserve(
+                estimated_vram, caller=f"rkllama-pull:{body.app_id}",
             )
-            if result.get("success"):
-                # rkllama pulls the weight into its own model store, not
-                # models_dir, so the disk scan in list_models() never sees it.
-                # Record the install as a registry breadcrumb (corroborated by
-                # live-backend evidence, see _model_to_dict) and force an
-                # immediate catalog refresh so the newly pulled model shows up
-                # in all_models() right away instead of waiting for the next
-                # poll tick.
-                try:
-                    registry.mark_installed(
-                        body.app_id, variant.get("id") or "latest", state="installed"
-                    )
-                except Exception:
-                    logger.warning(
-                        "Failed to record rkllama install for %s in registry",
-                        body.app_id, exc_info=True,
-                    )
-                if catalog is not None:
+            if reservation is None:
+                effective_free, _total = vram_mgr.available_vram()
+                return JSONResponse(
+                    {
+                        "error": (
+                            f"Insufficient VRAM: need ~{estimated_vram} MiB, "
+                            f"have {effective_free} MiB available. "
+                            f"Try a smaller model or free VRAM first."
+                        ),
+                    },
+                    status_code=503,
+                )
+
+        async def _install_and_record(on_progress):
+            try:
+                result = await installer.install(
+                    body.app_id, {}, variant=variant, on_progress=on_progress
+                )
+                if result.get("success"):
+                    # rkllama pulls the weight into its own model store, not
+                    # models_dir, so the disk scan in list_models() never sees it.
+                    # Record the install as a registry breadcrumb (corroborated by
+                    # live-backend evidence, see _model_to_dict) and force an
+                    # immediate catalog refresh so the newly pulled model shows up
+                    # in all_models() right away instead of waiting for the next
+                    # poll tick.
                     try:
-                        await catalog.refresh()
+                        registry.mark_installed(
+                            body.app_id, variant.get("id") or "latest", state="installed"
+                        )
                     except Exception:
                         logger.warning(
-                            "Backend catalog refresh after rkllama install failed for %s",
+                            "Failed to record rkllama install for %s in registry",
                             body.app_id, exc_info=True,
                         )
-            return result
+                    if catalog is not None:
+                        try:
+                            await catalog.refresh()
+                        except Exception:
+                            logger.warning(
+                                "Backend catalog refresh after rkllama install failed for %s",
+                                body.app_id, exc_info=True,
+                            )
+                return result
+            finally:
+                if vram_mgr is not None and reservation is not None:
+                    vram_mgr.release(reservation.reservation_id)
 
         dm.start_installer_task(download_id, _install_and_record)
         return {
@@ -602,6 +631,28 @@ async def pull_model(request: Request, body: PullRequest):
     model_name = body.model_name.strip()
     if not model_name:
         return JSONResponse({"error": "model_name is required"}, status_code=400)
+
+    # Atomic VRAM check-and-reserve (TOCTOU fix, taOS #1706) —
+    # two concurrent pulls could both see free VRAM and OOM.
+    vram_mgr = getattr(request.app.state, "vram_reservation", None)
+    reservation = None
+    if vram_mgr is not None and body.required_vram_mb > 0:
+        reservation = await vram_mgr.reserve(
+            body.required_vram_mb, caller=f"ollama-pull:{model_name}",
+        )
+        if reservation is None:
+            effective_free, _total = vram_mgr.available_vram()
+            return JSONResponse(
+                {
+                    "error": (
+                        f"Insufficient VRAM: need {body.required_vram_mb} MiB, "
+                        f"have {effective_free} MiB available. "
+                        f"Try a smaller model or free VRAM first."
+                    ),
+                },
+                status_code=503,
+            )
+
     try:
         http_client = request.app.state.http_client
         config = request.app.state.config
@@ -626,6 +677,22 @@ async def pull_model(request: Request, body: PullRequest):
     except Exception as e:
         logger.warning(f"Ollama pull failed: {e}")
         return JSONResponse({"error": str(e)}, status_code=502)
+    finally:
+        if vram_mgr is not None and reservation is not None:
+            vram_mgr.release(reservation.reservation_id)
+
+
+@router.get("/api/models/vram-reservations")
+async def vram_reservation_stats(request: Request):
+    """Return VRAM reservation manager stats for monitoring.
+
+    Includes real-time free/total VRAM from nvidia-smi, in-flight
+    reservations, and effective free VRAM after subtracting reservations.
+    """
+    vram_mgr = getattr(request.app.state, "vram_reservation", None)
+    if vram_mgr is None:
+        return {"available": False}
+    return vram_mgr.stats()
 
 
 @router.get("/api/models/recommended")
diff --git a/tinyagentos/vram_reservation.py b/tinyagentos/vram_reservation.py
new file mode 100644
index 000000000..1b2eee783
--- /dev/null
+++ b/tinyagentos/vram_reservation.py
@@ -0,0 +1,186 @@
+"""
+Atomic VRAM reservation — check-and-reserve to prevent TOCTOU races.
+
+When two concurrent model loads both check available VRAM before either
+consumes it, they can both pass and cause OOM.  This module provides
+atomic check-and-reserve semantics: check available VRAM, reserve what
+you need, then load the model.  Concurrent callers see the reserved
+capacity and back off.
+
+Pattern:
+
+    mgr = VramReservationManager()
+    reservation = await mgr.reserve(vram_mb=4096, caller="ollama-pull")
+    if reservation is not None:
+        try:
+            await load_model()
+        finally:
+            mgr.release(reservation.reservation_id)
+    else:
+        raise RuntimeError("Not enough VRAM")
+
+The reservation is *not* a context manager — callers must release
+explicitly (typically in a ``finally`` block) so the reservation can
+outlive a single ``async with`` block while the model loads.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import secrets
+import time
+from dataclasses import dataclass
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class VramReservation:
+    """A held VRAM reservation.  Release via manager.release()."""
+
+    reservation_id: str
+    vram_mb: int
+    caller: str
+    created_at: float
+
+
+class VramReservationManager:
+    """Atomic VRAM check-and-reserve for model loads.
+
+    Thread-safe via ``asyncio.Lock``.  Reads real-time VRAM from
+    nvidia-smi and subtracts in-flight reservations so concurrent
+    callers see the capacity already promised to others.
+
+    Zero-VRAM reservations (``vram_mb <= 0``) always succeed as a
+    lightweight marker — no lock acquisition, no probe, no accounting.
+    """
+
+    def __init__(self) -> None:
+        self._lock = asyncio.Lock()
+        self._reserved_vram_mb: int = 0
+        self._pending: dict[str, VramReservation] = {}
+
+    # ── public API ──────────────────────────────────────────────────
+
+    async def reserve(
+        self,
+        vram_mb: int,
+        caller: str = "",
+    ) -> VramReservation | None:
+        """Atomically check available VRAM and reserve *vram_mb* MiB.
+
+        Returns a ``VramReservation`` if admitted, ``None`` if
+        insufficient VRAM.  Caller **must** call :meth:`release` when
+        done — this object is not a context manager so the reservation
+        can outlive a single ``async with`` block while the model loads.
+
+        *vram_mb* ≤ 0 always succeeds without probing VRAM or acquiring
+        the lock (a lightweight no-op marker).
+        """
+        if vram_mb <= 0:
+            return VramReservation(
+                reservation_id=f"vr_{secrets.token_hex(4)}",
+                vram_mb=0,
+                caller=caller or "noop",
+                created_at=time.time(),
+            )
+
+        async with self._lock:
+            free_mb, total_mb = self._probe_vram()
+            effective_free = max(0, free_mb - self._reserved_vram_mb)
+
+            if effective_free < vram_mb:
+                logger.debug(
+                    "vram-reservation: denied — need %d MiB, have %d MiB "
+                    "(%d free − %d reserved)",
+                    vram_mb, effective_free, free_mb, self._reserved_vram_mb,
+                )
+                return None
+
+            reservation_id = f"vr_{secrets.token_hex(4)}"
+            reservation = VramReservation(
+                reservation_id=reservation_id,
+                vram_mb=vram_mb,
+                caller=caller,
+                created_at=time.time(),
+            )
+            self._reserved_vram_mb += vram_mb
+            self._pending[reservation_id] = reservation
+
+            logger.info(
+                "vram-reservation: granted %d MiB to %r (id=%s, "
+                "total reserved=%d MiB, effective free=%d MiB)",
+                vram_mb, caller, reservation_id,
+                self._reserved_vram_mb, effective_free - vram_mb,
+            )
+            return reservation
+
+    def release(self, reservation_id: str) -> bool:
+        """Release a reservation.  Idempotent — safe to call repeatedly.
+
+        Returns ``True`` if a reservation was actually released,
+        ``False`` if *reservation_id* was unknown or already released.
+        """
+        reservation = self._pending.pop(reservation_id, None)
+        if reservation is not None:
+            self._reserved_vram_mb -= reservation.vram_mb
+            logger.debug(
+                "vram-reservation: released %d MiB for %r (id=%s, "
+                "total reserved=%d MiB)",
+                reservation.vram_mb, reservation.caller,
+                reservation_id, self._reserved_vram_mb,
+            )
+            return True
+        return False
+
+    # ── read-only accessors ─────────────────────────────────────────
+
+    @property
+    def reserved_vram_mb(self) -> int:
+        """Total VRAM currently reserved (MiB)."""
+        return self._reserved_vram_mb
+
+    @property
+    def pending_count(self) -> int:
+        """Number of active reservations."""
+        return len(self._pending)
+
+    def available_vram(self) -> tuple[int, int]:
+        """Return ``(effective_free_mb, total_mb)`` after subtracting
+        in-flight reservations from the hardware probe."""
+        free_mb, total_mb = self._probe_vram()
+        effective_free = max(0, free_mb - self._reserved_vram_mb)
+        return effective_free, total_mb
+
+    def stats(self) -> dict:
+        """Return a snapshot for monitoring / debug endpoints."""
+        free_mb, total_mb = self._probe_vram()
+        return {
+            "free_vram_mb": free_mb,
+            "total_vram_mb": total_mb,
+            "reserved_vram_mb": self._reserved_vram_mb,
+            "effective_free_mb": max(0, free_mb - self._reserved_vram_mb),
+            "pending_reservations": self.pending_count,
+        }
+
+    # ── internal ────────────────────────────────────────────────────
+
+    @staticmethod
+    def _probe_vram() -> tuple[int, int]:
+        """Probe real-time free/total VRAM via nvidia-smi.
+
+        Returns ``(free_mb, total_mb)``, or ``(0, 0)`` when no
+        nvidia-smi is available or the probe fails.
+        """
+        try:
+            from tinyagentos.system_stats import read_nvidia_vram
+
+            pair = read_nvidia_vram()
+            if pair is not None:
+                used_mb, total_mb = pair
+                free_mb = max(0, total_mb - used_mb)
+                return free_mb, total_mb
+        except Exception:
+            logger.debug("vram-reservation: probe failed", exc_info=True)
+        return 0, 0

From e6c185a0faa682bd54b365455054167898db37a8 Mon Sep 17 00:00:00 2001
From: Hogne <227774406+hognek@users.noreply.github.com>
Date: Thu, 9 Jul 2026 12:57:50 +0200
Subject: [PATCH 2/2] docs: add TODO for per-model VRAM estimate replacement

jaylfc review: min_ram_mb heuristic is safe for v1 (over-reserves, fails
closed), but should be replaced with a real per-model VRAM estimate on
CUDA GGUF to avoid false 503s in multi-model setups.
---
 tinyagentos/routes/models.py | 4 ++++
 1 file changed, 4 insertions(+)

diff --git a/tinyagentos/routes/models.py b/tinyagentos/routes/models.py
index c7c8bf6f0..209a454de 100644
--- a/tinyagentos/routes/models.py
+++ b/tinyagentos/routes/models.py
@@ -374,6 +374,10 @@ async def download_model(request: Request, body: DownloadRequest):
         # check and OOM.
         vram_mgr = getattr(request.app.state, "vram_reservation", None)
         reservation = None
+        # TODO(#1725): replace min_ram_mb heuristic with a real per-model VRAM
+        # estimate.  On CUDA GGUF the host-RAM floor over-reserves and causes
+        # occasional false 503s in multi-model setups.  Safe for v1 (fails
+        # closed), but a model-card estimate would be more accurate.
         estimated_vram = int(variant.get("min_ram_mb", 0) or 0)  # heuristic
         if vram_mgr is not None and estimated_vram > 0:
             reservation = await vram_mgr.reserve(