runpod: let a pod learn a model at runtime

A pod knew only what it was seeded with at boot, so any request naming
another model got "not available" for the rest of that pod's life. That
makes a pod a fixed catalogue rather than an extension of the local
system, which is the whole point of it.

When a remote refuses a model it does not know, the gateway now teaches
it and retries once: resolve the model to something the remote can fetch
(HuggingFace id, URL, or an upload when it exists nowhere else), POST it
to the new /v1/models/register, and replay the request. Registration is
in memory — a pod is disposable — and idempotent, so a model already
loaded is never disturbed. Taught once per pod, tracked by (base, model).

The retry is bounded to one attempt and only fires on a 400/404 whose
body actually says the model is unknown: a 500 or an out-of-memory must
never turn into a retry loop against a rented GPU.

A model that resolves to no fetchable source is refused with that reason
rather than registered as something the pod could never load.
Co-Authored-By: 's avatarClaude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0182DhR1eFNPQyGmDrpHbedY
parent ee879bef
...@@ -97,6 +97,72 @@ async def model_uploaded(name: str = Query(...), bytes: int = Query(0), ...@@ -97,6 +97,72 @@ async def model_uploaded(name: str = Query(...), bytes: int = Query(0),
return {"name": os.path.basename(target), "path": target, "bytes": size} return {"name": os.path.basename(target), "path": target, "bytes": size}
@router.post("/v1/models/register", summary="Register a model at runtime")
async def register_models(request: Request, _auth=Depends(_require_api_auth)):
"""Teach a running instance about models it did not start with.
A pod is seeded at boot with what its owner knew then. But a pod is meant to
be an EXTENSION of the local system: a request can name any model at any
time, and the pod must be able to learn it, fetch it and serve it rather than
answer "not available" for the rest of its life.
Body: one model entry, or a list of them — the same shape models.json uses.
Each needs a ``path`` the receiver can actually fetch (a HuggingFace repo id
or a URL); a local path from another machine means nothing here. Registration
is in memory only, like the boot seed: a pod is disposable.
Idempotent: a model already known is left exactly as it is, so re-registering
never disturbs a loaded model.
"""
try:
body = await request.json()
except Exception as exc:
raise HTTPException(status_code=400, detail=f"invalid JSON: {exc}")
entries = body if isinstance(body, list) else [body]
added, known, refused = [], [], []
for entry in entries:
if not isinstance(entry, dict):
continue
path = str(entry.get("path") or "").strip()
section = str(entry.get("model_type") or "text_models")
if not path:
refused.append("(no path)")
continue
if section not in _SECTIONS:
refused.append(f"{path} (unknown model_type {section!r})")
continue
if path.startswith("/") and not os.path.exists(path):
# A path from the sender's filesystem. Say so plainly rather than
# register something that will fail at load time.
refused.append(f"{path} (local path that does not exist here — send a "
"HuggingFace id, a URL, or upload the weights)")
continue
state = _register_entry(entry, section)
(added if state == "added" else known).append(path)
return {"added": added, "already_known": known, "refused": refused}
def _register_entry(entry: dict, section: str) -> str:
"""Add one entry to the in-memory catalogue. Returns 'added' or 'known'."""
try:
from codai.admin.routes import config_manager
md = getattr(config_manager, "models_data", None)
if not isinstance(md, dict):
return "known"
lst = md.setdefault(section, [])
if not isinstance(lst, list):
return "known"
path = entry.get("path")
if any(isinstance(m, dict) and m.get("path") == path for m in lst):
return "known"
lst.append(dict(entry))
print(f"[register] {path} added to {section}", flush=True)
return "added"
except Exception as exc:
print(f"[register] failed: {exc}", flush=True)
return "known"
@router.post("/v1/models/upload", summary="Upload model weights") @router.post("/v1/models/upload", summary="Upload model weights")
async def upload_model(request: Request, name: str = Query(...), async def upload_model(request: Request, name: str = Query(...),
model_type: str = Query("text_models"), model_type: str = Query("text_models"),
......
...@@ -104,7 +104,7 @@ def file_origin(filename: str) -> str: ...@@ -104,7 +104,7 @@ def file_origin(filename: str) -> str:
#: the instance it is sent to (that is the whole point of sending it), so #: the instance it is sent to (that is the whole point of sending it), so
#: forwarding one would bounce the weights straight back out again. #: forwarding one would bounce the weights straight back out again.
_SKIP_PATHS = ("/v1/chat/completions", "/v1/completions", "/v1/models", _SKIP_PATHS = ("/v1/chat/completions", "/v1/completions", "/v1/models",
"/v1/models/upload", "/v1/models/uploaded", "/v1/models/upload", "/v1/models/uploaded", "/v1/models/register",
# The test run dispatches its own probe; forwarding the test # The test run dispatches its own probe; forwarding the test
# itself would test the remote's routing, not ours. # itself would test the remote's routing, not ours.
"/v1/models/test") "/v1/models/test")
...@@ -478,7 +478,7 @@ class RemoteGatewayMiddleware: ...@@ -478,7 +478,7 @@ class RemoteGatewayMiddleware:
release() release()
async def _send_upstream(self, scope, headers, body, base, reason, send, async def _send_upstream(self, scope, headers, body, base, reason, send,
api_key: str = ""): api_key: str = "", retried: bool = False):
import asyncio import asyncio
path = scope.get("path") path = scope.get("path")
qs = (scope.get("query_string") or b"").decode("latin-1") qs = (scope.get("query_string") or b"").decode("latin-1")
...@@ -517,6 +517,18 @@ class RemoteGatewayMiddleware: ...@@ -517,6 +517,18 @@ class RemoteGatewayMiddleware:
if "json" in ctype.lower(): if "json" in ctype.lower():
raw = await asyncio.to_thread(lambda: resp.content) raw = await asyncio.to_thread(lambda: resp.content)
resp.close() resp.close()
# "I do not know that model" is recoverable: teach the remote and try
# once more. This is what makes a pod an extension of this system
# rather than a fixed catalogue frozen at boot — any request may name
# a model the pod has never heard of.
if _is_unknown_model(resp.status_code, raw) and not retried:
model = _model_from_body(body, headers.get("content-type", ""))
if await asyncio.to_thread(teach_model, base, api_key, model):
return await self._send_upstream(
scope, headers, body, base, reason, send,
api_key=api_key, retried=True)
raw = _rewrite_file_urls(raw, base, _local_base(scope, headers)) raw = _rewrite_file_urls(raw, base, _local_base(scope, headers))
out = [h for h in out if h[0] != b"content-length"] out = [h for h in out if h[0] != b"content-length"]
out.append((b"content-length", str(len(raw)).encode())) out.append((b"content-length", str(len(raw)).encode()))
...@@ -724,6 +736,70 @@ def ensure_text_loras(base: str, api_key: str, entry: dict) -> dict: ...@@ -724,6 +736,70 @@ def ensure_text_loras(base: str, api_key: str, entry: dict) -> dict:
return out return out
#: (base_url, model) already taught to a remote. A pod keeps what it learns, and
#: a new pod has a new URL, so this never goes stale in a harmful way.
_REGISTERED: set = set()
def teach_model(base: str, api_key: str, model: str) -> bool:
"""Tell a remote about a model it does not know, so it can fetch and serve it.
A pod is seeded at boot with what we knew then. Making it an EXTENSION of the
local system means any request can name any model afterwards — so when a
remote says "not available", we send it that model's entry (resolved to
something it can fetch) and let it try again. Uploading is used only when the
model exists nowhere else.
Returns True when something was registered, i.e. a retry is worth making.
"""
import requests
if not model or (base, model) in _REGISTERED:
return False
try:
from codai.models.manager import _model_entry_for
entry = _model_entry_for(model) or {}
except Exception:
entry = {}
if not entry:
return False
from codai.api.runpod_worker import (parse_model_runpod, resolve_model_source,
seed_model_env)
block = entry.get("runpod") if isinstance(entry.get("runpod"), dict) else {}
mcfg = parse_model_runpod(block)
kind, value = resolve_model_source(entry, mcfg)
if kind == "upload":
ensure_model_uploaded(base, api_key, entry)
_REGISTERED.add((base, model))
return True
if kind not in ("hf", "url") or not value:
print(f"[remote-gateway] cannot teach {model!r} to {base}: no HuggingFace id "
"or URL it could fetch — set hf_repo/model_url, or source: upload",
flush=True)
return False
payload = seed_model_env(entry, "", value)
if not payload:
return False
headers = {"Content-Type": "application/json"}
if api_key:
headers["Authorization"] = f"Bearer {api_key}"
try:
r = requests.post(f"{base}/v1/models/register", data=payload,
headers=headers, timeout=60)
if not r.ok:
print(f"[remote-gateway] {base} refused {model!r}: "
f"{r.status_code} {(r.text or '')[:160]}", flush=True)
return False
print(f"[remote-gateway] taught {base} about {model!r} ({value})", flush=True)
_REGISTERED.add((base, model))
return True
except Exception as exc:
print(f"[remote-gateway] could not teach {model!r} to {base}: {exc}", flush=True)
return False
def ensure_model_uploaded(base: str, api_key: str, entry: dict) -> None: def ensure_model_uploaded(base: str, api_key: str, entry: dict) -> None:
"""Push a model's weights to a remote that cannot fetch them itself. """Push a model's weights to a remote that cannot fetch them itself.
...@@ -785,6 +861,22 @@ def ensure_model_uploaded(base: str, api_key: str, entry: dict) -> None: ...@@ -785,6 +861,22 @@ def ensure_model_uploaded(base: str, api_key: str, entry: dict) -> None:
print(f"[remote-gateway] upload of {name} to {base} failed: {exc}", flush=True) print(f"[remote-gateway] upload of {name} to {base} failed: {exc}", flush=True)
#: What a coderai says when a model is not in its catalogue.
_UNKNOWN_MODEL = ("is not available", "not allowed", "unknown model",
"model not found")
def _is_unknown_model(status: int, raw: bytes) -> bool:
"""True when the remote refused because it does not know the model."""
if status not in (400, 404):
return False
try:
text = (raw or b"").decode("utf-8", "replace").lower()
except Exception:
return False
return any(marker in text for marker in _UNKNOWN_MODEL)
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()
......
...@@ -473,6 +473,29 @@ Build and push the text image before using it: ...@@ -473,6 +473,29 @@ Build and push the text image before using it:
PROFILES=text ./packaging/runpod/publish_capability_images.sh PROFILES=text ./packaging/runpod/publish_capability_images.sh
``` ```
### A pod is an extension of this system, not a fixed catalogue
A pod starts knowing only what it was seeded with: the one model a per-model pool
rented it for, or the models of its capability. That is not enough — a request
can name any model at any time, and a pod that answers "not available" for the
rest of its life is a dead end rather than an extension of the local system.
So when a remote refuses a model it does not know, coderai **teaches it and
retries once**:
1. the remote answers `Model 'x' is not available`;
2. the gateway resolves that model to something the remote can fetch — a
HuggingFace repo id, a URL, or an upload when it exists nowhere else — and
posts it to `/v1/models/register`;
3. the request is replayed, and the remote downloads and serves the model.
Registration is in memory (a pod is disposable) and idempotent, so a model
already loaded is never disturbed. Each model is taught to each pod once.
A model that resolves to none of those — a local path with no `hf_repo`,
`model_url` or `source: upload` — is refused with that reason rather than
registered as something the pod could never load.
### Keeping a pod warm ### Keeping a pod warm
`keep_warm` on a model's runpod block (or a capability's) keeps one pod running `keep_warm` on a model's runpod block (or a capability's) keeps one pod running
......
...@@ -1284,3 +1284,88 @@ def test_a_capability_pod_is_told_what_it_can_serve(monkeypatch): ...@@ -1284,3 +1284,88 @@ def test_a_capability_pod_is_told_what_it_can_serve(monkeypatch):
assert [r["path"] for r in registered] == ["BAAI/bge-m3"] assert [r["path"] for r in registered] == ["BAAI/bge-m3"]
# The token must survive the same path — both travel in the pod's env. # The token must survive the same path — both travel in the pod's env.
assert plan["env"]["CODERAI_API_TOKEN"] == "tok" assert plan["env"]["CODERAI_API_TOKEN"] == "tok"
# --------------------------------------------------------------------------- #
# a pod as an extension of this system: it learns models at runtime
# --------------------------------------------------------------------------- #
class _LearningPod(BaseHTTPRequestHandler):
"""A pod with an empty catalogue that can be taught."""
protocol_version = "HTTP/1.1"
known: set = set()
seen: list = []
def log_message(self, *a):
pass
def _send(self, code, obj):
raw = json.dumps(obj).encode()
self.send_response(code)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(raw)))
self.end_headers()
self.wfile.write(raw)
def do_POST(self):
n = int(self.headers.get("Content-Length") or 0)
body = json.loads(self.rfile.read(n) or b"{}")
type(self).seen.append(self.path)
if self.path == "/v1/models/register":
for e in (body if isinstance(body, list) else [body]):
type(self).known.add(e["path"])
return self._send(200, {"added": sorted(type(self).known)})
if body.get("model") not in type(self).known:
return self._send(404, {"detail": f"Model '{body.get('model')}' is not "
"available. Use one of: "})
return self._send(200, {"data": [{"embedding": [0.1] * 8}]})
@pytest.fixture()
def learning_pod(monkeypatch):
_LearningPod.known = set()
_LearningPod.seen = []
url, shutdown = _serve(_LearningPod)
import codai.models.manager as mgr
from codai.api import remote_gateway as gw
monkeypatch.setattr(mgr, "_model_entry_for", lambda n: (
{"path": "BAAI/bge-m3", "model_type": "embedding_models"}
if n == "BAAI/bge-m3" else None))
monkeypatch.setattr(gw, "_REGISTERED", set())
yield url, gw
shutdown()
def test_an_unknown_model_answer_is_recognised():
from codai.api.remote_gateway import _is_unknown_model
assert _is_unknown_model(404, b'{"detail": "Model \'x\' is not available. Use one of: "}')
assert _is_unknown_model(400, b'{"error": "unknown model"}')
# A real failure must not be mistaken for one, or we would retry forever.
assert not _is_unknown_model(500, b'{"detail": "out of memory"}')
assert not _is_unknown_model(200, b'{"data": []}')
def test_a_pod_is_taught_a_model_it_does_not_know(learning_pod):
url, gw = learning_pod
assert _LearningPod.known == set() # empty catalogue at boot
assert gw.teach_model(url, "", "BAAI/bge-m3") is True
assert "BAAI/bge-m3" in _LearningPod.known
# Teaching the same pod twice is wasted work: it keeps what it learns.
_LearningPod.seen.clear()
assert gw.teach_model(url, "", "BAAI/bge-m3") is False
assert _LearningPod.seen == []
def test_a_model_the_pod_could_never_fetch_is_refused_with_a_reason(learning_pod,
monkeypatch):
url, gw = learning_pod
import codai.models.manager as mgr
monkeypatch.setattr(mgr, "_model_entry_for",
lambda n: {"path": "/AI/local/only",
"model_type": "embedding_models"})
# A local path means nothing on a pod: say so rather than register something
# that would fail later at load time.
assert gw.teach_model(url, "", "local-only") is False
assert _LearningPod.known == set()
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