front: own the generation queue (per-model admission gate)

Move request queuing from the engine to the front so the engine does only
generation. New codai/frontproxy/reqqueue.py FrontQueue: a per-model concurrency
gate sized to the model's max_instances (per-model from models.json, else the
global models.max_model_instances). A text-generation request acquires a slot
before dispatch; when all slots are busy it waits (FIFO) and shows as "queued"
with its position on the Tasks page; beyond server.queue_max_size waiting it gets
HTTP 503. A client disconnect while queued cancels the wait and drops it from the
queue with no slot leak.

Wired into both dispatch paths in app.py (direct proxy + broker route) so direct
and brokered requests share one queue, scoped to text generation only (images/
audio/embeddings pass through unqueued). Slots are released on stream completion,
send failure, or disconnect. Queue key is the canonical model id so all aliases/
paths of a model share one gate. Front never over-subscribes the engine, so the
engine's own queue stays a safety net that effectively never fills.

The engine-side QueueManager + KV-affinity instance pool are intentionally left
in place for now (trimming them is a follow-up once this is verified live).

Verified: FrontQueue unit tests (capacity, FIFO hand-off, QueueFull, cancel
cleanup, position, capacity>1) and a FrontProxy integration test (capacity
resolution, alias-key collapsing, queued task surfacing).
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_011DDv7BchtZQWsnPG6Jm49m

codai/frontproxy/app.py

@@ -27,6 +27,7 @@ from starlette.background import BackgroundTask
 from codai.frontproxy.registry import EngineRegistry
 from codai.frontproxy.engine_supervisor import EngineSupervisor
 from codai.frontproxy import router as _router
+from codai.frontproxy.reqqueue import FrontQueue, QueueFull
 
 # Hop-by-hop headers that must not be forwarded verbatim (RFC 7230 §6.1) plus
 # length/host headers that the client/StreamingResponse recompute.
@@ -60,6 +61,9 @@ class FrontProxy:
         # freshly-added one) would flicker out of the aggregated /v1/models while
         # it loads. Keyed by engine name.
         self._engine_models_cache: dict = {}
+        # Front-managed generation queue: admission control + ordering + queue
+        # position, sized per-model to max_instances so the engine never queues.
+        self.reqqueue = FrontQueue()
         self.registry = EngineRegistry()
         self.supervisor: Optional[EngineSupervisor] = None
         # Per-run secret shared only with the engines (passed via env at spawn). The
@@ -226,6 +230,22 @@ class FrontProxy:
         # Signed with the internal token; the engine accepts it only if it matches.
         if self.internal_token:
             send_headers["x-coderai-broker-authed"] = self.internal_token
+        # Front-managed generation queue (text only) — same per-model gate as the
+        # direct proxy path, so brokered and direct requests share one queue.
+        _qkey = None
+        if (method.upper() == "POST" and _router.is_inference_path(path)
+                and self._task_kind(path) == "text"):
+            _qkey = self._queue_key(model)
+            try:
+                await self.reqqueue.acquire(
+                    _qkey, self._model_capacity(model), self._queue_max_waiting(),
+                    rid=engine.name + ":" + (model or ""), model=model or "",
+                    engine=engine.name)
+            except QueueFull:
+                return {"status_code": 503,
+                        "headers": {"content-type": "application/json"},
+                        "body": b'{"error":"Server busy: the generation queue is '
+                                b'full, please retry shortly."}'}
         # Count brokered inference as in-flight too (with metadata) so it shows on
         # the Tasks page even when the engine is too busy to report it itself.
         _rid = engine.enter_request(
@@ -244,6 +264,8 @@ class FrontProxy:
                              % (engine.id, exc)).encode()}
         finally:
             engine.exit_request(_rid)
+            if _qkey is not None:
+                await self.reqqueue.release(_qkey)
         # Surface the engine's actual reply so a brokered request that "doesn't get
         # executed" (e.g. an instant small error body) is diagnosable from the log.
         print("[front] broker route: %s %s -> engine#%s(%s) status=%s bytes=%d preview=%r"
@@ -299,7 +321,10 @@ class FrontProxy:
                        # list should display so each model shows once, by its id.
                        "model_id": (m.get("id") or m.get("path")
                                     or m.get("alias") or "").strip() or None,
-                       "engine_fallback": bool(m.get("engine_fallback"))}
+                       "engine_fallback": bool(m.get("engine_fallback")),
+                       # Per-model concurrency ceiling — the front queue sizes its
+                       # gate to this so it never over-subscribes the engine.
+                       "max_instances": m.get("max_instances")}
                 for field_ in (m.get("path"), m.get("id"), m.get("alias")):
                     if not field_:
                         continue
