runpod(phase2): serverless request path + per-model config + no-local-download

End-to-end serverless: a model tagged backend:runpod (mode:serverless) proxies
OpenAI chat/completions to a RunPod serverless endpoint. Runs on a remote GPU,
holds NO local weights.

- codai/api/runpod_worker.py: RunpodModelConfig + parse_model_runpod (mode/pools/
  criteria/gpu/price/scaling/cost/serverless), serverless_base_url + auth_headers.
- codai/backends/runpod.py: RunpodBackend — thin OpenAI proxy (chat/stream, Bearer
  auth for serverless), uses_local_vram=False; pods/auto raise a clear "next phase"
  error for now.
- manager.py: runpod in _ENGINE_BACKENDS + _runpod_name_claims (pin-only) +
  runpod_should_handle + load_model dispatch branch. _model_is_runpod() gate:
  set_default_model skips local download/cache for runpod-served models.
- router.py + registry.py: runpod is an explicit-pin capability; added to every
  engine's default caps (no local VRAM → any node hosts the proxy; prefers primary).
- admin routes: whitelist per-model `runpod` + `runpod_spillover` blocks; skip
  local-VRAM estimate for runpod models.
- models.html: RunPod backend option + full per-model sub-form (mode w/ explainer,
  serverless endpoint, served model, pools/criteria/gpu/price/spot, min/max pods,
  idle timeout, image/disk/port/env, spend limit+period) + prefill/collect.

