runpod: place any model individually, keep orchestration local, ship LoRAs

PER-MODEL PLACEMENT FOR ANY KIND. Sending a model to RunPod used to work
only for text: resolve_pod_engine looked at whether the path ended in
.gguf, and RunpodBackend implements chat only. A diffusion model pinned
to runpod would have been handed to vLLM, which cannot serve it. Now the
model's OWN KIND picks the pod: an image/video/TTS/STT/embedding model
gets that capability's published coderai image, a GGUF gets llama.cpp, an
HF repo gets vLLM. The gateway routes by model BEFORE capability, so two
models of the same kind can differ — one video model pinned local with
`placement: "local"` while another runs on its own pod with its own
budget, which capability-level routing could never express.

A pod rented for one model starts with an EMPTY catalogue and would
refuse every request for it, so pod_plan seeds CODERAI_SEED_MODELS with
that model's entry and the pod registers it at startup (in memory only —
a pod is disposable, and writing it would surprise anyone setting the
variable on a real install). Only fields that describe the model travel:
never local paths, which mean nothing there, and never this deployment's
placement settings, which would have the pod renting pods of its own.

ORCHESTRATION STAYS LOCAL. /v1/pipelines, /v1/characters and
/v1/environments are no longer gatewayable capabilities. They are a
sequence of calls to other endpoints, and they already dispatch each step
through the front as its own /v1 request — so every step is placed by its
own model's config, and a chain can mix local and remote freely.
Forwarding the chain would have moved the logic to a pod and placed every
step by that pod's catalogue instead.

LORAS FOLLOW THE REQUEST. A LoRA is a file on this machine; a pod has
never seen it. The blob store is content-addressed, so each adapter is
hashed, checked against the remote's /v1/loras/blob/<hash>, uploaded only
when missing, and the reference rewritten to that id — repeat requests
upload nothing. QLoRA needs nothing extra: the quantisation is in how the
base model loads, the adapter is the same safetensors file. An HF id or
URL is left alone for the remote to fetch.