@@ -315,6 +340,28 @@ class FrontProxy:
                         info[f[:-5]] = rec
         return info
 
+    def _queue_key(self, model: Optional[str]) -> str:
+        """Stable gate key for a model: its canonical id (so every alias/path of
+        the same model shares one gate), else the raw model string."""
+        info = self._model_info(model)
+        return (info.get("model_id") or model or "").lower()
+
+    def _model_capacity(self, model: Optional[str]) -> int:
+        """Per-model concurrency = its max_instances, falling back to the global
+        server default. This is the number of front queue slots for the model."""
+        info = self._model_info(model)
+        mi = info.get("max_instances")
+        if mi:
+            try:
+                return max(1, int(mi))
+            except (TypeError, ValueError):
+                pass
+        _models = getattr(self.config, "models", None)
+        return max(1, int(getattr(_models, "max_model_instances", 1) or 1))
+
+    def _queue_max_waiting(self) -> int:
+        return max(0, int(getattr(self.config.server, "queue_max_size", 6) or 0))
+
     def _required_cap(self, path: str, model: Optional[str]) -> Optional[str]:
         ds4 = getattr(self.config, "ds4", None)
         info = self._model_info(model)
@@ -704,6 +751,26 @@ class FrontProxy:
                     "pausable": False,
                     "restartable": False,
                 })
+        # Front-queued (not yet running) generations: requests waiting on the
+        # front's per-model gate. The engine hasn't seen them yet, so only the
+        # front can report them — show each with its queue position.
+        for q in self.reqqueue.snapshot():
+            merged.append({
+                "id": f"queued-{q.get('rid')}-{q.get('position')}",
+                "kind": "text",
+                "title": (q.get("model") or "generation"),
+                "model": q.get("model") or "",
+                "status": "queued",
+                "position": q.get("position"),
+                "step": 0, "total": 0, "rate": 0.0,
+                "message": f"queued (#{q.get('position')})",
+                "started_at": q.get("enqueued_at"),
+                "engine": q.get("engine"),
+                "active": True,
+                "cancellable": True,
+                "pausable": False,
+                "restartable": False,
+            })
         return merged
 
     # -------------------------------------------------------------------- proxy
@@ -729,6 +796,26 @@ class FrontProxy:
                 {"error": "No engine is ready yet (still starting/loading)."},
                 status_code=503)
 
+        # Front-managed generation queue (text only). Acquire a per-model slot
+        # before dispatching: if all max_instances slots are busy this awaits
+        # (showing as "queued" on the Tasks page) until one frees; if too many are
+        # already waiting it returns 503. A client disconnect while queued cancels
+        # this await and drops it from the queue. Other inference kinds (images,
+        # audio, embeddings…) pass through unqueued.
+        _qkey = None
+        if (method == "POST" and _router.is_inference_path(path)
+                and self._task_kind(path) == "text"):
+            _qkey = self._queue_key(model)
+            try:
+                await self.reqqueue.acquire(
+                    _qkey, self._model_capacity(model), self._queue_max_waiting(),
+                    rid=engine.name + ":" + (model or ""), model=model or "",
+                    engine=engine.name)
+            except QueueFull:
+                return JSONResponse(
+                    {"error": "Server busy: the generation queue is full, "
+                              "please retry shortly."}, status_code=503)
+
         url = engine.url + path
         headers = self._filter_headers(request.headers, _DROP_REQ)
         content = body_bytes if body_bytes is not None else request.stream()
@@ -747,6 +834,8 @@ class FrontProxy:
             rp_resp = await self._long.send(rp_req, stream=True)
         except Exception as exc:
             engine.exit_request(_rid)
+            if _qkey is not None:
+                await self.reqqueue.release(_qkey)
             return JSONResponse(
                 {"error": f"Engine#{engine.id} unreachable: {exc}"}, status_code=502)
 
@@ -755,6 +844,8 @@ class FrontProxy:
                 await rp_resp.aclose()
             finally:
                 engine.exit_request(_rid)
+                if _qkey is not None:
+                    await self.reqqueue.release(_qkey)
 
         # Measure throughput from the SSE stream the front relays, and publish it on
         # the in-flight metadata. This gives the Tasks page a live it/s for the

codai/frontproxy/reqqueue.py