Serverless verified wired in-image (ef1fd7a3802c). Pods pool/scaler, cost ledger,
spillover routing, and stats UI are later phases.
Co-Authored-By: 's avatarClaude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01R6iugoNeEgt9PkajKrFyq4
parent 7c057af2
...@@ -2673,8 +2673,11 @@ async def api_model_configure(request: Request, username: str = Depends(require_ ...@@ -2673,8 +2673,11 @@ async def api_model_configure(request: Request, username: str = Depends(require_
# Auto-estimate used_vram_gb from file size if not provided. For a not-yet- # Auto-estimate used_vram_gb from file size if not provided. For a not-yet-
# downloaded HF model this returns None (nothing on disk to measure); the # downloaded HF model this returns None (nothing on disk to measure); the
# download-completion backfill fills it in once the weights land. # download-completion backfill fills it in once the weights land.
# RunPod-served models hold NO local weights (the remote pod pulls them), so
# never estimate local VRAM for them.
_is_runpod = (str(data.get("backend") or "").strip().lower() == "runpod")
used_vram_gb = data.get("used_vram_gb") used_vram_gb = data.get("used_vram_gb")
if used_vram_gb is None: if used_vram_gb is None and not _is_runpod:
used_vram_gb = _estimate_used_vram_gb(path) used_vram_gb = _estimate_used_vram_gb(path)
# Build settings entry # Build settings entry
...@@ -2754,6 +2757,82 @@ async def api_model_configure(request: Request, username: str = Depends(require_ ...@@ -2754,6 +2757,82 @@ async def api_model_configure(request: Request, username: str = Depends(require_
else: else:
entry.pop("colibri", None) entry.pop("colibri", None)
# Per-model runpod block (mode/pools/criteria/gpu/price/scaling/cost/serverless).
# Stored as a normalized dict on the entry; the backend reads it via
# codai.api.runpod_worker.parse_model_runpod. Only meaningful for backend:runpod
# (pods|serverless|auto) or a local model's spillover target.
if "runpod" in data:
src = data.get("runpod") if isinstance(data.get("runpod"), dict) else {}
rpo = {}
# strings
for k in ("mode", "selection_criteria", "gpu_type", "image", "served_model",
"endpoint_id", "cost_period"):
v = src.get(k)
if isinstance(v, str) and v.strip():
rpo[k] = v.strip()
# cloud_types: list or comma string
ct = src.get("cloud_types")
if isinstance(ct, str):
ct = [c.strip().upper() for c in ct.split(",") if c.strip()]
elif isinstance(ct, list):
ct = [str(c).strip().upper() for c in ct if str(c).strip()]
else:
ct = None
if ct:
rpo["cloud_types"] = ct
# bools
for k in ("allow_spot",):
if k in src and src.get(k) is not None and src.get(k) != "":
sv = src.get(k)
rpo[k] = (sv.lower() in ("1", "true", "on", "yes")
if isinstance(sv, str) else bool(sv))
# ints
for k in ("container_disk_gb", "volume_gb", "port", "ctx", "min_pods",
"max_pods", "scale_up_inflight_per_pod", "idle_timeout_s",
"min_workers", "max_workers"):
v = src.get(k)
if v not in (None, ""):
try:
rpo[k] = int(v)
except (TypeError, ValueError):
pass
# floats
for k in ("min_vram_gb", "max_hourly_usd", "cost_limit_usd"):
v = src.get(k)
if v not in (None, ""):
try:
rpo[k] = float(v)
except (TypeError, ValueError):
pass
# env dict (extra pod env)
env = src.get("env")
if isinstance(env, dict) and env:
rpo["env"] = {str(k): str(v) for k, v in env.items()}
if rpo:
entry["runpod"] = rpo
else:
entry.pop("runpod", None)
# Per-model spillover-to-RunPod block for a LOCAL model (local is primary; burst
# to RunPod on the configured triggers). Kept separate from the `runpod` block so
# a model can be local-primary with a runpod overflow target.
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"):
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")
if isinstance(sv, str) else bool(sv))
# the runpod target spec reuses the same shape as the `runpod` block
tgt = src.get("target")
if isinstance(tgt, dict) and tgt:
sp["target"] = tgt
if sp.get("enabled"):
entry["runpod_spillover"] = sp
else:
entry.pop("runpod_spillover", None)
# A GGUF LLM is served by llama.cpp. Its multimodal projector (mmproj) gives # A GGUF LLM is served by llama.cpp. Its multimodal projector (mmproj) gives
# it VISION INPUT, which is the `image_to_text` capability served through # it VISION INPUT, which is the `image_to_text` capability served through
# llama.cpp — NOT the diffusers `vision_models`/`image_models` categories # llama.cpp — NOT the diffusers `vision_models`/`image_models` categories
......
This diff is collapsed.
# 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.
"""RunPod worker — per-model config parsing + endpoint resolution.
This module owns the RunPod-specific model config (parsed from the model's
models.json ``runpod`` block) and turns it into a live OpenAI base URL that
:mod:`codai.backends.runpod` proxies to.
Two modes (per-model ``mode``):
* ``serverless`` — RunPod autoscales an endpoint; coderai just proxies to
``<serverless_base>/<endpoint_id>/openai/v1`` with the account API key.
RunPod owns the worker lifecycle; there is nothing for coderai to provision.
* ``pods`` — coderai provisions, load-balances and scales a POOL of GPU pods
itself (lifecycle in this module; see RunpodPodPool). [built in a later phase]
* ``auto`` — pick pods-vs-serverless by the selection criteria. [later phase]
The account settings (API key, endpoints, global caps) live in
:class:`codai.config.RunpodConfig`; this module reads the per-model block.
"""
import os
from dataclasses import dataclass, field
from typing import Optional
# --------------------------------------------------------------------------- #
# Per-model config
# --------------------------------------------------------------------------- #
@dataclass
class RunpodModelConfig:
"""Parsed per-model ``runpod`` block. All fields optional with sane defaults."""
mode: str = "pods" # pods | serverless | auto
# --- selection (pods) ---
cloud_types: list = field(default_factory=lambda: ["SECURE"]) # allowed pools
selection_criteria: str = "cheaper" # cheaper | faster
gpu_type: str = "" # explicit RunPod gpuTypeId (optional)
min_vram_gb: float = 0.0
max_hourly_usd: float = 0.0 # per-model $/hr GPU ceiling (0 = none)
allow_spot: bool = False
# --- serving (pods) ---
image: str = "" # blank = default vLLM-OpenAI image
served_model: str = "" # HF id the pod/endpoint serves
container_disk_gb: int = 40
volume_gb: int = 0
port: int = 8000
ctx: int = 0
env: dict = field(default_factory=dict) # extra pod env (HF_TOKEN, etc.)
# --- scaling (pods) ---
min_pods: int = 0
max_pods: int = 1
scale_up_inflight_per_pod: int = 4
idle_timeout_s: int = 300 # destroy a pod this long after its last request
# --- serverless ---
endpoint_id: str = "" # reference an existing serverless endpoint
min_workers: int = 0
max_workers: int = 1
# --- cost budget (cumulative spend, distinct from the $/hr rate cap) ---
cost_limit_usd: float = 0.0 # 0 = unlimited
cost_period: str = "unlimited" # hour | day | week | month | unlimited
@property
def is_serverless(self) -> bool:
return (self.mode or "").lower() == "serverless"
# Default OpenAI-compatible pod image (vLLM's official OpenAI server). The pod
# pulls the model named by ``served_model`` (needs HF_TOKEN in env for gated repos).
DEFAULT_POD_IMAGE = "vllm/vllm-openai:latest"
def _as_bool(v, default=False):
if isinstance(v, bool):
return v
if isinstance(v, str):
return v.strip().lower() in ("1", "true", "yes", "on")
return default
def _as_float(v, default=0.0):
try:
return float(v)
except (TypeError, ValueError):
return default
def _as_int(v, default=0):
try:
return int(v)
except (TypeError, ValueError):
return default
def parse_model_runpod(block: Optional[dict]) -> RunpodModelConfig:
"""Build a RunpodModelConfig from a model entry's ``runpod`` dict (or {})."""
b = block or {}
cloud = b.get("cloud_types")
if isinstance(cloud, str):
cloud = [c.strip().upper() for c in cloud.split(",") if c.strip()]
elif isinstance(cloud, list):
cloud = [str(c).strip().upper() for c in cloud if str(c).strip()]
else:
cloud = None
env = b.get("env")
if not isinstance(env, dict):
env = {}
cfg = RunpodModelConfig()
cfg.mode = (b.get("mode") or cfg.mode).strip().lower()
if cloud:
cfg.cloud_types = cloud
cfg.selection_criteria = (b.get("selection_criteria") or cfg.selection_criteria).strip().lower()
cfg.gpu_type = (b.get("gpu_type") or "").strip()
cfg.min_vram_gb = _as_float(b.get("min_vram_gb"), cfg.min_vram_gb)
cfg.max_hourly_usd = _as_float(b.get("max_hourly_usd"), cfg.max_hourly_usd)
cfg.allow_spot = _as_bool(b.get("allow_spot"), cfg.allow_spot)
cfg.image = (b.get("image") or "").strip()
cfg.served_model = (b.get("served_model") or "").strip()
cfg.container_disk_gb = _as_int(b.get("container_disk_gb"), cfg.container_disk_gb)
cfg.volume_gb = _as_int(b.get("volume_gb"), cfg.volume_gb)
cfg.port = _as_int(b.get("port"), cfg.port) or 8000
cfg.ctx = _as_int(b.get("ctx"), cfg.ctx)
cfg.env = {str(k): str(v) for k, v in env.items()}
cfg.min_pods = max(0, _as_int(b.get("min_pods"), cfg.min_pods))
cfg.max_pods = max(1, _as_int(b.get("max_pods"), cfg.max_pods))
cfg.scale_up_inflight_per_pod = max(1, _as_int(b.get("scale_up_inflight_per_pod"),
cfg.scale_up_inflight_per_pod))
cfg.idle_timeout_s = max(0, _as_int(b.get("idle_timeout_s"), cfg.idle_timeout_s))
cfg.endpoint_id = (b.get("endpoint_id") or "").strip()
cfg.min_workers = max(0, _as_int(b.get("min_workers"), cfg.min_workers))
cfg.max_workers = max(1, _as_int(b.get("max_workers"), cfg.max_workers))
cfg.cost_limit_usd = _as_float(b.get("cost_limit_usd"), cfg.cost_limit_usd)
cfg.cost_period = (b.get("cost_period") or cfg.cost_period).strip().lower()
return cfg
def model_runpod_block(model_name: str) -> dict:
"""Fetch the ``runpod`` block from the model's models.json entry, or {}."""
try:
from codai.models.manager import _model_entry_for
entry = _model_entry_for(model_name)
if entry and isinstance(entry.get("runpod"), dict):
return entry["runpod"]
except Exception:
pass
return {}
# --------------------------------------------------------------------------- #
# Serverless resolution (RunPod owns the lifecycle — we only build the URL)
# --------------------------------------------------------------------------- #
def serverless_base_url(account_cfg, mcfg: RunpodModelConfig) -> str:
"""Return the OpenAI base URL for a serverless endpoint, e.g.
``https://api.runpod.ai/v2/<endpoint_id>/openai/v1``."""
eid = (mcfg.endpoint_id or "").strip()
if not eid:
raise RuntimeError(
"RunPod serverless: no endpoint_id set on the model's runpod config. "
"Create a serverless endpoint in RunPod (a vLLM worker) and paste its id.")
base = (getattr(account_cfg, "serverless_base", "") or "https://api.runpod.ai/v2").rstrip("/")
return f"{base}/{eid}/openai/v1"
def auth_headers(account_cfg) -> dict:
"""Authorization header for RunPod serverless (Bearer API key)."""
key = (getattr(account_cfg, "api_key", "") or "").strip()
return {"Authorization": f"Bearer {key}"} if key else {}
This diff is collapsed.
...@@ -43,12 +43,15 @@ def _short_stem(key: str) -> str: ...@@ -43,12 +43,15 @@ def _short_stem(key: str) -> str:
# whisper, but not transformers, ds4 or colibri. # whisper, but not transformers, ds4 or colibri.
# k3 (kimi-k3-in-c) is a CPU engine → any node can host it; kt (ktransformers/SGLang) # k3 (kimi-k3-in-c) is a CPU engine → any node can host it; kt (ktransformers/SGLang)
# is CPU+GPU → the GPU-capable nodes. # is CPU+GPU → the GPU-capable nodes.
# RunPod ("runpod") proxies to a remote GPU and uses NO local VRAM, so every node
# can host it — it's added to all cap sets. The front prefers the primary/default
# engine to host it (routing falls to default_engine before least_loaded).
_DEFAULT_CAPS = { _DEFAULT_CAPS = {
"nvidia": {"transformers", "gguf", "whisper", "ds4", "colibri", "k3", "kt"}, "nvidia": {"transformers", "gguf", "whisper", "ds4", "colibri", "k3", "kt", "runpod"},
"cuda": {"transformers", "gguf", "whisper", "ds4", "colibri", "k3", "kt"}, "cuda": {"transformers", "gguf", "whisper", "ds4", "colibri", "k3", "kt", "runpod"},
"vulkan": {"gguf", "whisper", "k3"}, "vulkan": {"gguf", "whisper", "k3", "runpod"},
"opencl": {"gguf", "whisper", "k3"}, "opencl": {"gguf", "whisper", "k3", "runpod"},
"auto": {"transformers", "gguf", "whisper", "ds4", "colibri", "k3", "kt"}, "auto": {"transformers", "gguf", "whisper", "ds4", "colibri", "k3", "kt", "runpod"},
} }
......
...@@ -112,7 +112,7 @@ def required_capability(model: Optional[str], path: Optional[str] = None, ...@@ -112,7 +112,7 @@ def required_capability(model: Optional[str], path: Optional[str] = None,
return "whisper" return "whisper"
# 1. Explicit engine-backend pin wins (authoritative), like the manager resolver. # 1. Explicit engine-backend pin wins (authoritative), like the manager resolver.
b = (backend or "").lower() b = (backend or "").lower()
if b in ("colibri", "ds4", "k3", "kt", "vllm"): if b in ("colibri", "ds4", "k3", "kt", "vllm", "runpod"):
return b return b
m = (model or "").lower() m = (model or "").lower()
......
...@@ -234,7 +234,7 @@ def _gguf_architecture(path: str): ...@@ -234,7 +234,7 @@ def _gguf_architecture(path: str):
# --------------------------------------------------------------------------- # # --------------------------------------------------------------------------- #
# Ordered by arbitration preference for ambiguous names. New engines append here. # Ordered by arbitration preference for ambiguous names. New engines append here.
_ENGINE_BACKENDS = ("ds4", "colibri", "k3", "kt", "vllm") _ENGINE_BACKENDS = ("ds4", "colibri", "k3", "kt", "vllm", "runpod")
def _engine_config(engine: str): def _engine_config(engine: str):
...@@ -245,6 +245,7 @@ def _engine_config(engine: str): ...@@ -245,6 +245,7 @@ def _engine_config(engine: str):
"k3": get_active_k3_config, "k3": get_active_k3_config,
"kt": get_active_ktransformers_config, "kt": get_active_ktransformers_config,
"vllm": get_active_vllm_config, "vllm": get_active_vllm_config,
"runpod": get_active_runpod_config,
}.get(engine) }.get(engine)
return getter() if getter else None return getter() if getter else None
...@@ -357,12 +358,19 @@ def _vllm_name_claims(model_name: str) -> bool: ...@@ -357,12 +358,19 @@ def _vllm_name_claims(model_name: str) -> bool:
return False return False
def _runpod_name_claims(model_name: str) -> bool:
"""RunPod can serve any model, so it must NEVER auto-claim by a name marker.
Routes ONLY via an explicit ``backend: "runpod"`` pin on the model entry."""
return False
_ENGINE_NAME_CLAIMS = { _ENGINE_NAME_CLAIMS = {
"ds4": _ds4_name_claims, "ds4": _ds4_name_claims,
"colibri": _colibri_name_claims, "colibri": _colibri_name_claims,
"k3": _k3_name_claims, "k3": _k3_name_claims,
"kt": _kt_name_claims, "kt": _kt_name_claims,
"vllm": _vllm_name_claims, "vllm": _vllm_name_claims,
"runpod": _runpod_name_claims,
} }
...@@ -427,6 +435,29 @@ def vllm_should_handle(model_name: str) -> bool: ...@@ -427,6 +435,29 @@ def vllm_should_handle(model_name: str) -> bool:
return resolve_engine_backend(model_name) == "vllm" return resolve_engine_backend(model_name) == "vllm"
def runpod_should_handle(model_name: str) -> bool:
"""True when runpod is the resolved engine backend for ``model_name``."""
return resolve_engine_backend(model_name) == "runpod"
def _model_is_runpod(model_name: str, config=None) -> bool:
"""True for a RunPod-served model — from the runtime config's backend pin (or its
embedded _raw_cfg), else the resolver. Such models hold NO local weights, so the
manager must never download/cache them or estimate local VRAM for them."""
if isinstance(config, dict):
b = (config.get("backend") or "").strip().lower()
if not b:
raw = config.get("_raw_cfg")
if isinstance(raw, dict):
b = (raw.get("backend") or "").strip().lower()
if b == "runpod":
return True
try:
return resolve_engine_backend(model_name) == "runpod"
except Exception:
return False
def _trim_cpu_ram() -> None: def _trim_cpu_ram() -> None:
"""Return freed CPU heap memory to the OS (and let the kernel reclaim swap). """Return freed CPU heap memory to the OS (and let the kernel reclaim swap).
...@@ -610,6 +641,18 @@ class ModelManager: ...@@ -610,6 +641,18 @@ class ModelManager:
self.tool_parser = ModelParserAdapter(model_name=model_name) self.tool_parser = ModelParserAdapter(model_name=model_name)
return return
# RunPod: proxy matching models to a remote GPU on RunPod (serverless
# endpoint or a managed pod pool). Uses NO local VRAM and downloads
# nothing — the remote worker holds the weights. Pin-selected only.
if runpod_should_handle(model_name):
from codai.backends.runpod import RunpodBackend
print(f"Routing '{model_name}' to RunPod (remote GPU) backend")
self.backend_type = "runpod"
self.backend = RunpodBackend(get_active_runpod_config())
self.backend.load_model(model_name, **kwargs)
self.tool_parser = ModelParserAdapter(model_name=model_name)
return
available = detect_available_backends() available = detect_available_backends()
# Check if model is a GGUF file. The name alone isn't reliable: a gguf's # Check if model is a GGUF file. The name alone isn't reliable: a gguf's
...@@ -1475,6 +1518,13 @@ class MultiModelManager: ...@@ -1475,6 +1518,13 @@ class MultiModelManager:
self.model_backend_types[model_name] = backend_type self.model_backend_types[model_name] = backend_type
self._remember_registered_type(model_name, "text") self._remember_registered_type(model_name, "text")
# RunPod-served models hold no local weights — the remote pod/endpoint pulls
# them. Never download/cache locally; the backend resolves the remote URL on
# first request.
if _model_is_runpod(model_name, self.config.get(model_name)):
print(f"Model '{model_name}' is RunPod-served — skipping local download/cache")
return
# Download/cache the model at startup if it's a URL or HF ID # Download/cache the model at startup if it's a URL or HF ID
resolved_model = self.load_model(model_name) resolved_model = self.load_model(model_name)
if resolved_model != model_name: if resolved_model != model_name:
......
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