per-model queueing: parallel serving across co-loaded models

Multiple loaded models must each have their own queue — one model's
backlog or GPU ownership must never block requests for another model
that is loaded and idle:

- queue admission is per model (is_full_for): 429 only when the
  REQUEST'S model already has queue_max_size waiters (global 4×
  backstop bounds memory); previously ONE hot model's backlog 429'd
  every other model.
- GpuSwapGate: requests for a model already RESIDENT on the engine
  bypass the ownership gate — no swap is needed and its own per-model
  queue governs concurrency. The gate now serializes only requests
  that would actually trigger a model swap (its real purpose). Three
  co-resident embedders can each serve a request in parallel.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_014S8VtAvG499SsCbeESRK7V

codai/__init__.py

@@ -16,7 +16,7 @@
 
 # Canonical product version for CoderAI — single source of truth. Both the API
 # metadata and the admin web UI read from here.
-__version__ = "0.1.47"
+__version__ = "0.1.48"
 
 # Configure the CUDA caching allocator BEFORE torch is imported anywhere.
 # expandable_segments lets the allocator return freed pages to the driver even

codai/api/ratelimit.py

@@ -160,10 +160,22 @@ class RateLimitMiddleware(BaseHTTPMiddleware):
         if path in self._EXEMPT_PATHS:
             return await call_next(request)
 
-        # Queue-size enforcement for authenticated API requests (not for status polls)
+        # Queue-size enforcement for authenticated API requests (not for status
+        # polls). PER-MODEL: the request is rejected only when its own model's
+        # queue is full — other loaded models keep accepting and run in
+        # parallel. (request.body() caches, so downstream handlers still read it.)
         if path not in self._EXEMPT_PATHS and any(path.startswith(p) for p in _QUEUED_PREFIXES):
             from codai.queue.manager import queue_manager
-            if await queue_manager.is_full():
+            _model = None
+            if request.method == "POST" and "json" in (
+                    request.headers.get("content-type") or ""):
+                try:
+                    import json as _json
+                    _model = (_json.loads(await request.body() or b"{}")
+                              or {}).get("model")
+                except Exception:
+                    _model = None
+            if await queue_manager.is_full_for(_model or ""):
                 return JSONResponse(
                     status_code=429,
                     content={

codai/frontproxy/app.py

@@ -697,6 +697,20 @@ class FrontProxy:
         engine name for inference without an explicit model."""
         return self._queue_key(model) or getattr(engine, "name", "") or "?"
 
+    def _resident_on(self, engine, key: str) -> bool:
+        """True when the model behind `key` is already loaded on `engine` (per
+        its last health poll), so serving it needs NO swap/eviction."""
+        try:
+            if not key:
+                return False
+            for m in (engine.loaded_models or ()):
+                canon = (self._model_info(str(m)).get("model_id") or str(m)).lower()
+                if canon == key or str(m).lower() == key:
+                    return True
+        except Exception:
+            pass
+        return False
+
     async def _swap_acquire(self, engine, model, path, method):
         """Acquire this engine's shared-GPU swap slot for a GPU-inference request.
         Returns a (gate, key) token for _swap_release, or None when no gate applies
@@ -707,6 +721,13 @@ class FrontProxy:
         if gate is None:
             return None
         key = self._swap_owner_key(engine, model)
+        # A model that is ALREADY resident on this engine needs no swap — its
+        # own per-model queue governs its concurrency, and serializing it
+        # behind the current GPU owner would block cross-model parallelism
+        # (e.g. three co-resident embedders serving one request each). The gate
+        # only serializes requests that would trigger an actual model swap.
+        if self._resident_on(engine, key):
+            return None
         await gate.acquire(key)
         return (gate, key)
 

codai/queue/manager.py

@@ -91,6 +91,18 @@ class QueueManager:
         async with self.lock:
             return len(self.waiting) >= self.max_size
 
+    async def is_full_for(self, model_key: str) -> bool:
+        """PER-MODEL admission: the queue is full for a request only when THAT
+        model already has max_size waiters — one hot model's backlog must not
+        reject requests for other (loaded, idle) models. A generous global
+        backstop (4× max_size across all models) still bounds total memory."""
+        async with self.lock:
+            if len(self.waiting) >= self.max_size * 4:
+                return True
+            if not model_key:
+                return len(self.waiting) >= self.max_size
+            return self._waiting_counts_locked().get(model_key, 0) >= self.max_size
+
     async def acquire(self, request_id: str, model_key: str,
                       prefix_key: str = "") -> SchedulerLease:
         waiter = None