+# CoderAI - OpenAI-compatible API server
+# Copyright (C) 2026 Stefy Lanza <stefy@nexlab.net>
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+
+"""Front-managed request queue (admission control) for text generation.
+
+Architecture: the engine handles only generation; the front owns the queue. The
+front sizes a per-model concurrency gate to the model's ``max_instances`` and
+never dispatches more than that many concurrent generations to the engine — so
+the engine's own queue effectively never fills, while ordering, queue depth,
+queue position and the "queued" Tasks entries are all owned and reported here.
+
+Each model key gets ``capacity`` slots. ``acquire()`` grants a slot immediately
+when one is free, otherwise enqueues an asyncio waiter (FIFO) and blocks until a
+slot frees. Beyond ``max_waiting`` queued requests it raises :class:`QueueFull`
+(the caller returns HTTP 503). A client that disconnects while queued cancels its
+``acquire()`` await, which drops it from the queue with no slot leak.
+"""
+import asyncio
+
+
+class QueueFull(Exception):
+    """Raised by FrontQueue.acquire when the per-model wait queue is at capacity."""
+
+    def __init__(self, depth: int):
+        super().__init__(f"queue full ({depth} waiting)")
+        self.depth = depth
+
+
+class _Waiter:
+    __slots__ = ("rid", "model", "engine", "enqueued_at", "event")
+
+    def __init__(self, rid, model, engine, enqueued_at):
+        self.rid = rid
+        self.model = model
+        self.engine = engine
+        self.enqueued_at = enqueued_at
+        self.event = asyncio.Event()
+
+
+class FrontQueue:
+    """Per-model concurrency gate with a bounded FIFO wait queue."""
+
+    def __init__(self):
+        self._lock = asyncio.Lock()
+        self._active: dict = {}     # model_key -> slots currently in use
+        self._waiting: dict = {}    # model_key -> list[_Waiter] (FIFO)
+
+    async def acquire(self, key: str, capacity: int, max_waiting: int,
+                      rid: str = "", model: str = "", engine: str = "") -> None:
+        """Acquire a generation slot for ``key``.
+
+        Returns once a slot is held (immediately, or after waiting). Raises
+        :class:`QueueFull` if ``max_waiting`` requests are already queued. The
+        caller MUST pair every successful acquire with exactly one
+        :meth:`release` for the same ``key``.
+        """
+        key = key or ""
+        capacity = max(1, int(capacity or 1))
+        async with self._lock:
+            active = self._active.get(key, 0)
+            if active < capacity:
+                self._active[key] = active + 1
+                return
+            wq = self._waiting.setdefault(key, [])
+            if len(wq) >= max(0, int(max_waiting)):
+                raise QueueFull(len(wq))
+            import time as _t
+            waiter = _Waiter(rid=rid, model=model, engine=engine, enqueued_at=_t.time())
+            wq.append(waiter)
+
+        # Wait outside the lock. On grant, the releasing task has handed us its
+        # slot (the active count was kept constant), so we must NOT re-increment.
+        try:
+            await waiter.event.wait()
+        except BaseException:
+            # Cancelled (e.g. client disconnected) or errored while queued.
+            async with self._lock:
+                wq = self._waiting.get(key) or []
+                if waiter in wq:
+                    wq.remove(waiter)            # still queued — never held a slot
+                    if not wq:
+                        self._waiting.pop(key, None)
+                else:
+                    # Already granted a slot in the race with cancellation — give
+                    # it back so the next waiter (or the count) is correct.
+                    await self._release_locked(key)
+            raise
+
+    async def release(self, key: str) -> None:
+        key = key or ""
+        async with self._lock:
+            await self._release_locked(key)
+
+    async def _release_locked(self, key: str) -> None:
+        wq = self._waiting.get(key) or []
+        if wq:
+            # Hand the freed slot directly to the next waiter: active count stays
+            # constant (one finishes, one starts).
+            waiter = wq.pop(0)
+            if not wq:
+                self._waiting.pop(key, None)
+            waiter.event.set()
+        else:
+            n = self._active.get(key, 0) - 1
+            if n > 0:
+                self._active[key] = n
+            else:
+                self._active.pop(key, None)
+
+    def snapshot(self) -> list:
+        """Best-effort list of currently-queued (not yet running) requests, for the
+        Tasks page. Each entry: {rid, model, engine, position, enqueued_at}."""
+        out = []
+        for key, wq in list(self._waiting.items()):
+            for i, w in enumerate(list(wq)):
+                out.append({"rid": w.rid, "model": w.model, "engine": w.engine,
+                            "position": i + 1, "enqueued_at": w.enqueued_at})
+        return out