Also: a vLLM pod now refuses a local model path up front, with the same
reasoning as the GGUF guard — otherwise it boots for minutes and dies —
and passes `quantization` through, since a local load_in_4bit does not
carry over to vLLM.
Co-Authored-By: 's avatarClaude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0182DhR1eFNPQyGmDrpHbedY
parent 3acf9102
...@@ -72,9 +72,6 @@ _CAPABILITY_PREFIXES = ( ...@@ -72,9 +72,6 @@ _CAPABILITY_PREFIXES = (
("/v1/rerank", "rerank"), ("/v1/rerank", "rerank"),
("/v1/ocr", "ocr"), ("/v1/ocr", "ocr"),
("/v1/loras", "loras"), ("/v1/loras", "loras"),
("/v1/characters", "characters"),
("/v1/environments", "environments"),
("/v1/pipelines", "pipelines"),
("/v1/faceswap", "faceswap"), ("/v1/faceswap", "faceswap"),
) )
...@@ -105,6 +102,14 @@ def file_origin(filename: str) -> str: ...@@ -105,6 +102,14 @@ def file_origin(filename: str) -> str:
#: Handled by RemoteOpenAIBackend instead — see the module docstring. #: Handled by RemoteOpenAIBackend instead — see the module docstring.
_SKIP_PATHS = ("/v1/chat/completions", "/v1/completions", "/v1/models") _SKIP_PATHS = ("/v1/chat/completions", "/v1/completions", "/v1/models")
#: Orchestration endpoints: a sequence of calls to other endpoints, not a model.
#: They are NEVER forwarded as a whole — the chain stays here and each step goes
#: out as its own /v1 request through the front (codai/broker/asgi_bridge.py), so
#: every step is placed by its OWN model's configuration. Forwarding the
#: orchestration instead would move the logic to a pod and place every step by
#: that pod's catalogue, which is the opposite of what these are for.
_ORCHESTRATION_PREFIXES = ("/v1/pipelines", "/v1/characters", "/v1/environments")
#: `model` field in a multipart body, without paying for a full form parse. #: `model` field in a multipart body, without paying for a full form parse.
_MULTIPART_MODEL = re.compile( _MULTIPART_MODEL = re.compile(
rb'name="model"\r?\n(?:[^\r\n]*\r?\n)*?\r?\n([^\r\n]*)', re.IGNORECASE) rb'name="model"\r?\n(?:[^\r\n]*\r?\n)*?\r?\n([^\r\n]*)', re.IGNORECASE)
...@@ -168,6 +173,34 @@ def capability_pods() -> dict: ...@@ -168,6 +173,34 @@ def capability_pods() -> dict:
for k, v in pods.items()} for k, v in pods.items()}
def model_placement(model: str) -> tuple:
"""Where this specific model runs: ("local"|"url"|"pod"|"", detail).
Per-model placement beats the capability map, so two models of the SAME kind
can differ — one video model pinned local while another runs on a pod. A
model with nothing set returns "" and falls through to the capability
setting.
"""
if not model:
return "", None
try:
from codai.models.manager import _model_entry_for
entry = _model_entry_for(model) or {}
except Exception:
entry = {}
# An explicit local pin opts this model OUT of a capability-wide remote.
if str(entry.get("placement") or "").strip().lower() == "local":
return "local", None
url = _model_remote(model)
if url:
return "url", url
backend = str(entry.get("backend") or "").strip().lower()
block = entry.get("runpod") if isinstance(entry.get("runpod"), dict) else None
if backend == "runpod" and block is not None:
return "pod", (entry, block)
return "", None
def _model_remote(model: str) -> str: def _model_remote(model: str) -> str:
if not model: if not model:
return "" return ""
...@@ -260,6 +293,8 @@ def resolve_target(path: str, method: str, query: str, body: bytes, ...@@ -260,6 +293,8 @@ def resolve_target(path: str, method: str, query: str, body: bytes,
"""Decide where this request should go, or None to serve it locally.""" """Decide where this request should go, or None to serve it locally."""
if not path.startswith("/v1/") or path in _SKIP_PATHS: if not path.startswith("/v1/") or path in _SKIP_PATHS:
return None return None
if path.startswith(_ORCHESTRATION_PREFIXES):
return None
# A generated file is fetched from whichever remote rendered it. # A generated file is fetched from whichever remote rendered it.
if path.startswith("/v1/files/"): if path.startswith("/v1/files/"):
base = file_origin(path[len("/v1/files/"):]) base = file_origin(path[len("/v1/files/"):])
...@@ -269,9 +304,18 @@ def resolve_target(path: str, method: str, query: str, body: bytes, ...@@ -269,9 +304,18 @@ def resolve_target(path: str, method: str, query: str, body: bytes,
if not model and query: if not model and query:
from urllib.parse import parse_qs from urllib.parse import parse_qs
model = (parse_qs(query).get("model") or [""])[0] model = (parse_qs(query).get("model") or [""])[0]
url = _model_remote(model)
if url: # Per model first, so two models of the same kind can be placed differently.
return Target(url=url, reason=f"model {model!r}") where, detail = model_placement(model)
if where == "local":
return None # pinned here, whatever the capability says
if where == "url":
return Target(url=detail, reason=f"model {model!r}")
if where == "pod":
entry, block = detail
from codai.api.runpod_worker import get_model_pod_pool
return Target(pool=get_model_pod_pool(model, entry, block),
reason=f"model {model!r} on its own RunPod pod")
if not cap: if not cap:
return None return None
url = capability_endpoints().get(cap, "") url = capability_endpoints().get(cap, "")
...@@ -352,6 +396,10 @@ class RemoteGatewayMiddleware: ...@@ -352,6 +396,10 @@ class RemoteGatewayMiddleware:
return await _error(send, 502, return await _error(send, 502,
f"could not reach a remote for this request: {exc}") f"could not reach a remote for this request: {exc}")
try: try:
# Ship any LoRA the request names, so the remote can actually load it.
if b'"loras"' in (body or b"") and "json" in headers.get("content-type", ""):
body = await asyncio.to_thread(
sync_loras, base, target.api_key, body)
await self._send_upstream(scope, headers, body, base, target.reason, send, await self._send_upstream(scope, headers, body, base, target.reason, send,
api_key=target.api_key) api_key=target.api_key)
finally: finally:
...@@ -441,6 +489,77 @@ def _rewrite_file_urls(raw: bytes, remote_base: str, local_base: str) -> bytes: ...@@ -441,6 +489,77 @@ def _rewrite_file_urls(raw: bytes, remote_base: str, local_base: str) -> bytes:
return _FILE_URL.sub(_sub, raw) return _FILE_URL.sub(_sub, raw)
def sync_loras(base: str, api_key: str, body: bytes) -> bytes:
"""Make sure a remote can load the LoRAs this request names.
A LoRA lives as a file on THIS machine — trained here, or uploaded here. A
pod has never seen it, so a request naming `{"model": "/AI/loras/x.safetensors"}`
would fail there. The blob store is content-addressed, so each adapter is
hashed, checked against the remote's /v1/loras/blob/<hash>, uploaded only
when missing, and the reference rewritten to that id. Repeat requests upload
nothing.
QLoRA adapters need no special handling: the quantisation lives in how the
BASE model is loaded, while the adapter is the same safetensors file.
Anything that cannot be resolved to a local file — an HF repo id, a URL —
is left untouched: the remote can fetch those itself. Failure anywhere
returns the body unchanged, so the request still goes out and fails (or
succeeds) on the remote's own terms rather than here.
"""
import hashlib
import requests
try:
obj = json.loads(body or b"{}")
except Exception:
return body
if not isinstance(obj, dict):
return body
specs = obj.get("loras")
if not isinstance(specs, list) or not specs:
return body
from codai.api.loras import resolve_lora_ref
headers = {"Authorization": f"Bearer {api_key}"} if api_key else {}
changed = False
for spec in specs:
if not isinstance(spec, dict):
continue
if str(spec.get("id") or "").startswith("sha256:"):
continue # already content-addressed
try:
local = resolve_lora_ref(spec)
except Exception:
local = None
if not local or not os.path.isfile(local):
continue # an HF id or URL — the remote resolves it
try:
with open(local, "rb") as fh:
data = fh.read()
digest = hashlib.sha256(data).hexdigest()
have = requests.get(f"{base}/v1/loras/blob/{digest}",
headers=headers, timeout=30)
if have.status_code != 200:
up = requests.post(f"{base}/v1/loras/upload", data=data,
headers={**headers,
"Content-Type": "application/octet-stream"},
timeout=1800)
up.raise_for_status()
print(f"[remote-gateway] uploaded LoRA {os.path.basename(local)} "
f"({len(data) / 1e6:.0f} MB) to {base}", flush=True)
except Exception as exc:
print(f"[remote-gateway] could not send LoRA {local} to {base}: {exc}",
flush=True)
continue
# Keep weight/name; drop the local path, which means nothing over there.
for key in ("model", "path", "url", "file", "data"):
spec.pop(key, None)
spec["id"] = f"sha256:{digest}"
changed = True
return json.dumps(obj).encode() if changed else body
async def _error(send, status: int, message: str): async def _error(send, status: int, message: str):
msg = json.dumps({"error": {"message": message, msg = json.dumps({"error": {"message": message,
"type": "remote_gateway_error"}}).encode() "type": "remote_gateway_error"}}).encode()
......
This diff is collapsed.
...@@ -978,6 +978,12 @@ class ConfigManager: ...@@ -978,6 +978,12 @@ class ConfigManager:
"aliases": {} "aliases": {}
} }
# A pod rented for one model starts with an empty catalogue and would
# refuse every request for it. CODERAI_SEED_MODELS carries the entries it
# must know about, merged in at startup (existing entries win, so this
# can never overwrite a real deployment's catalogue).
self._seed_models_from_env()
# Load auth.json # Load auth.json
if self.auth_path.exists(): if self.auth_path.exists():
with open(self.auth_path, 'r') as f: with open(self.auth_path, 'r') as f:
...@@ -1298,6 +1304,39 @@ class ConfigManager: ...@@ -1298,6 +1304,39 @@ class ConfigManager:
with open(self.config_path, 'w') as f: with open(self.config_path, 'w') as f:
json.dump(config_dict, f, indent=2) json.dump(config_dict, f, indent=2)
def _seed_models_from_env(self) -> None:
"""Register models named by CODERAI_SEED_MODELS (JSON list of entries).
Used by RunPod capability pods, which are rented for a specific model and
have no models.json of their own. Never persisted: a pod is disposable,
and writing it would surprise anyone who set the variable on a real
installation.
"""
raw = os.environ.get("CODERAI_SEED_MODELS", "").strip()
if not raw:
return
try:
entries = json.loads(raw)
except Exception as exc:
print(f"[seed] CODERAI_SEED_MODELS is not valid JSON: {exc}", flush=True)
return
if isinstance(entries, dict):
entries = [entries]
if not isinstance(entries, list):
return
for e in entries:
if not isinstance(e, dict) or not e.get("path"):
continue
section = e.get("model_type") or "text_models"
lst = self.models_data.setdefault(section, [])
if not isinstance(lst, list):
continue
path = str(e["path"])
if any(isinstance(m, dict) and m.get("path") == path for m in lst):
continue # a real entry already covers it — leave it alone
lst.append(dict(e))
print(f"[seed] registered {path} in {section}", flush=True)
def save_models(self): def save_models(self):
"""Save models.json to disk.""" """Save models.json to disk."""
with open(self.models_path, 'w') as f: with open(self.models_path, 'w') as f:
......
...@@ -476,8 +476,11 @@ def test_every_v1_endpoint_is_either_mapped_or_deliberately_skipped(): ...@@ -476,8 +476,11 @@ def test_every_v1_endpoint_is_either_mapped_or_deliberately_skipped():
paths = {m for f in root.glob("*.py") for m in pat.findall(f.read_text())} paths = {m for f in root.glob("*.py") for m in pat.findall(f.read_text())}
assert paths, "no /v1 routes found — did the decorator style change?" assert paths, "no /v1 routes found — did the decorator style change?"
from codai.api.remote_gateway import _ORCHESTRATION_PREFIXES
unmapped = sorted(p for p in paths unmapped = sorted(p for p in paths
if not capability_for(p) and p not in _SKIP_PATHS) if not capability_for(p) and p not in _SKIP_PATHS
and not p.startswith(_ORCHESTRATION_PREFIXES))
assert unmapped == [], f"unmapped /v1 endpoints: {unmapped}" assert unmapped == [], f"unmapped /v1 endpoints: {unmapped}"
...@@ -625,3 +628,151 @@ def test_a_capability_pool_picks_up_the_default_image(monkeypatch): ...@@ -625,3 +628,151 @@ def test_a_capability_pool_picks_up_the_default_image(monkeypatch):
monkeypatch.setattr(rw, "_pools", {}) monkeypatch.setattr(rw, "_pools", {})
own = rw.get_capability_pool("video", {"image": "me/mine:1"}) own = rw.get_capability_pool("video", {"image": "me/mine:1"})
assert own.mcfg.image == "me/mine:1" assert own.mcfg.image == "me/mine:1"
# --------------------------------------------------------------------------- #
# per-model placement: two models of the SAME kind, placed differently
# --------------------------------------------------------------------------- #
@pytest.fixture()
def video_models(monkeypatch):
entries = {
"wan-local": {"path": "Wan-AI/Wan2.2-I2V-A14B", "model_type": "video_models",
"placement": "local"},
"wan-remote": {"path": "Wan-AI/Wan2.2-TI2V-5B", "model_type": "video_models",
"backend": "runpod",
"runpod": {"mode": "pods", "max_hourly_usd": 0.9}},
"wan-plain": {"path": "some/other-video", "model_type": "video_models"},
}
import codai.models.manager as mgr
import codai.api.runpod_worker as rw
monkeypatch.setattr(mgr, "_model_entry_for", lambda n: entries.get(n))
monkeypatch.setattr(mgr, "get_active_runpod_config",
lambda: type("A", (), {"enabled": True})())
monkeypatch.setattr(rw, "_pools", {})
monkeypatch.setattr(rw, "_ensure_scaler", lambda: None)
return entries
def _body(model):
return json.dumps({"model": model, "prompt": "x"}).encode()
def test_two_video_models_can_be_placed_differently(video_models, monkeypatch):
import codai.api.remote_gateway as gw
from codai.api.remote_gateway import resolve_target
import codai.api.runpod_worker as rw
path = "/v1/video/generations"
# The whole video capability is remote…
monkeypatch.setattr(gw, "capability_endpoints", lambda: {"video": "http://pod:8000"})
# …but a model pinned local stays local.
assert resolve_target(path, "POST", "", _body("wan-local"), "application/json") is None
# A model pinned to RunPod gets its OWN pod, with its own budget.
t = resolve_target(path, "POST", "", _body("wan-remote"), "application/json")
assert t.pool is not None and t.pool.mcfg.max_hourly_usd == 0.9
# And a model with nothing set follows the capability.
t2 = resolve_target(path, "POST", "", _body("wan-plain"), "application/json")
assert t2.url == "http://pod:8000" and t2.pool is None
def test_a_video_models_pod_uses_the_video_image_and_is_told_about_the_model(video_models):
from codai.api.runpod_worker import pod_plan, parse_model_runpod
entry = video_models["wan-remote"]
plan = pod_plan(parse_model_runpod(entry["runpod"]), "wan-remote", "wan-remote",
entry=entry, api_key="tok")
# vLLM cannot serve a diffusion model: the model's KIND picks the image.
assert plan["image"] == "ghcr.io/nextime/coderai-video:latest"
assert plan["health_path"] == "/healthz"
# A fresh pod has an empty catalogue and must be told what it is serving.
seeded = json.loads(plan["env"]["CODERAI_SEED_MODELS"])
assert seeded[0]["path"] == "Wan-AI/Wan2.2-TI2V-5B"
assert seeded[0]["model_type"] == "video_models"
assert plan["env"]["CODERAI_API_TOKEN"] == "tok"
def test_a_local_only_path_is_not_seeded():
"""A /AI/... path means nothing on a rented machine; don't pretend it does."""
from codai.api.runpod_worker import seed_model_env
assert seed_model_env({"path": "/AI/models/local-only", "model_type": "video_models"}) == ""
assert seed_model_env({"path": "org/repo", "model_type": "video_models"}) != ""
# --------------------------------------------------------------------------- #
# orchestration stays local
# --------------------------------------------------------------------------- #
@pytest.mark.parametrize("path", [
"/v1/pipelines/story", "/v1/pipelines/image-to-video",
"/v1/characters/generate", "/v1/environments/generate"])
def test_orchestration_endpoints_are_never_forwarded(path, monkeypatch):
"""These are a sequence of calls to other endpoints. The chain stays here and
each step is placed by its own model; forwarding the chain would place every
step by the pod's catalogue instead."""
import codai.api.remote_gateway as gw
from codai.api.remote_gateway import resolve_target, capability_for
monkeypatch.setattr(gw, "capability_endpoints",
lambda: {"pipelines": "http://pod:8000",
"characters": "http://pod:8000",
"environments": "http://pod:8000"})
monkeypatch.setattr(gw, "_model_remote", lambda m: "http://pod:8000")
assert capability_for(path) == ""
assert resolve_target(path, "POST", "", b'{"model":"x"}', "application/json") is None
# --------------------------------------------------------------------------- #
# LoRAs follow the request to the remote
# --------------------------------------------------------------------------- #
def test_loras_are_uploaded_to_the_remote_once(tmp_path, monkeypatch):
import hashlib
from codai.api import remote_gateway as gw
weights = tmp_path / "style.safetensors"
weights.write_bytes(b"fake-lora-weights" * 100)
digest = hashlib.sha256(weights.read_bytes()).hexdigest()
monkeypatch.setattr("codai.api.loras.resolve_lora_ref",
lambda spec: str(weights) if spec.get("model") else None)
calls = {"checked": 0, "uploaded": 0}
have = {"blob": False}
class _Resp:
def __init__(self, code): self.status_code = code
def raise_for_status(self): pass
def fake_get(url, **kw):
calls["checked"] += 1
return _Resp(200 if have["blob"] else 404)
def fake_post(url, **kw):
calls["uploaded"] += 1
have["blob"] = True
return _Resp(200)
monkeypatch.setattr("requests.get", fake_get)
monkeypatch.setattr("requests.post", fake_post)
body = json.dumps({"model": "sdxl", "prompt": "x",
"loras": [{"model": str(weights), "weight": 0.8}]}).encode()
out = gw.sync_loras("http://pod:8000", "tok", body)
spec = json.loads(out)["loras"][0]
# The local path is replaced by the content hash the remote now holds.
assert spec["id"] == f"sha256:{digest}" and spec["weight"] == 0.8
assert "model" not in spec
assert calls["uploaded"] == 1
# Second request: the remote already has it, so nothing is uploaded again.
gw.sync_loras("http://pod:8000", "tok", out)
assert calls["uploaded"] == 1
def test_an_unresolvable_lora_is_left_for_the_remote(monkeypatch):
"""An HF repo id is not a local file — the remote fetches it itself."""
from codai.api import remote_gateway as gw
monkeypatch.setattr("codai.api.loras.resolve_lora_ref", lambda spec: None)
body = json.dumps({"loras": [{"model": "org/some-lora", "weight": 1.0}]}).encode()
assert gw.sync_loras("http://pod:8000", "", body) is body
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment