front: burst to RunPod when the local model is busy, and to pods

Spillover already existed, but not in the shape that is actually useful:
it fired only when the WAIT QUEUE overflowed (on_concurrency_full) or
when no GPU could serve the model, and it could only reach a serverless
endpoint — the code said outright that a pods target "goes through the
engine and is not handled here", which meant bursting to our own pods was
not implemented at all.

Two additions:

`on_busy` — offload the moment there is no free local slot, instead of
queueing behind the running generation. Built on FrontQueue.try_acquire,
a non-blocking take: the front has to learn the model is occupied WITHOUT
committing to wait for it. Wired into both the buffered and the SSE
dispatch paths.

`target.mode: "pods"` — borrow a pod from the managed pool (provision on
demand, health-gate, scale, reap, bill against the same caps) for one
request and hand it straight back, so an overlap that never recurs is
reaped rather than left running. Serverless and pods now share one
`_spill_open` primitive, with a Response wrapper for the buffered path
and a chunk generator for the SSE path.

Local stays the safety net in both directions: a burst that cannot be
opened falls back to the local queue, never the reverse.

Two bugs caught before shipping, both in my own first cut: the direct
path referenced _swap_tok before it is assigned (NameError on every
burst), and the SSE path left _qkey set after bursting, so cleanup would
have released a slot it never held — handing a phantom slot to the next
waiter and over-subscribing the model.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0182DhR1eFNPQyGmDrpHbedY

codai/admin/routes.py

@@ -2826,7 +2826,10 @@ async def api_model_configure(request: Request, username: str = Depends(require_
     if "runpod_spillover" in data:
         src = data.get("runpod_spillover") if isinstance(data.get("runpod_spillover"), dict) else {}
         sp = {}
-        for k in ("enabled", "on_concurrency_full", "on_no_gpu", "on_local_error"):
+        # on_busy: offload as soon as the local model has no free slot, rather
+        # than waiting for the wait-queue to overflow (on_concurrency_full).
+        for k in ("enabled", "on_busy", "on_concurrency_full", "on_no_gpu",
+                  "on_local_error"):
             if k in src and src.get(k) is not None and src.get(k) != "":
                 sv = src.get(k)
                 sp[k] = (sv.lower() in ("1", "true", "on", "yes")

codai/frontproxy/reqqueue.py

@@ -91,6 +91,23 @@ class FrontQueue:
                     await self._release_locked(key)
             raise
 
+    async def try_acquire(self, key: str, capacity: int) -> bool:
+        """Take a slot only if one is free RIGHT NOW; never wait.
+
+        This is what "burst on busy" is built on: a model configured to offload
+        rather than queue needs to know the local model is occupied *before*
+        committing to wait for it. A False here means the next request would have
+        been queued behind a running generation.
+        """
+        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 True
+        return False
+
     async def release(self, key: str) -> None:
         key = key or ""
         async with self._lock:

docs/remote-execution.md

@@ -333,6 +333,51 @@ Notes:
 * Requests larger than `max_body_mb` are served locally rather than buffered in
   RAM; raise it if you forward large uploads.
 
+## Per-model placement: local, remote, or burst
+
+Three answers to "where does this model run", set per model.
+
+**Always remote** — `backend: "runpod"` on the model entry, with a `runpod` block
+choosing pods or serverless. No local VRAM, nothing downloaded here.
+
+**Local only** — the default. Requests queue behind the running generation, up to
+`max_instances` concurrently and `queue_max_waiting` in the queue.
+
+**Local first, burst on overlap** — local serves what it can; overflow goes to
+RunPod:
+
+```json
+{
+  "path": "/AI/guffcache/gemma-4-31B-Q4_0.gguf",
+  "max_instances": 1,
+  "runpod_spillover": {
+    "enabled": true,
+    "on_busy": true,
+    "target": {"mode": "pods", "engine": "llamacpp",
+               "hf_gguf": "bartowski/gemma-4-31B-GGUF:Q4_0",
+               "max_hourly_usd": 0.8, "idle_timeout_s": 180}
+  }
+}
+```
+
+Triggers, most eager first:
+
+| Trigger | Fires when |
+|---|---|
+| `on_busy` | no free local slot right now — the second concurrent request offloads instead of queueing |
+| `on_concurrency_full` | the wait queue itself is full (would otherwise be a 503) |
+| `on_no_gpu` | no local GPU can serve the model at all |
+
+`target.mode` is `pods` (a coderai-managed pod: provisioned on demand, health-
+gated, scaled, reaped, billed against the same caps) or `serverless` (your own
+RunPod endpoint id). `served_model` renames the model for the remote.
+
+The burst is a borrow: a pod is taken for one request and handed straight back,
+so an overlap that never recurs is reaped after `idle_timeout_s` rather than
+lingering. And a burst that cannot be opened — cold pod failed, budget cap hit,
+misconfigured — falls back to the local queue rather than erroring: local is
+always the safety net, never the other way round.
+
 ## Deployment shapes this enables
 
 * **One small coderai, many remote GPUs** — the local instance holds the API,

tests/test_burst_placement.py

+"""Per-model placement: local / remote / burst-to-RunPod.
+
+The queue arithmetic is what matters here — a leaked slot silently caps a model's
+concurrency forever, and a phantom release hands a slot to a waiter that should
+still be waiting.
+"""
+
+import asyncio
+import sys
+from pathlib import Path
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from codai.frontproxy.reqqueue import FrontQueue, QueueFull
+
+
+def run(coro):
+    return asyncio.get_event_loop_policy().new_event_loop().run_until_complete(coro)
+
+
+def test_try_acquire_takes_a_free_slot_and_refuses_when_full():
+    async def go():
+        q = FrontQueue()
+        assert await q.try_acquire("m", 1) is True      # free
+        assert await q.try_acquire("m", 1) is False     # busy -> burst instead
+        await q.release("m")
+        assert await q.try_acquire("m", 1) is True      # freed again
+    run(go())
+
+
+def test_try_acquire_respects_capacity_above_one():
+    async def go():
+        q = FrontQueue()
+        assert [await q.try_acquire("m", 3) for _ in range(3)] == [True, True, True]
+        assert await q.try_acquire("m", 3) is False
+    run(go())
+
+
+def test_try_acquire_never_waits_even_with_a_queue_behind_it():
+    async def go():
+        q = FrontQueue()
+        await q.acquire("m", 1, 4)                       # slot held
+        waiter = asyncio.ensure_future(q.acquire("m", 1, 4))
+        await asyncio.sleep(0)                           # let it enqueue
+        # The point of try_acquire: it returns rather than joining that queue.
+        assert await asyncio.wait_for(q.try_acquire("m", 1), timeout=0.5) is False
+        await q.release("m")                             # hands the slot to waiter
+        await asyncio.wait_for(waiter, timeout=0.5)
+        await q.release("m")
+    run(go())
+
+
+def test_a_granted_slot_is_not_double_counted():
+    """try_acquire must take exactly one slot — the burst path then skips the
+    blocking acquire, and releasing once must fully free the model."""
+    async def go():
+        q = FrontQueue()
+        assert await q.try_acquire("m", 1) is True
+        await q.release("m")
+        # If try_acquire had counted twice, this would still be occupied.
+        assert await q.try_acquire("m", 1) is True
+    run(go())
+
+
+def test_queue_full_still_raises_for_non_burst_models():
+    async def go():
+        q = FrontQueue()
+        await q.acquire("m", 1, 1)
+        w = asyncio.ensure_future(q.acquire("m", 1, 1))
+        await asyncio.sleep(0)
+        with pytest.raises(QueueFull):
+            await q.acquire("m", 1, 1)
+        w.cancel()
+        try:
+            await w
+        except asyncio.CancelledError:
+            pass
+    run(go())
+
+
+# --------------------------------------------------------------------------- #
+# the front's placement decision
+# --------------------------------------------------------------------------- #
+class _Front:
+    """Just the bits of the front the placement decision touches."""
+
+    from codai.frontproxy.app import FrontProxy as _FP
+    _spill_on_busy = _FP._spill_on_busy
+
+    def __init__(self, info):
+        self._info = info
+
+    def _model_info(self, model):
+        return self._info
+
+
+def test_spill_on_busy_only_when_enabled_and_triggered():
+    assert _Front({}) ._spill_on_busy("m") is False
+    assert _Front({"runpod_spillover": {"enabled": True}})._spill_on_busy("m") is False
+    assert _Front({"runpod_spillover": {"on_busy": True}})._spill_on_busy("m") is False
+    assert _Front({"runpod_spillover": {"enabled": True, "on_busy": True}}) \
+        ._spill_on_busy("m") is True
+    # A model that only bursts when the QUEUE overflows must not burst on busy.
+    assert _Front({"runpod_spillover": {"enabled": True, "on_concurrency_full": True}}) \
+        ._spill_on_busy("m") is False
+
+
+def test_model_is_renamed_for_the_remote():
+    import json
+    from codai.frontproxy.app import FrontProxy
+
+    body = json.dumps({"model": "local-name", "messages": []}).encode()
+    assert json.loads(FrontProxy._rewrite_model(body, "remote-name"))["model"] \
+        == "remote-name"
+    # No served_model configured -> body untouched, byte for byte.
+    assert FrontProxy._rewrite_model(body, "") is body
+    # Garbage in, same garbage out rather than an exception.
+    assert FrontProxy._rewrite_model(b"not json", "x") == b"not json"