cluster: nodes as engines, one GGUF over llama.cpp RPC, vLLM/SGLang...

cluster: nodes as engines, one GGUF over llama.cpp RPC, vLLM/SGLang multi-node, pools, compat-filtered model form

Several coderai installs as one:

- Cluster nodes (cluster.nodes): another install is an engine of this front —
  polled on its /cluster/state, assigned models, pinned by name, routed by
  capability and load; entries pushed to /cluster/reload-config and localized
  there; Load/Unload reach it through token-guarded twins. Live add/remove.
- llama.cpp RPC: codai/backends/ggml_rpc.py registers rpc-server endpoints
  through ctypes and gives every load an explicit device list (local cards,
  then the model's rpc_servers); the split and the VRAM pool see remote
  memory. cluster.rpc_servers runs this machine's rpc-server processes;
  packaging/build-rpc-server.sh builds it from the vendored llama.cpp;
  build.sh and the images build llama-cpp-python with GGML_RPC.
- vLLM on Ray (pipeline parallel across nodes) and SGLang --nnodes: per-model
  vllm/kt blocks over the global config, node start/stop commands, ray head
  and worker mode in boot.sh.
- Pools: several URLs per capability remote and several machines per host
  model, with failover.
- Models page: engine/backend selects narrowed live to compatible pairs,
  impossible pairs refused; Settings → Cluster; /admin/cluster page.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FKm7Z56iKjHgNkHc6VhWKz

README.md

@@ -125,6 +125,28 @@ Full guide: [`docs/runpod.md`](docs/runpod.md) ·
 remote placement of every model type, the capability images and the `host` backend:
 [`docs/remote-execution.md`](docs/remote-execution.md).
 
+### Several machines as one (cluster)
+
+- **Cluster nodes**: other coderai installs on your network become engines of this
+  one — polled, assigned models, pinned by name from the model page (*Engine / card*),
+  routed by capability and load; the node's own front picks the card. A node needs only
+  an API token. Load/Unload from the head reach the node; the Cluster page shows every
+  node's engines, VRAM and reachability
+- **One GGUF over several machines**: llama.cpp RPC — this machine runs `rpc-server`
+  processes (Settings → Cluster) that lend its cards; a model names them under *Cards on
+  other machines* and its layer split spreads over local and remote cards alike. Built
+  from the same llama.cpp as the bundled llama-cpp-python (`packaging/build-rpc-server.sh`)
+- **One HF model over several machines**: vLLM on Ray (tensor parallel inside a node,
+  pipeline parallel across nodes) and SGLang `--nnodes` — coderai starts the peers by
+  their configured commands, waits for their GPUs, launches, tears down; per model or
+  as defaults
+- **Pools**: several URLs for one capability, several machines for one `host` model —
+  healthy + least busy wins, a dead one is skipped and the request moves on
+- **No impossible pairs**: the model form greys out engines that cannot run the chosen
+  backend and backends the pinned engine cannot run, and refuses to save a wrong pair
+
+Full guide: [`docs/cluster.md`](docs/cluster.md).
+
 ### Image Generation
 - **Text-to-Image**: Stable Diffusion, SDXL, Flux, and GGUF image models (via stable-diffusion.cpp)
 - **Image-to-Image**: Style transfer and image editing
@@ -533,6 +555,7 @@ The [`docs/`](docs/) directory carries the deep dives — one per engine and sub
 | [`frontend-engine-split.md`](docs/frontend-engine-split.md) | Front proxy, engine subprocesses, routing |
 | [`runpod.md`](docs/runpod.md) | Renting remote GPUs: pods, serverless, budgets, the reaper |
 | [`remote-execution.md`](docs/remote-execution.md) | Running any model type elsewhere: the capability gateway, per-model placement, weights and adapters on a pod, test runs |
+| [`cluster.md`](docs/cluster.md) | Several machines as one: cluster nodes as engines, one GGUF over llama.cpp RPC, vLLM/SGLang multi-node, pools of hosts and remotes |
 | [`vllm.md`](docs/vllm.md) | vLLM as a first-class engine node |
 | [`deepseek-ds4.md`](docs/deepseek-ds4.md) · [`glm-colibri.md`](docs/glm-colibri.md) · [`kimi-k3.md`](docs/kimi-k3.md) · [`ktransformers.md`](docs/ktransformers.md) | The native MoE engines |
 | [`ocr.md`](docs/ocr.md) | The OCR subsystem and schema store |
@@ -646,6 +669,7 @@ Other top-level blocks, each with its own tab on the Settings page:
 | `server.engine_specs` / `engines` / `engine_gpus` | The front/engine split — how many engines, which GPU each owns, and their capabilities |
 | `ds4`, `colibri`, `k3`, `ktransformers`, `vllm` | Per-engine enablement, install dirs, model ids and build settings |
 | `runpod` | RunPod account settings and global cost caps — see [`docs/runpod.md`](docs/runpod.md) |
+| `cluster` | Other installs used as engines (`nodes`), the `rpc_servers` this machine lends, `advertise_host` — see [`docs/cluster.md`](docs/cluster.md) |
 | `ocr` | OCR engines, DPI, concurrency, detection mode, structured-extraction model and venv paths |
 | `broker` | AISBF broker client (below) |
 
@@ -969,6 +993,13 @@ To force Vulkan to use only the AMD GPU:
 { "vulkan": { "device_id": 1, "single_gpu": true } }
 ```
 
+### Multi-machine (cluster nodes, llama.cpp RPC, vLLM on Ray)
+
+See [`docs/cluster.md`](docs/cluster.md). In short: a node is another coderai with an
+API token; an RPC server is `cluster.rpc_servers` on the lending machine plus
+`rpc_servers` on the model; vLLM spans machines with `pipeline_parallel_size` and
+`nodes` on the model's vLLM block.
+
 ### Low VRAM
 
 ```json

build.sh

@@ -170,6 +170,21 @@ echo ""
 echo -e "${BLUE}Installing dependencies for $BACKEND backend...${NC}"
 echo ""
 
+# rpc-server: the process a machine runs to lend its cards to a GGUF loaded
+# elsewhere. Built from the very llama.cpp the wheel above vendors — the RPC
+# protocol is versioned and the two must match. Best effort: a missing cmake
+# or toolkit must not fail the install; the cluster page says when it is absent.
+build_rpc_server() {
+    if command -v cmake >/dev/null 2>&1; then
+        echo -e "${YELLOW}Building llama.cpp rpc-server (cluster: cards on other machines)...${NC}"
+        PYTHON="$(command -v python)" bash "$(dirname "$0")/packaging/build-rpc-server.sh" \
+            || echo -e "${YELLOW}Warning: rpc-server build failed — run packaging/build-rpc-server.sh later${NC}"
+    else
+        echo -e "${YELLOW}cmake not found: skipping rpc-server (packaging/build-rpc-server.sh)${NC}"
+    fi
+}
+
+
 if [ "$BACKEND" = "nvidia" ]; then
     # NVIDIA/CUDA backend
     echo -e "${YELLOW}Installing PyTorch with CUDA support...${NC}"
@@ -274,7 +289,7 @@ elif [ "$BACKEND" = "vulkan" ]; then
     
     # Build with Vulkan support (add CUDA too if available)
     echo -e "${YELLOW}Building llama-cpp-python with Vulkan support...${NC}"
-    _LLAMA_CMAKE="-DGGML_VULKAN=ON"
+    _LLAMA_CMAKE="-DGGML_VULKAN=ON -DGGML_RPC=ON"
     if command -v nvcc &> /dev/null || [ -d "/usr/local/cuda" ]; then
         _LLAMA_CMAKE="$_LLAMA_CMAKE -DGGML_CUDA=ON"
         echo -e "${GREEN}  ✓ Also enabling CUDA support (NVIDIA detected)${NC}"
@@ -284,6 +299,7 @@ elif [ "$BACKEND" = "vulkan" ]; then
         exit 1
     }
     
+    build_rpc_server
     echo -e "${YELLOW}Installing Vulkan-specific requirements...${NC}"
     pip install -r requirements-vulkan.txt
     echo -e "${YELLOW}Building whispercpp with Vulkan support for GPU-accelerated transcription...${NC}"
@@ -388,7 +404,7 @@ elif [ "$BACKEND" = "vulkan-nvidia" ]; then
     # Note: llama.cpp doesn't have a compile-time option to disable specific GPUs
     # The device selection happens at runtime via environment variables
     echo -e "${YELLOW}Building llama-cpp-python with Vulkan support...${NC}"
-    _LLAMA_CMAKE="-DGGML_VULKAN=ON"
+    _LLAMA_CMAKE="-DGGML_VULKAN=ON -DGGML_RPC=ON"
     if command -v nvcc &> /dev/null || [ -d "/usr/local/cuda" ]; then
         _LLAMA_CMAKE="$_LLAMA_CMAKE -DGGML_CUDA=ON"
         echo -e "${GREEN}  ✓ Also enabling CUDA support (NVIDIA detected)${NC}"
@@ -398,6 +414,7 @@ elif [ "$BACKEND" = "vulkan-nvidia" ]; then
         exit 1
     }
     
+    build_rpc_server
     echo -e "${YELLOW}Installing Vulkan-specific requirements...${NC}"
     pip install -r requirements-vulkan.txt
     
@@ -452,7 +469,7 @@ elif [ "$BACKEND" = "cuda" ]; then
     # Build llama-cpp-python with CUDA support (add Vulkan too if available)
     echo -e "${YELLOW}Building llama-cpp-python with CUDA support...${NC}"
     echo -e "${YELLOW}This may take several minutes...${NC}"
-    _LLAMA_CMAKE="-DGGML_CUDA=ON"
+    _LLAMA_CMAKE="-DGGML_CUDA=ON -DGGML_RPC=ON"
     if pkg-config --exists vulkan 2>/dev/null; then
         _LLAMA_CMAKE="$_LLAMA_CMAKE -DGGML_VULKAN=ON"
         echo -e "${GREEN}  ✓ Also enabling Vulkan support (Vulkan detected)${NC}"
@@ -467,6 +484,7 @@ elif [ "$BACKEND" = "cuda" ]; then
         exit 1
     }
     
+    build_rpc_server
     echo -e "${YELLOW}Installing Vulkan-specific requirements...${NC}"
     pip install -r requirements-vulkan.txt
     
@@ -645,6 +663,10 @@ elif [ "$BACKEND" = "all" ]; then
         fi
         echo -e "${GREEN}  ✓ Enabling Vulkan support${NC}"
     fi
+    # The RPC backend: cards on other machines as ggml devices (docs/cluster.md).
+    if [ -n "$CMAKE_ARGS" ]; then
+        CMAKE_ARGS="$CMAKE_ARGS -DGGML_RPC=ON"
+    fi
     
     if [ -n "$CMAKE_ARGS" ]; then
         echo -e "${YELLOW}  Building with: $CMAKE_ARGS${NC}"
@@ -658,6 +680,7 @@ elif [ "$BACKEND" = "all" ]; then
     fi
     
     # Install Vulkan-specific requirements
+    build_rpc_server
     echo -e "${YELLOW}Installing Vulkan-specific requirements...${NC}"
     pip install -r requirements-vulkan.txt || {
         echo -e "${YELLOW}Warning: Some Vulkan packages failed to install${NC}"

codai/__init__.py

@@ -16,7 +16,7 @@
 
 # Canonical product version for CoderAI — single source of truth. Both the API
 # metadata and the admin web UI read from here.
-__version__ = "0.2.18"
+__version__ = "0.2.19"
 
 # Configure the CUDA caching allocator BEFORE torch is imported anywhere.
 # expandable_segments lets the allocator return freed pages to the driver even

codai/admin/templates/base.html

@@ -79,6 +79,7 @@
         <a href="{{ root_path }}/admin/archive" class="nav-link {% if '/archive' in request.url.path %}active{% endif %}">Archive</a>
         <a href="{{ root_path }}/admin/settings" class="nav-link {% if '/settings' in request.url.path %}active{% endif %}">Settings</a>
         <a href="{{ root_path }}/admin/runpod" class="nav-link {% if '/runpod' in request.url.path %}active{% endif %}">RunPod</a>
+        <a href="{{ root_path }}/admin/cluster" class="nav-link {% if '/cluster' in request.url.path %}active{% endif %}">Cluster</a>
         {% endif %}
         <button class="nav-link nav-donate-btn" onclick="document.getElementById('donateModal').classList.add('show')">&#9829; Donate</button>
       </div>

codai/api/app.py

@@ -655,7 +655,7 @@ async def internal_reload_config(request: Request):
             fresh.load()
             for _f in ("remotes", "runpod", "ds4", "colibri", "k3", "ktransformers",
                        "vllm", "server", "models", "offload", "image", "jobs",
-                       "enhance", "ocr", "thermal", "broker"):
+                       "enhance", "ocr", "thermal", "broker", "cluster"):
                 if hasattr(fresh.config, _f):
                     setattr(_cm.config, _f, getattr(fresh.config, _f))
     except Exception as exc:

codai/api/kt_worker.py

@@ -142,12 +142,36 @@ def _launch_cmd(cfg, host: str, port: int, model_path: str) -> list:
     ctx = int(getattr(cfg, "ctx", 0) or 0)
     if ctx > 0:
         cmd += ["--context-length", str(ctx)]
+    tp = int(getattr(cfg, "tp_size", 0) or 0)
+    if tp > 0:
+        cmd += ["--tp-size", str(tp)]
+    # Multi-node: this process is rank 0; the other ranks were started by
+    # their configured commands (codai/cluster/multinode.SglangNodes).
+    nn = int(getattr(cfg, "nnodes", 1) or 1)
+    if nn > 1:
+        cmd += ["--nnodes", str(nn), "--node-rank", "0",
+                "--dist-init-addr", dist_init_addr(cfg)]
     extra = (getattr(cfg, "extra_args", "") or "").strip()
     if extra:
         cmd += shlex.split(extra)
     return cmd
 
 
+def dist_init_addr(cfg) -> str:
+    """host:port of rank 0 as the other ranks reach it."""
+    v = (getattr(cfg, "dist_init_addr", "") or "").strip()
+    if v:
+        return v
+    from codai.cluster.rpc import advertise_host
+    adv = ""
+    try:
+        from codai.admin.routes import config_manager
+        adv = getattr(getattr(config_manager.config, "cluster", None), "advertise_host", "") or ""
+    except Exception:
+        pass
+    return f"{advertise_host(adv)}:20000"
+
+
 def ensure_service(cfg, model_path: Optional[str] = None,
                    ready_timeout: float = 3600.0) -> str:
     """Launch (or reuse) the SGLang server for a ktransformers model; return its base URL."""
@@ -203,12 +227,27 @@ def ensure_service(cfg, model_path: Optional[str] = None,
                         applied_env[k.strip()] = v
         env_note = ("  (" + " ".join(f"{k}={v}" for k, v in applied_env.items()) + ")"
                     if applied_env else "")
+        # Multi-node: ranks 1..n-1 come up first (they block on rank 0's
+        # rendezvous address, which is what this launch is about to bind).
+        peers = None
+        nn = int(getattr(cfg, "nnodes", 1) or 1)
+        if nn > 1:
+            from codai.cluster.multinode import SglangNodes
+            peers = SglangNodes(getattr(cfg, "nodes", None), nn, dist_init_addr(cfg))
+            peers.start(model=resolved, tp_size=int(getattr(cfg, "tp_size", 0) or 0),
+                        ctx=int(getattr(cfg, "ctx", 0) or 0))
         print(f"[kt] launching SGLang: {' '.join(cmd)}{env_note}", flush=True)
         tail = collections.deque(maxlen=60)
-        proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
-                                text=True, bufsize=1, env=env)
+        try:
+            proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
+                                    text=True, bufsize=1, env=env)
+        except Exception:
+            if peers is not None:
+                peers.stop(model=resolved)
+            raise
         threading.Thread(target=_pump_logs, args=(proc, tail), daemon=True).start()
-        _services[svc_key] = {"proc": proc, "port": port, "url": url}
+        _services[svc_key] = {"proc": proc, "port": port, "url": url, "peers": peers,
+                              "model": resolved}
 
     def _tail_msg():
         joined = " | ".join(l.strip()[:300] for l in list(tail) if l.strip()).strip()
@@ -245,6 +284,12 @@ def stop_service(model_id: str) -> None:
             proc.kill()
         except Exception:
             pass
+    peers = svc.get("peers")
+    if peers is not None:
+        try:
+            peers.stop(model=svc.get("model", ""))
+        except Exception as exc:
+            print(f"[kt] peer teardown failed: {exc}", flush=True)
     print(f"[kt] service for {model_id} stopped", flush=True)
 
 

codai/api/model_test.py

@@ -466,8 +466,9 @@ def _no_remote_reason(model: str, entry: dict, capability: str) -> str:
         return ""
     if isinstance(entry.get("runpod"), dict) and entry["runpod"]:
         return ""
-    if str(entry.get("backend") or "").lower() == "host" and \
-            str((entry.get("host") or {}).get("url") or "").strip():
+    if str(entry.get("backend") or "").lower() == "host" and (
+            str((entry.get("host") or {}).get("url") or "").strip()
+            or (entry.get("host") or {}).get("hosts")):
         return ""
     try:
         from codai.api.remote_gateway import capability_endpoints, capability_pods
@@ -597,7 +598,12 @@ def _placement_summary(model: str, entry: dict, force: str) -> dict:
     if kind == "host":
         _entry, hblock = detail
         how = "started on demand" if str(hblock.get("start_cmd") or "").strip() else "always on"
-        return {"where": "remote", "target": f"host {hblock.get('url')} ({how})"}
+        try:
+            from codai.api.host_worker import parse_hosts
+            urls = [h.url for h in parse_hosts(hblock)]
+        except Exception:
+            urls = [hblock.get("url")]
+        return {"where": "remote", "target": f"host {', '.join(str(u) for u in urls)} ({how})"}
     backend = str((entry or {}).get("backend") or "").lower()
     if backend == "runpod":
         return {"where": "runpod", "target": "RunPod backend"}

codai/api/remote_gateway.py

@@ -41,6 +41,7 @@ from codai.api import pod_http
 import json
 import os
 import time
+import threading
 import re
 from typing import Optional, Tuple
 
@@ -150,25 +151,132 @@ def _remotes_config():
         return None
 
 
-def capability_endpoints() -> dict:
-    """The configured capability -> URL map (config first, then env)."""
+def _as_url_list(v) -> list:
+    """An endpoint value: one URL, a comma/newline-separated string, or a list."""
+    if isinstance(v, (list, tuple)):
+        items = [str(x) for x in v]
+    else:
+        items = str(v or "").replace("\n", ",").split(",")
+    out = []
+    for x in items:
+        x = x.strip().rstrip("/")
+        if x and x not in out:
+            out.append(x)
+    return out
+
+
+def capability_endpoint_lists() -> dict:
+    """capability -> [URL, …] (config first, then env). Several URLs for one
+    capability form a pool: the first healthy, least-busy one takes each
+    request, and a URL that stops answering is skipped until it is back."""
     cfg = _remotes_config()
     out = {}
     if cfg is not None and getattr(cfg, "enabled", True):
         eps = getattr(cfg, "endpoints", None)
         if isinstance(eps, dict):
-            out.update({str(k).strip().lower(): str(v).strip()
-                        for k, v in eps.items() if str(v).strip()})
+            for k, v in eps.items():
+                urls = _as_url_list(v)
+                if urls:
+                    out[str(k).strip().lower()] = urls
     # CODERAI_REMOTE_IMAGES_URL=… etc. — handy for a container with no config edit.
     for _, cap in _CAPABILITY_PREFIXES:
         if cap in out:
             continue
         val = os.environ.get(f"CODERAI_REMOTE_{cap.upper()}_URL", "").strip()
         if val:
-            out[cap] = val
+            out[cap] = _as_url_list(val)
     return out
 
 
+def capability_endpoints() -> dict:
+    """The configured capability -> URL map (the first URL of each pool)."""
+    return {cap: urls[0] for cap, urls in capability_endpoint_lists().items() if urls}
+
+
+class EndpointPool:
+    """Several fixed URLs serving one capability.
+
+    Health is probed lazily and remembered briefly; a URL that fails a request
+    is marked down for a while and the request moves to the next one
+    (``failover``). Nothing is started or billed — these are machines that
+    are simply there.
+    """
+    _HEALTH_TTL = 15.0
+    _DOWN_FOR = 30.0
+
+    def __init__(self, capability: str, urls: list, api_key: str = ""):
+        self.capability = capability
+        self.urls = list(urls)
+        self.api_key = api_key
+        self._lock = threading.Lock()
+        self._inflight = {u: 0 for u in self.urls}
+        self._down_until = {}
+        self._healthy_until = {}
+
+    def _healthy(self, url: str) -> bool:
+        now = time.time()
+        if self._down_until.get(url, 0) > now:
+            return False
+        if self._healthy_until.get(url, 0) > now:
+            return True
+        headers = {"Authorization": f"Bearer {self.api_key}"} if self.api_key else {}
+        try:
+            r = pod_http.get(url + "/healthz", headers=headers, timeout=4)
+            ok = r.status_code == 200
+        except Exception:
+            ok = False
+        if ok:
+            self._healthy_until[url] = now + self._HEALTH_TTL
+        else:
+            self._down_until[url] = now + self._DOWN_FOR
+        return ok
+
+    def acquire(self, timeout: float = 0.0, affinity: str = ""):
+        with self._lock:
+            order = sorted(self.urls, key=lambda u: self._inflight.get(u, 0))
+        for url in order:
+            if self._healthy(url):
+                with self._lock:
+                    self._inflight[url] = self._inflight.get(url, 0) + 1
+                return url, url
+        raise RuntimeError(
+            f"none of the {len(self.urls)} remote(s) for {self.capability!r} answers "
+            f"/healthz: {', '.join(self.urls)}")
+
+    def release(self, handle) -> None:
+        with self._lock:
+            if handle in self._inflight and self._inflight[handle] > 0:
+                self._inflight[handle] -= 1
+
+    def failover(self, failed_url: str):
+        """Mark a URL down and hand back another healthy one, or None."""
+        with self._lock:
+            self._down_until[failed_url] = time.time() + self._DOWN_FOR
+            self._healthy_until.pop(failed_url, None)
+        self.release(failed_url)
+        try:
+            _handle, url = self.acquire()
+            return url
+        except Exception:
+            return None
+
+
+_ENDPOINT_POOLS: dict = {}
+_ENDPOINT_POOLS_LOCK = threading.Lock()
+
+
+def get_endpoint_pool(capability: str, urls: list) -> EndpointPool:
+    cfg = _remotes_config()
+    key = (capability, tuple(urls))
+    with _ENDPOINT_POOLS_LOCK:
+        pool = _ENDPOINT_POOLS.get(key)
+        if pool is None:
+            pool = EndpointPool(capability, urls,
+                                api_key=(getattr(cfg, "api_key", "") if cfg else "") or "")
+            _ENDPOINT_POOLS[key] = pool
+        return pool
+
+
 def capability_pods() -> dict:
     """The configured capability -> RunPod pod block map (``remotes.pods``).
 
@@ -216,7 +324,8 @@ def model_placement(model: str) -> tuple:
     # token, optionally a command to start it. Same acquire/release shape as a
     # pod, none of the renting machinery.
     hblock = entry.get("host") if isinstance(entry.get("host"), dict) else None
-    if backend == "host" and hblock is not None and str(hblock.get("url") or "").strip():
+    if backend == "host" and hblock is not None and (
+            str(hblock.get("url") or "").strip() or hblock.get("hosts")):
         return "host", (entry, hblock)
     return "", None
 
@@ -466,10 +575,14 @@ def resolve_target(path: str, method: str, query: str, body: bytes,
                 "service_url or a RunPod pod on the model, or a remote for its "
                 "capability")
         return None
-    url = capability_endpoints().get(cap, "")
+    urls = capability_endpoint_lists().get(cap, [])
+    url = urls[0] if urls else ""
     # "runpod" as the endpoint is shorthand for "use the pod block for this
     # capability" — so the common case needs no second config block.
     if url and url.strip().lower() != "runpod":
+        if len(urls) > 1:
+            return Target(pool=get_endpoint_pool(cap, urls),
+                          reason=f"capability {cap!r} ({len(urls)} remotes)")
         return Target(url=url, reason=f"capability {cap!r}")
     pods = capability_pods()
     if url or cap in pods:
@@ -595,8 +708,9 @@ class RemoteGatewayMiddleware:
             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)
+            _fail = (target.pool.failover if hasattr(target.pool, "failover") else None)
             await self._send_upstream(scope, headers, body, base, target.reason, send,
-                                      api_key=target.api_key)
+                                      api_key=target.api_key, on_unreachable=_fail)
             task_registry.finish(tid, "done")
         except Exception as exc:
             task_registry.finish(tid, "error", str(exc)[:200])
@@ -605,7 +719,8 @@ class RemoteGatewayMiddleware:
             release()
 
     async def _send_upstream(self, scope, headers, body, base, reason, send,
-                             api_key: str = "", retried: bool = False):
+                             api_key: str = "", retried: bool = False,
+                             on_unreachable=None):
         import asyncio
         path = scope.get("path")
         qs = (scope.get("query_string") or b"").decode("latin-1")
@@ -647,6 +762,16 @@ class RemoteGatewayMiddleware:
                 await asyncio.sleep(_wait)
                 resp = await asyncio.to_thread(_call)
         except Exception as exc:
+            # A pool of remotes: this one is down, move the request to the
+            # next healthy one before giving up.
+            if on_unreachable is not None:
+                nxt = await asyncio.to_thread(on_unreachable, base)
+                if nxt and nxt != base:
+                    print(f"[remote-gateway] {base} unreachable ({exc}); "
+                          f"failing over to {nxt}", flush=True)
+                    return await self._send_upstream(
+                        scope, headers, body, nxt, reason, send, api_key=api_key,
+                        retried=retried, on_unreachable=on_unreachable)
             return await _error(send, 502, f"remote endpoint {base} unreachable: {exc}")
 
         ctype = resp.headers.get("Content-Type") or "application/json"

codai/api/vllm_worker.py

@@ -130,6 +130,14 @@ def _launch_cmd(py, cfg, host: str, port: int, model_path: str,
     tp = int(getattr(cfg, "tensor_parallel_size", 0) or 0)
     if tp > 0:
         cmd += ["--tensor-parallel-size", str(tp)]
+    pp = int(getattr(cfg, "pipeline_parallel_size", 0) or 0)
+    if pp > 1:
+        cmd += ["--pipeline-parallel-size", str(pp)]
+    executor = (getattr(cfg, "distributed_executor_backend", "") or "").strip()
+    if not executor and needs_ray(cfg):
+        executor = "ray"
+    if executor:
+        cmd += ["--distributed-executor-backend", executor]
     mns = int(getattr(cfg, "max_num_seqs", 0) or 0)
     if mns > 0:
         cmd += ["--max-num-seqs", str(mns)]
@@ -195,6 +203,52 @@ def _remote_serves(url: str, name: str) -> bool:
     return any(i == name or i.rstrip("/").split("/")[-1] == tail for i in ids)
 
 
+def needs_ray(cfg) -> bool:
+    """Multi-node: other machines' GPUs join through a ray cluster."""
+    from codai.cluster.multinode import parse_nodes
+    if (getattr(cfg, "distributed_executor_backend", "") or "").strip().lower() == "ray":
+        return True
+    if int(getattr(cfg, "pipeline_parallel_size", 1) or 1) > 1:
+        return True
+    if (getattr(cfg, "ray_address", "") or "").strip():
+        return True
+    return bool(parse_nodes(getattr(cfg, "nodes", None)))
+
+
+def _local_gpu_count(env: dict) -> int:
+    vis = (env.get("CUDA_VISIBLE_DEVICES") or "").strip()
+    if vis:
+        return len([x for x in vis.split(",") if x.strip()])
+    try:
+        from codai.backends.gpu_probe import total_memory_bytes
+        return max(1, len(total_memory_bytes()))
+    except Exception:
+        return 1
+
+
+def _start_ray(cfg, py: str, env: dict):
+    """Bring the ray cluster up for this launch; returns (RayCluster, address)."""
+    from codai.cluster.multinode import RayCluster
+    from codai.cluster.rpc import advertise_host
+    adv = ""
+    try:
+        from codai.admin.routes import config_manager
+        adv = getattr(getattr(config_manager.config, "cluster", None), "advertise_host", "") or ""
+    except Exception:
+        pass
+    adv = advertise_host(adv)
+    tp = max(1, int(getattr(cfg, "tensor_parallel_size", 1) or 1))
+    pp = max(1, int(getattr(cfg, "pipeline_parallel_size", 1) or 1))
+    cluster = RayCluster(py, adv, port=int(getattr(cfg, "ray_port", 6379) or 6379),
+                         address=getattr(cfg, "ray_address", "") or "",
+                         nodes=getattr(cfg, "nodes", None),
+                         ready_timeout_s=float(getattr(cfg, "nodes_ready_timeout_s", 600) or 600))
+    address = cluster.start(gpus_needed=tp * pp, local_gpus=_local_gpu_count(env))
+    env.setdefault("VLLM_HOST_IP", adv)
+    env["RAY_ADDRESS"] = address
+    return cluster, address
+
+
 def ensure_service(cfg, model_path: Optional[str] = None,
                    served_name: Optional[str] = None,
                    ready_timeout: float = 3600.0) -> str:
@@ -265,14 +319,22 @@ def ensure_service(cfg, model_path: Optional[str] = None,
                     k, v = tok.split("=", 1)
                     if k.strip():
                         env[k.strip()] = v; applied[k.strip()] = v
+        ray = None
+        if needs_ray(cfg):
+            ray, _addr = _start_ray(cfg, py, env)
         print(f"[vllm] launching: {' '.join(cmd)}"
               + (f"  ({' '.join(f'{k}={v}' for k, v in applied.items())})" if applied else ""),
               flush=True)
         tail = collections.deque(maxlen=80)
-        proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
-                                text=True, bufsize=1, env=env)
+        try:
+            proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
+                                    text=True, bufsize=1, env=env)
+        except Exception:
+            if ray is not None:
+                ray.stop()
+            raise
         threading.Thread(target=_pump_logs, args=(proc, tail), daemon=True).start()
-        _services[svc_key] = {"proc": proc, "port": port, "url": url}
+        _services[svc_key] = {"proc": proc, "port": port, "url": url, "ray": ray}
 
     def _tail_msg():
         # The last six lines of a vLLM crash are the wrapper's own traceback
@@ -320,6 +382,12 @@ def stop_service(svc_key: str) -> None:
             proc.kill()
         except Exception:
             pass
+    ray = svc.get("ray")
+    if ray is not None:
+        try:
+            ray.stop()
+        except Exception as exc:
+            print(f"[vllm] ray teardown for {svc_key} failed: {exc}", flush=True)
     print(f"[vllm] service for {svc_key} stopped", flush=True)
 
 

codai/backends/ktransformers.py

@@ -70,6 +70,11 @@ class KtransformersBackend(ModelBackend):
         if _ctx > 0:
             self._ctx = _ctx
         model_path = self._resolve_model_dir(model_name)
+        # This model's own `kt` block over the global template: the machines it
+        # spans (nnodes + their commands), tp_size, KT weights, flags.
+        from codai.backends.overrides import KT_FIELDS, apply, model_block
+        self._cfg = apply(self._cfg, model_block(model_name, "kt") or
+                          model_block(model_name, "ktransformers"), KT_FIELDS, "kt")
         _resolved, self._svc_key = kt_worker.resolve_service_key(self._cfg, model_path)
         self._url = kt_worker.ensure_service(self._cfg, model_path=model_path)
         # SGLang serves under --served-model-name = the configured model_id; send that.

codai/backends/overrides.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.
+
+"""Per-model blocks laid over a global engine config.
+
+The global ``vllm``/``ktransformers`` sections are templates; a model's own
+``vllm``/``kt`` block in models.json sets what differs for THAT model —
+parallelism, the machines it spans, memory fraction, extra flags. Only the
+listed fields may be overridden: a model must never redirect the engine to a
+different install, venv or service.
+"""
+
+import dataclasses
+
+VLLM_FIELDS = ("ctx", "gpu", "gpu_memory_utilization", "tensor_parallel_size",
+               "pipeline_parallel_size", "distributed_executor_backend", "ray_address",
+               "ray_port", "nodes", "nodes_ready_timeout_s", "max_num_seqs", "dtype",
+               "quantization", "extra_args", "extra_env")
+
+KT_FIELDS = ("ctx", "kt_weight_path", "nnodes", "dist_init_addr", "tp_size", "nodes",
+             "extra_args", "extra_env")
+
+
+def apply(cfg, block: dict, fields: tuple, label: str = ""):
+    """``cfg`` with the allowed keys of ``block`` replaced; logs what changed."""
+    if not isinstance(block, dict) or not block:
+        return cfg
+    overrides = {}
+    for k in fields:
+        if k not in block:
+            continue
+        v = block[k]
+        if v in (None, ""):
+            continue
+        cur = getattr(cfg, k, None)
+        if isinstance(cur, bool):
+            v = str(v).strip().lower() in ("1", "true", "yes", "on") if isinstance(v, str) else bool(v)
+        elif isinstance(cur, int):
+            try:
+                v = int(v)
+            except (TypeError, ValueError):
+                continue
+        elif isinstance(cur, float):
+            try:
+                v = float(v)
+            except (TypeError, ValueError):
+                continue
+        overrides[k] = v
+    if not overrides:
+        return cfg
+    try:
+        out = dataclasses.replace(cfg, **overrides)
+    except Exception as exc:
+        print(f"[{label or 'engine'}] per-model overrides not applied: {exc}", flush=True)
+        return cfg
+    print(f"[{label or 'engine'}] per-model overrides: "
+          + ", ".join(f"{k}={v!r}" for k, v in overrides.items()), flush=True)
+    return out
+
+
+def model_block(model_name: str, key: str) -> dict:
+    """The named block of this model's models.json entry, or {}."""
+    try:
+        from codai.models.manager import _model_entry_for
+        entry = _model_entry_for(model_name) or {}
+    except Exception:
+        return {}
+    b = entry.get(key)
+    return b if isinstance(b, dict) else {}

codai/backends/vllm.py

@@ -71,6 +71,10 @@ class VllmBackend(ModelBackend):
         if _ctx > 0:
             self._ctx = _ctx
         model_path = self._resolve_model_dir(model_name)
+        # This model's own `vllm` block over the global template: parallelism,
+        # the machines it spans, memory fraction, flags.
+        from codai.backends.overrides import VLLM_FIELDS, apply, model_block
+        self._cfg = apply(self._cfg, model_block(model_name, "vllm"), VLLM_FIELDS, "vllm")
         # Per-model: serve each model-list entry under ITS OWN name (like nvidia/radeon),
         # not a single engine-wide model_id. Fall back to the config model_id only for the
         # single-model convenience (a model with no list entry).

codai/backends/vulkan.py

@@ -1246,6 +1246,16 @@ class VulkanBackend(ModelBackend):
                 # Otherwise we'd needlessly offload layers to CPU thinking only one
                 # card's free VRAM is available.
                 _free = _pooled_free_vram_gb(cross=bool(kwargs.get('gpu_split')))
+                # Remote cards (llama.cpp RPC) are part of the pool the layers can
+                # go to — otherwise their share would be spilled to CPU up front.
+                _rpc_pool = kwargs.get('rpc_servers', (kwargs.get('_raw_cfg') or {}).get('rpc_servers'))
+                if _rpc_pool:
+                    try:
+                        from codai.backends import ggml_rpc as _rpcm
+                        _rpcm.register_servers(_rpc_pool)
+                        _free += sum(_rpcm.free_gb(_rpc_pool))
+                    except Exception as _re:
+                        print(f"  rpc servers  : not counted in the VRAM pool ({_re})")
                 # Apply the VRAM caps to the usable pool too, so that capping a card
                 # proactively offloads the remainder to CPU here (rather than only
                 # reacting to an OOM in the load-retry loop). Per-model secondary cap
@@ -1616,6 +1626,20 @@ class VulkanBackend(ModelBackend):
         _ts = kwargs.get('tensor_split', _raw_cfg.get('tensor_split'))
         if _ts is None:
             _ts = kwargs.get('_global_tensor_split')
+        # Cards on OTHER machines, through llama.cpp's RPC backend: each
+        # ``host:port`` where an rpc-server listens becomes one more device in
+        # the split, after this machine's cards (codai/backends/ggml_rpc.py).
+        # A model that names RPC servers is split by definition.
+        from codai.backends import ggml_rpc as _rpc
+        _rpc_eps = _rpc.normalize_endpoints(
+            kwargs.get('rpc_servers', _raw_cfg.get('rpc_servers')))
+        _rpc_free = []
+        if _rpc_eps:
+            _rpc_devs = _rpc.register_servers(_rpc_eps)      # raises when unusable
+            _rpc_free = _rpc.free_gb(_rpc_eps)
+            _rpc_desc = ", ".join(ep + " (" + ", ".join(d) + ")" for ep, d in _rpc_devs.items())
+            print(f"  rpc servers  : {_rpc_desc} free {[round(f, 1) for f in _rpc_free]} GB")
+            _gpu_split = True
         # llama.cpp device order: all visible CUDA devices first, then Vulkan. Count
         # each so we can tell own-backend from foreign when both are exposed (the
         # engine was spawned with cross-GPU visibility for some split model).
@@ -1644,6 +1668,10 @@ class VulkanBackend(ModelBackend):
                 # Auto: distribute proportionally to each device's free VRAM (a 24 GB
                 # 3090 + 8 GB RX 580 → ~0.75/0.25), so the bigger card carries more.
                 _free_dev = _per_device_free_vram_gb()
+                if _rpc_free:
+                    # Remote cards come after the local ones in llama.cpp's
+                    # device order, so their free memory joins the ratio there.
+                    _free_dev = list(_free_dev) + list(_rpc_free)
                 # mmproj-aware headroom: a vision projector (CLIP/mmproj) ALWAYS loads
                 # on main_gpu (its own buffer + compute), and the model's output/KV
                 # anchor is there too. If the proportional split fills main_gpu to the
@@ -1844,7 +1872,8 @@ class VulkanBackend(ModelBackend):
             _max_attempts = 6   # initial try + 5 progressive CPU-offload retries
             for _attempt in range(_max_attempts):
                 try:
-                    self.model = Llama(**llama_kwargs)
+                    with _rpc.load_with_devices(_rpc_eps):
+                        self.model = Llama(**llama_kwargs)
                     if _attempt > 0:
                         print(f"  Loaded after CPU offload "
                               f"(n_gpu_layers={llama_kwargs.get('n_gpu_layers')})")

codai/cluster/__init__.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.
+
+"""Several coderai installs acting as one.
+
+* :mod:`codai.cluster.nodes` — other installs used as engines of this front
+  (the head polls them, assigns models to them, routes to them by capability).
+* :mod:`codai.cluster.rpc` — the llama.cpp ``rpc-server`` processes this
+  machine contributes, so one GGUF can span cards on several machines.
+* :mod:`codai.cluster.compat` — which engine can run which backend, shared by
+  the Models page (live filtering) and the save-time check.
+"""

codai/cluster/compat.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.
+
+"""Which engine can run which backend — one table, used twice.
+
+The Models page narrows its Engine / card and Backend selects to what fits
+(codai/admin/templates/models.html, ``_applyCompat``) and the save path
+refuses what does not (codai/admin/routes.validate_engine_pin). Both follow
+the router's notion of a *required capability*, so a pin that passes here is
+one the front will actually honour instead of quietly falling back.
+"""
+
+from typing import Iterable, Optional, Tuple
+
+#: Model backend (the Backend select) → capability an engine must offer.
+#: None = decided by the path (gguf → "gguf", else "transformers");
+#: "" = no engine requirement at all (placement is elsewhere).
+BACKEND_CAP = {
+    "auto": None, "nvidia": None, "cpu": None,
+    "vulkan": "gguf", "opencl": "gguf",
+    "colibri": "colibri", "ds4": "ds4", "k3": "k3", "kt": "kt", "vllm": "vllm",
+    "runpod": "runpod", "host": "",
+    "wav2vec2": "transformers", "whisper-hf": "transformers", "nemo": "transformers",
+    "vosk": "", "whisper-server": "whisper",
+}
+
+#: Backends that name a GPU vendor: the engine's own backend must match or the
+#: model silently runs on a different API than the one it asked for.
+VENDOR_BACKENDS = {
+    "nvidia": ("nvidia", "cuda", "auto"),
+    "vulkan": ("vulkan", "auto"),
+    "opencl": ("opencl", "auto"),
+}
+
+
+def required_cap(model_backend: str, path: str = "") -> Optional[str]:
+    b = (model_backend or "auto").strip().lower()
+    cap = BACKEND_CAP.get(b, None)
+    if cap is None:
+        p = (path or "").lower()
+        return "gguf" if (p.endswith(".gguf") or "gguf" in p) else "transformers"
+    return cap or None
+
+
+def check(engine_backend: str, engine_caps: Iterable[str], model_backend: str,
+          path: str = "") -> Tuple[bool, str]:
+    """(ok, reason). ``ok`` False = the pin cannot work; a non-empty reason with
+    ``ok`` True is a warning (it runs, but not the way the backend says)."""
+    b = (model_backend or "auto").strip().lower()
+    if b == "host":
+        return False, "a host model runs on the machine named in its host block, not on an engine"
+    cap = required_cap(b, path)
+    caps = {str(c).lower() for c in (engine_caps or ())}
+    eb = (engine_backend or "auto").lower()
+    if cap and cap not in caps:
+        return False, (f"needs '{cap}' but this engine ({eb}) only provides "
+                       f"{sorted(caps) or '(nothing)'}")
+    vend = VENDOR_BACKENDS.get(b)
+    if vend and eb not in vend and not eb.startswith("nvidia"):
+        return True, (f"backend '{b}' on a {eb} engine: the engine's own API "
+                      f"is used instead")
+    if vend and b in ("vulkan", "opencl") and eb.startswith("nvidia"):
+        return True, f"backend '{b}' on the NVIDIA engine runs through CUDA"
+    return True, ""

codai/cluster/nodes.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.
+
+"""Another coderai as an engine of this one.
+
+The front already knows how to drive N engines it cannot see inside: it polls
+``/internal/engine-state``, hands each one the models it owns, and proxies
+requests by capability and load. A node is that, one network hop away. The
+node's own front answers the same questions for its whole install — the
+union of its engines' loaded models, the sum of their VRAM, every capability
+any of them has — on ``/cluster/state``, guarded by an API token of the node.
+So the head needs a URL, a token, and a way to trust the node's certificate;
+this module turns a ``cluster.nodes`` entry into exactly that.
+"""
+
+import ssl
+from dataclasses import dataclass, field
+from typing import List, Optional
+
+import httpx
+
+
+@dataclass
+class NodeSpec:
+    name: str
+    url: str
+    api_key: str = ""
+    verify: str = "system"          # system | off | pem
+    ca_pem: str = ""                # the node's certificate (or CA) when verify=pem
+    capabilities: List[str] = field(default_factory=list)   # blank = as reported
+    enabled: bool = True
+    timeout_s: float = 4.0
+
+    def headers(self) -> dict:
+        return {"Authorization": f"Bearer {self.api_key}"} if self.api_key else {}
+
+    def verify_arg(self):
+        """What httpx's ``verify=`` takes for this node."""
+        if self.verify == "off":
+            return False
+        if self.verify == "pem" and self.ca_pem.strip():
+            ctx = ssl.create_default_context()
+            ctx.load_verify_locations(cadata=self.ca_pem)
+            return ctx
+        return True
+
+
+def parse_nodes(raw) -> List[NodeSpec]:
+    """``cluster.nodes`` → specs, skipping entries that cannot be used."""
+    out = []
+    seen = set()
+    for i, n in enumerate(raw or []):
+        if not isinstance(n, dict):
+            continue
+        url = str(n.get("url") or "").strip().rstrip("/")
+        if not url:
+            continue
+        name = str(n.get("name") or "").strip() or f"node{i + 1}"
+        if name.lower() in seen:
+            continue
+        seen.add(name.lower())
+        caps = n.get("capabilities")
+        if isinstance(caps, str):
+            caps = [c.strip() for c in caps.split(",") if c.strip()]
+        spec = NodeSpec(
+            name=name, url=url,
+            api_key=str(n.get("api_key") or "").strip(),
+            verify=str(n.get("verify") or "system").strip().lower() or "system",
+            ca_pem=str(n.get("ca_pem") or ""),
+            capabilities=[str(c).strip().lower() for c in (caps or []) if str(c).strip()],
+            enabled=bool(n.get("enabled", True)),
+        )
+        try:
+            spec.timeout_s = float(n.get("timeout_s") or spec.timeout_s)
+        except (TypeError, ValueError):
+            pass
+        out.append(spec)
+    return out
+
+
+def enabled_nodes(cluster_cfg) -> List[NodeSpec]:
+    if cluster_cfg is None or not getattr(cluster_cfg, "enabled", False):
+        return []
+    specs = parse_nodes(getattr(cluster_cfg, "nodes", None))
+    t = float(getattr(cluster_cfg, "poll_timeout_s", 4.0) or 4.0)
+    for s in specs:
+        s.timeout_s = t
+    return [s for s in specs if s.enabled]
+
+
+def sync_client(spec: NodeSpec) -> httpx.Client:
+    """A short-timeout client for state polls and control pushes."""
+    return httpx.Client(timeout=spec.timeout_s, headers=spec.headers(),
+                        verify=spec.verify_arg())
+
+
+def async_clients(spec: NodeSpec):
+    """(short, long) async clients the front proxies through — the same two
+    shapes it keeps for local engines, but carrying this node's token and
+    trust settings."""
+    short = httpx.AsyncClient(timeout=spec.timeout_s, headers=spec.headers(),
+                              verify=spec.verify_arg())
+    long = httpx.AsyncClient(
+        timeout=httpx.Timeout(connect=10.0, read=None, write=None, pool=None),
+        headers=spec.headers(), verify=spec.verify_arg())
+    return short, long
+
+
+# --------------------------------------------------------------- node side
+def aggregate_state(registry, rpc_endpoints: Optional[list] = None,
+                    node_name: str = "") -> dict:
+    """What this install looks like as ONE engine, from its front's registry.
+
+    The head merges this into its own registry: loaded models across every
+    engine here, VRAM summed, tasks concatenated, capabilities unioned — so a
+    node with an NVIDIA and a Radeon engine advertises both ``transformers``
+    and ``gguf`` and the head sends it either kind of model; the node's own
+    front then routes to the right card, as it always did.
+    """
+    loaded, info, tasks, caps = [], [], [], set()
+    total = free = 0.0
+    any_vram = False
+    healthy = False
+    cooling = None
+    inflight = 0
+    engines = []
+    for e in registry.all():
+        if getattr(e, "role", "engine") == "system" or getattr(e, "remote", False):
+            continue      # a node reports ITS cards, never nodes of its own
+        engines.append({"name": e.name, "backend": e.backend, "healthy": e.healthy,
+                        "vram": e.vram, "loaded": sorted(e.loaded_models)})
+        if not e.healthy:
+            continue
+        healthy = True
+        caps |= set(e.capabilities or ())
+        loaded.extend(e.loaded_models)
+        info.extend(e.loaded_info or [])
+        tasks.extend(e.tasks or [])
+        inflight += int(getattr(e, "inflight", 0) or 0)
+        if e.vram:
+            any_vram = True
+            try:
+                total += float(e.vram.get("total") or 0.0)
+                free += float(e.vram.get("free") or 0.0)
+            except (TypeError, ValueError):
+                pass
+        if e.cooling and cooling is None:
+            cooling = e.cooling
+    # A node never rents pods on the head's behalf and never nests nodes.
+    caps.discard("runpod")
+    return {
+        "node": node_name,
+        "healthy": healthy,
+        "loaded_models": sorted(set(loaded)),
+        "loaded_info": info,
+        "tasks": tasks,
+        "vram": ({"total": round(total, 2), "free": round(free, 2),
+                  "used": round(max(0.0, total - free), 2)} if any_vram else None),
+        "cooling": cooling,
+        "capabilities": sorted(caps),
+        "inflight": inflight,
+        "engines": engines,
+        "rpc_servers": list(rpc_endpoints or []),
+    }
+
+
+#: Fields that describe THIS head's placement, not the model: a node must not
+#: pin the model to an engine of the head, rent pods for it, or spread it over
+#: cards the head sees. ``rpc_servers`` stays — the node loads the weights and
+#: is the one that spreads them.
+_HEAD_ONLY = ("engine", "engine_fallback", "runpod", "host", "gpu_split",
+              "tensor_split", "split_strategy", "split_secondary_cap_gb",
+              "node_path", "node_paths", "cluster_source", "placement")
+
+
+def portable_entry(entry: dict) -> dict:
+    """A models.json entry as the head sends it: the entry itself, plus where
+    else the weights can be fetched from when the path means nothing there."""
+    e = dict(entry)
+    try:
+        from codai.api.runpod_worker import parse_model_runpod, resolve_model_source
+        block = e.get("runpod") if isinstance(e.get("runpod"), dict) else {}
+        kind, value = resolve_model_source(e, parse_model_runpod(block))
+        if kind in ("hf", "url") and value:
+            e["cluster_source"] = value
+    except Exception:
+        pass
+    return e
+
+
+def localize_entries(entries: list, node_name: str = "") -> tuple:
+    """(payload for /v1/models/register, refused) on the NODE: pick the path
+    this machine can use, drop the head's placement fields."""
+    import os
+    payload, refused = [], []
+    for entry in entries or []:
+        if not isinstance(entry, dict):
+            continue
+        path = str(entry.get("path") or "").strip()
+        paths = entry.get("node_paths") if isinstance(entry.get("node_paths"), dict) else {}
+        override = str(paths.get(node_name) or entry.get("node_path") or "").strip()
+        src = str(entry.get("cluster_source") or "").strip()
+        chosen = ""
+        for cand in (os.path.expanduser(override) if override else "", path):
+            if cand and (not cand.startswith("/") or os.path.exists(cand)):
+                chosen = cand
+                break
+            if cand and cand.startswith("/") and os.path.exists(os.path.expanduser(cand)):
+                chosen = os.path.expanduser(cand)
+                break
+        if not chosen and src:
+            chosen = src
+        if not chosen:
+            refused.append(f"{entry.get('alias') or path}: {path!r} does not exist on "
+                           f"this node and no HuggingFace id/URL is known — set "
+                           f"'Path on the node' on the head's model page")
+            continue
+        drop = set(_HEAD_ONLY)
+        if entry.get("rpc_servers"):
+            # The split describes the node's devices plus its RPC peers, in the
+            # order the node's llama.cpp sees them — it belongs to the node.
+            drop -= {"gpu_split", "tensor_split", "split_strategy", "split_secondary_cap_gb"}
+        out = {k: v for k, v in entry.items() if k not in drop}
+        out["path"] = chosen
+        payload.append(out)
+    return payload, refused

codai/cluster/overview.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.
+
+"""Everything this front can send work to, on one page.
+
+Local engines, cluster nodes, this machine's and the nodes' RPC servers, the
+hosts named by ``backend: host`` models, the capability remotes, the pods —
+each with whether it answers right now. Built in a worker thread from the
+front's registry and a few short probes; probes are cached so the page can
+refresh every few seconds without hammering anything.
+"""
+
+import json
+import os
+import threading
+import time
+
+_probe_cache = {}          # url -> (expires, ok, detail)
+_probe_lock = threading.Lock()
+_PROBE_TTL = 20.0
+
+
+def _probe(url: str, path: str = "/healthz", api_key: str = "",
+           timeout: float = 3.0) -> tuple:
+    """(ok, detail) for a base URL, cached for a while."""
+    key = (url, path, bool(api_key))
+    now = time.time()
+    with _probe_lock:
+        hit = _probe_cache.get(key)
+        if hit and hit[0] > now:
+            return hit[1], hit[2]
+    ok, detail = False, ""
+    try:
+        from codai.api import pod_http
+        headers = {"Authorization": f"Bearer {api_key}"} if api_key else {}
+        t0 = time.time()
+        r = pod_http.get(url.rstrip("/") + path, headers=headers, timeout=timeout)
+        ok = r.status_code == 200
+        detail = f"HTTP {r.status_code} in {(time.time() - t0) * 1000:.0f} ms"
+    except Exception as exc:
+        detail = str(exc)[:160]
+    with _probe_lock:
+        _probe_cache[key] = (now + _PROBE_TTL, ok, detail)
+    return ok, detail
+
+
+def _models_data(front) -> dict:
+    path = getattr(front, "_models_path", None)
+    if not path:
+        return {}
+    try:
+        with open(path) as f:
+            return json.load(f)
+    except Exception:
+        return {}
+
+
+def _host_targets(front) -> list:
+    """Every host named by a ``backend: host`` model, probed."""
+    out = []
+    data = _models_data(front)
+    for cat, lst in data.items():
+        if not isinstance(lst, list):
+            continue
+        for m in lst:
+            if not isinstance(m, dict) or str(m.get("backend") or "").lower() != "host":
+                continue
+            block = m.get("host") if isinstance(m.get("host"), dict) else {}
+            try:
+                from codai.api.host_worker import parse_hosts
+                slots = parse_hosts(block)
+            except Exception:
+                slots = []
+            name = m.get("alias") or os.path.basename(str(m.get("path") or "")) or "?"
+            for h in slots:
+                ok, detail = _probe(h.url, h.health_path, h.api_key)
+                out.append({"model": name, "url": h.url, "on_demand": bool(h.start_cmd),
+                            "ok": ok, "detail": detail})
+    return out
+
+
+def _remote_targets(front) -> list:
+    out = []
+    try:
+        from codai.api.remote_gateway import capability_endpoint_lists
+        eps = capability_endpoint_lists()
+    except Exception:
+        eps = {}
+    rem = getattr(front.config, "remotes", None)
+    key = (getattr(rem, "api_key", "") or "") if rem else ""
+    for cap, urls in sorted(eps.items()):
+        for u in urls:
+            if u.lower() == "runpod":
+                out.append({"capability": cap, "url": "runpod", "ok": None,
+                            "detail": "served by a pod pool"})
+                continue
+            ok, detail = _probe(u, "/healthz", key)
+            out.append({"capability": cap, "url": u, "ok": ok, "detail": detail})
+    return out
+
+
+def _pods() -> list:
+    try:
+        from codai.api.runpod_worker import pods_status
+        return pods_status()
+    except Exception:
+        return []
+
+
+def _rpc_reachable(endpoint: str) -> bool:
+    import socket
+    host, _, port = endpoint.rpartition(":")
+    try:
+        with socket.create_connection((host, int(port)), timeout=1.5):
+            return True
+    except Exception:
+        return False
+
+
+def _rpc_targets(front, local_rpc: list) -> list:
+    """This machine's RPC servers and every node's, one row each."""
+    out = []
+    for r in local_rpc or []:
+        d = dict(r)
+        d["node"] = "(this machine)"
+        d["reachable"] = bool(d.get("listening"))
+        out.append(d)
+    for e in front.registry.remotes():
+        for r in (e.rpc_servers or []):
+            d = dict(r)
+            d["node"] = e.name
+            ep = d.get("endpoint") or ""
+            d["reachable"] = _rpc_reachable(ep) if ep else False
+            out.append(d)
+    return out
+
+
+def build_overview(front, local_rpc: list) -> dict:
+    engines = front.engines_list()
+    local = [e for e in engines if not e.get("remote")]
+    nodes = [e for e in engines if e.get("remote")]
+    ccfg = getattr(front.config, "cluster", None)
+    return {
+        "generated_at": time.time(),
+        "cluster_enabled": bool(getattr(ccfg, "enabled", False)) if ccfg else False,
+        "node_name": (getattr(ccfg, "node_name", "") or "") if ccfg else "",
+        "serve": bool(getattr(ccfg, "serve", True)) if ccfg else True,
+        "engines": local,
+        "nodes": nodes,
+        "rpc_servers": _rpc_targets(front, local_rpc),
+        "hosts": _host_targets(front),
+        "remotes": _remote_targets(front),
+        "pods": _pods(),
+    }

codai/config.py

@@ -388,6 +388,53 @@ class RemotesConfig:
     pods: dict = field(default_factory=dict)
 
 
+@dataclass
+class ClusterConfig:
+    """Other coderai installs on your network, used as engines of this one.
+
+    A **node** is a whole coderai running somewhere else — its own front, its
+    own cards, its own admin — that this front treats exactly like one of the
+    engine processes it spawns: it polls the node's aggregated engine state,
+    assigns models to it, pins a model to it by name from the Models page, and
+    routes requests there by capability and load. The node needs nothing but an
+    API token; the head needs the node's URL and that token.
+
+    The same section also declares the **llama.cpp RPC servers** THIS machine
+    contributes (`rpc_servers`), so a GGUF model on another node can be spread
+    over this machine's cards too (per-model ``rpc_servers`` on the model). The
+    binary is built by ``packaging/build-rpc-server.sh`` from the very same
+    llama.cpp the bundled llama-cpp-python was built from — the RPC protocol is
+    versioned and the two must match.
+
+    Nodes: ``[{"name": "box2", "url": "https://box2:8776", "api_key": "…",
+              "verify": "system" | "off" | "pem", "ca_pem": "-----BEGIN…",
+              "capabilities": [...], "enabled": true}]``
+    ``capabilities`` blank = whatever the node reports (union of its engines).
+    """
+    enabled: bool = False
+    nodes: list = field(default_factory=list)
+    # What THIS install calls itself when it is a node (matches the head's
+    # ``cluster.nodes[].name`` and a model's ``node_paths`` key). Blank = hostname.
+    node_name: str = ""
+    # How often (s) the head polls each node for state — the same short timeout
+    # the local engines get, but a node is a network hop away.
+    poll_timeout_s: float = 4.0
+    # Answer /cluster/* to callers holding a valid API token of this install.
+    # Off = this install can be a head but never a node.
+    serve: bool = True
+    # llama.cpp RPC servers this machine runs (one process each):
+    # [{"name": "3090", "host": "0.0.0.0", "port": 50052, "device": "CUDA0",
+    #   "mem_gb": 0, "threads": 0, "enabled": true}]
+    # ``device`` names a ggml device (CUDA0, Vulkan1 …; blank = the server's
+    # default); ``mem_gb`` caps what it advertises (0 = all).
+    rpc_servers: list = field(default_factory=list)
+    rpc_bin: str = ""                        # path to rpc-server; blank = auto-find
+    # The address other machines reach THIS one at (for the RPC endpoints it
+    # advertises and the ray head it starts). Blank = best guess from the
+    # default route.
+    advertise_host: str = ""
+
+
 @dataclass
 class Ds4Config:
     """DeepSeek V4 via ds4 (antirez/DwarfStar) external-worker configuration.
@@ -561,6 +608,14 @@ class KtransformersConfig:
     extra_args: str = ""                   # extra flags for sglang.launch_server (e.g. --tp-size, KT knobs)
     extra_env: str = ""                    # free-form KEY=VALUE env for the subprocess
     auto_build: bool = False               # pip-install SGLang+kt-kernel if missing (heavy; off by default)
+    # Multi-node SGLang: this process is rank 0 of ``nnodes``; every other rank is
+    # started by its ``start_cmd`` (templated with {rank}, {nnodes},
+    # {dist_init_addr}, {model}) — ssh, docker, a systemd unit, whatever starts
+    # the same SGLang on that machine. Per-model ``kt`` blocks override these.
+    nnodes: int = 1
+    dist_init_addr: str = ""               # host:port of rank 0 as the others reach it; blank = advertise_host:20000
+    tp_size: int = 0                       # --tp-size across ALL nodes' GPUs (0 = SGLang default)
+    nodes: list = field(default_factory=list)   # [{"name","start_cmd","stop_cmd"}] for ranks 1..n-1
 
 
 @dataclass
@@ -605,6 +660,19 @@ class VllmConfig:
     extra_args: str = ""                 # extra flags for the api_server
     extra_env: str = ""                  # free-form KEY=VALUE env for the subprocess
     auto_build: bool = False             # create the isolated venv + pip install vllm if missing
+    # Multi-node vLLM over Ray: tensor parallel inside a node, pipeline parallel
+    # across nodes. With ``nodes`` set (or ``pipeline_parallel_size`` > 1, or the
+    # executor forced to ray) coderai starts a ray head here, runs each node's
+    # ``start_cmd`` (templated with {ray_address} — the coderai-vllm image joins
+    # with CODERAI_RAY_ADDRESS), waits until ray sees tp*pp GPUs, then launches
+    # vLLM with --distributed-executor-backend ray. Per-model ``vllm`` blocks
+    # override all of these.
+    pipeline_parallel_size: int = 1      # --pipeline-parallel-size
+    distributed_executor_backend: str = ""   # "" auto | "mp" | "ray"
+    ray_address: str = ""                # join an EXISTING ray cluster instead of starting a head
+    ray_port: int = 6379                 # the head's port when coderai starts it
+    nodes: list = field(default_factory=list)   # [{"name","start_cmd","stop_cmd","gpus"}]
+    nodes_ready_timeout_s: int = 600     # how long to wait for the nodes' GPUs to join
 
 
 @dataclass
@@ -753,6 +821,7 @@ class Config:
     jobs: JobsConfig = field(default_factory=JobsConfig)
     enhance: EnhanceConfig = field(default_factory=EnhanceConfig)
     remotes: RemotesConfig = field(default_factory=RemotesConfig)
+    cluster: ClusterConfig = field(default_factory=ClusterConfig)
     ds4: Ds4Config = field(default_factory=Ds4Config)
     colibri: ColibriConfig = field(default_factory=ColibriConfig)
     k3: K3Config = field(default_factory=K3Config)
@@ -945,6 +1014,7 @@ class ConfigManager:
                 jobs=_dc(JobsConfig, config_data.get("jobs", {})),
                 enhance=_dc(EnhanceConfig, config_data.get("enhance", {})),
                 remotes=_dc(RemotesConfig, config_data.get("remotes", {})),
+                cluster=_dc(ClusterConfig, config_data.get("cluster", {})),
                 ds4=_dc(Ds4Config, config_data.get("ds4", {})),
                 colibri=_dc(ColibriConfig, config_data.get("colibri", {})),
                 k3=_dc(K3Config, config_data.get("k3", {})),
@@ -1129,6 +1199,7 @@ class ConfigManager:
                 "endpoints": dict(self.config.remotes.endpoints or {}),
                 "pods": dict(self.config.remotes.pods or {}),
             },
+            "cluster": __import__("dataclasses").asdict(self.config.cluster),
             "ds4": {
                 "enabled": self.config.ds4.enabled,
                 "service_url": self.config.ds4.service_url,
@@ -1198,6 +1269,10 @@ class ConfigManager:
                 "extra_args": self.config.ktransformers.extra_args,
                 "extra_env": self.config.ktransformers.extra_env,
                 "auto_build": self.config.ktransformers.auto_build,
+                "nnodes": self.config.ktransformers.nnodes,
+                "dist_init_addr": self.config.ktransformers.dist_init_addr,
+                "tp_size": self.config.ktransformers.tp_size,
+                "nodes": list(self.config.ktransformers.nodes or []),
             },
             "vllm": {
                 "enabled": self.config.vllm.enabled,
@@ -1217,6 +1292,12 @@ class ConfigManager:
                 "extra_args": self.config.vllm.extra_args,
                 "extra_env": self.config.vllm.extra_env,
                 "auto_build": self.config.vllm.auto_build,
+                "pipeline_parallel_size": self.config.vllm.pipeline_parallel_size,
+                "distributed_executor_backend": self.config.vllm.distributed_executor_backend,
+                "ray_address": self.config.vllm.ray_address,
+                "ray_port": self.config.vllm.ray_port,
+                "nodes": list(self.config.vllm.nodes or []),
+                "nodes_ready_timeout_s": self.config.vllm.nodes_ready_timeout_s,
             },
             "runpod": {
                 "enabled": self.config.runpod.enabled,

codai/frontproxy/registry.py

@@ -99,10 +99,27 @@ class Engine:
     # front can synthesize Tasks-page entries for work it dispatched — visible even
     # when the engine is too GIL-busy generating to answer its own /admin/api/tasks.
     active: dict = field(default_factory=dict, repr=False, compare=False)
+    # A cluster NODE: another coderai install used as an engine of this front.
+    # Nothing is spawned; the node's own front is polled on ``state_path`` and
+    # proxied to through ``http_short``/``http_long`` (clients carrying the
+    # node's token and trust settings — codai/cluster/nodes.py). The node's
+    # engines, cards and thermal state are its own business.
+    remote: bool = False
+    state_path: str = "/internal/engine-state"
+    http_short: object = field(default=None, repr=False, compare=False)
+    http_long: object = field(default=None, repr=False, compare=False)
+    http_sync: object = field(default=None, repr=False, compare=False)
+    node_engines: list = field(default_factory=list)   # what the node reports
+    rpc_servers: list = field(default_factory=list)    # RPC endpoints it advertises
+    last_error: str = ""
+    caps_fixed: bool = False       # cluster.nodes narrowed the capabilities: keep them
+    node_sig: object = field(default=None, repr=False, compare=False)
 
     def __post_init__(self):
         if not self.url:
             self.url = f"http://127.0.0.1:{self.port}"
+        if self.remote:
+            self.state_path = "/cluster/state"
         if not self.name:
             self.name = f"engine#{self.id}"
         if not self.capabilities:
@@ -130,6 +147,10 @@ class Engine:
         diverts new traffic elsewhere and the existing requests can finish."""
         if self.draining:
             return False
+        if self.remote:
+            # No process to ask: a node is alive exactly when its last poll
+            # answered. (A busy node still answers — its front is torch-free.)
+            return self.healthy
         p = self.proc
         try:
             return p is None or p.poll() is None
@@ -205,13 +226,32 @@ class EngineRegistry:
                 match = match or e
         return match
 
+    def remove(self, engine_id: int) -> None:
+        with self._lock:
+            self._engines.pop(engine_id, None)
+
+    def remotes(self) -> List[Engine]:
+        with self._lock:
+            return [e for e in self._engines.values() if e.remote]
+
     def update_state(self, engine_id: int, *, healthy: bool,
                      loaded_models=None, loaded_info=None, vram=None,
-                     tasks=None, cooling=False) -> None:
+                     tasks=None, cooling=False, capabilities=None,
+                     node_engines=None, rpc_servers=None, last_error=None) -> None:
         with self._lock:
             e = self._engines.get(engine_id)
             if not e:
                 return
+            # A node reports what it can do; believe it unless the operator
+            # narrowed the list in cluster.nodes.
+            if capabilities is not None and e.remote and not e.caps_fixed:
+                e.capabilities = set(capabilities) or e.capabilities
+            if node_engines is not None:
+                e.node_engines = list(node_engines)
+            if rpc_servers is not None:
+                e.rpc_servers = list(rpc_servers)
+            if last_error is not None:
+                e.last_error = last_error
             if e.draining:        # a restart is pending — stay out of rotation
                 healthy = False
             e.healthy = healthy

codai/frontproxy/ui_pages.py

@@ -160,6 +160,13 @@ def register_ui_pages(app: FastAPI, config_dir) -> bool:
             return redir
         return _tmpl(request, "runpod.html", {"username": u, "is_admin": True})
 
+    @app.get("/admin/cluster", include_in_schema=False)
+    async def _cluster_page(request: Request):
+        u, redir = _auth_or_redirect(request, admin=True)
+        if redir:
+            return redir
+        return _tmpl(request, "cluster.html", {"username": u, "is_admin": True})
+
     @app.get("/admin/archive", include_in_schema=False)
     async def _archive_page(request: Request):
         u, redir = _auth_or_redirect(request, admin=True)

codai/models/manager.py

@@ -1676,6 +1676,12 @@ class MultiModelManager:
                         kwargs[_kvk] = _kvv
                 if _raw and '_raw_cfg' not in kwargs:
                     kwargs['_raw_cfg'] = _raw
+                # Cards on other machines (llama.cpp RPC), per model only.
+                _rpc = config.get('rpc_servers') if isinstance(config, dict) else None
+                if not _rpc:
+                    _rpc = _raw.get('rpc_servers')
+                if _rpc:
+                    kwargs['rpc_servers'] = _rpc
                 # Cross-backend GPU pooling: per-model overrides global (global_args).
                 kwargs['gpu_split'] = _cfg_or_global('gpu_split', 'gpu_split', False)
                 _ts_val = _cfg_or_global('tensor_split', 'tensor_split', None)
@@ -1846,6 +1852,12 @@ class MultiModelManager:
                         kwargs[_kvk] = _kvv
                 if _raw and '_raw_cfg' not in kwargs:
                     kwargs['_raw_cfg'] = _raw
+                # Cards on other machines (llama.cpp RPC), per model only.
+                _rpc = config.get('rpc_servers') if isinstance(config, dict) else None
+                if not _rpc:
+                    _rpc = _raw.get('rpc_servers')
+                if _rpc:
+                    kwargs['rpc_servers'] = _rpc
                 # Cross-backend GPU pooling: per-model overrides global (global_args).
                 kwargs['gpu_split'] = _cfg_or_global('gpu_split', 'gpu_split', False)
                 _ts_val = _cfg_or_global('tensor_split', 'tensor_split', None)

docs/frontend-engine-split.md

@@ -291,3 +291,10 @@ it is **respawned**; the front and sibling engines keep serving. The front's own
   transformers), matching the engine's own `is_gguf` heuristic. A transformers model
   whose name happens to contain "gguf" would be mis-routed — rename or declare an
   alias if that ever bites.
+
+## Engines on other machines
+
+A **cluster node** — another coderai install — joins the registry as one
+more engine: polled on its `/cluster/state`, assigned models, pinned by
+name, routed by capability and load, never spawned. See
+[`cluster.md`](cluster.md).

docs/remote-execution.md

@@ -15,6 +15,8 @@ by what you are moving.
 | Any text model, GGUF included | `service_url` on the model entry | models.json |
 | Images, video, embeddings, rerank, OCR, TTS, STT, voice, audio, stems, 3D, pipelines | remote gateway | `remotes.endpoints` in config.json |
 | Any model, on a machine you already have (always-on or started by command) | `host` backend | `"backend": "host"` on the model entry |
+| Another whole coderai as an engine of this one (routing by capability/load, pins by name) | cluster node | `cluster.nodes` in config.json — [`cluster.md`](cluster.md) |
+| One model over several machines' cards | llama.cpp RPC / vLLM on Ray / SGLang multi-node | per model — [`cluster.md`](cluster.md) |
 
 ---
 

packaging/build-rpc-server.sh

+#!/usr/bin/env bash
+# Build llama.cpp's rpc-server from the SAME llama.cpp the installed
+# llama-cpp-python was built from, and put it beside the venv's python.
+#
+# The RPC protocol is versioned: a client and a server from different
+# llama.cpp commits refuse each other. llama-cpp-python's sdist vendors the
+# exact llama.cpp tree its wheel was built from, so that tree — not a fresh
+# clone of master — is what rpc-server must come from. The binary is what a
+# machine runs to lend its cards to a GGUF loaded elsewhere
+# (codai/cluster/rpc.py starts it from cluster.rpc_servers; the app's own
+# loads reach it through codai/backends/ggml_rpc.py).
+#
+# Usage:  packaging/build-rpc-server.sh [--cuda] [--vulkan] [--dest DIR]
+#   Backends default to whatever the installed llama-cpp-python has (it is
+#   probed by looking at its lib/ directory). The python is the one on PATH
+#   (activate the venv first) unless PYTHON is set.
+set -euo pipefail
+
+PYTHON="${PYTHON:-python}"
+CUDA=""; VULKAN=""; DEST=""
+while [ $# -gt 0 ]; do
+    case "$1" in
+        --cuda) CUDA=1 ;;
+        --vulkan) VULKAN=1 ;;
+        --dest) DEST="$2"; shift ;;
+        -h|--help) sed -n '2,16p' "$0"; exit 0 ;;
+        *) echo "unknown option $1" >&2; exit 2 ;;
+    esac
+    shift
+done
+
+VER=$("$PYTHON" -c "import llama_cpp; print(llama_cpp.__version__)")
+LIBDIR=$("$PYTHON" -c "import llama_cpp, os; print(os.path.join(os.path.dirname(llama_cpp.__file__), 'lib'))")
+BINDIR=$("$PYTHON" -c "import sys, os; print(os.path.dirname(sys.executable))")
+DEST="${DEST:-$BINDIR}"
+if [ -z "$CUDA$VULKAN" ]; then
+    ls "$LIBDIR" | grep -q 'libggml-cuda' && CUDA=1 || true
+    ls "$LIBDIR" | grep -q 'libggml-vulkan' && VULKAN=1 || true
+fi
+echo "llama-cpp-python $VER  (cuda=${CUDA:-0} vulkan=${VULKAN:-0})  -> $DEST/rpc-server"
+
+WORK=$(mktemp -d)
+trap 'rm -rf "$WORK"' EXIT
+cd "$WORK"
+"$PYTHON" -m pip download --no-deps --no-binary :all: --no-build-isolation \
+    "llama-cpp-python==$VER" -d . >/dev/null 2>&1 || \
+"$PYTHON" -m pip download --no-deps --no-binary :all: "llama-cpp-python==$VER" -d .
+SDIST=$(ls llama_cpp_python-*.tar.gz | head -1)
+tar -xzf "$SDIST"
+SRC=$(ls -d llama_cpp_python-*/vendor/llama.cpp | head -1)
+[ -f "$SRC/CMakeLists.txt" ] || { echo "vendored llama.cpp not found in $SDIST" >&2; exit 1; }
+
+CM=(-DGGML_RPC=ON -DBUILD_SHARED_LIBS=OFF -DLLAMA_BUILD_TESTS=OFF -DLLAMA_BUILD_EXAMPLES=OFF
+    -DLLAMA_BUILD_SERVER=OFF -DLLAMA_BUILD_TOOLS=ON -DLLAMA_CURL=OFF -DGGML_NATIVE=OFF -DGGML_AVX2=ON)
+[ -n "$CUDA" ] && CM+=(-DGGML_CUDA=ON ${CMAKE_CUDA_ARCHITECTURES:+-DCMAKE_CUDA_ARCHITECTURES="$CMAKE_CUDA_ARCHITECTURES"})
+[ -n "$VULKAN" ] && CM+=(-DGGML_VULKAN=ON)
+cmake -S "$SRC" -B build "${CM[@]}" ${EXTRA_CMAKE_ARGS:-} >/dev/null
+cmake --build build --target rpc-server -j"$(nproc)"
+BIN=$(find build -type f -name rpc-server | head -1)
+[ -x "$BIN" ] || { echo "rpc-server did not build" >&2; exit 1; }
+install -m 755 "$BIN" "$DEST/rpc-server"
+echo "installed $DEST/rpc-server"
+"$DEST/rpc-server" --help 2>&1 | head -3 || true

packaging/linux/Dockerfile.oci

@@ -65,9 +65,20 @@ RUN mkdir -p /opt/wheels
 
 RUN set -eux; \
     pkg="llama-cpp-python${LLAMA_CPP_PYTHON_VERSION:+==${LLAMA_CPP_PYTHON_VERSION}}"; \
-    CMAKE_ARGS="-DGGML_CUDA=ON -DGGML_VULKAN=ON" \
+    CMAKE_ARGS="-DGGML_CUDA=ON -DGGML_VULKAN=ON -DGGML_RPC=ON" \
       uv pip wheel --python /opt/coderai/python/bin/python3 --wheel-dir /opt/wheels --no-deps --no-cache "$pkg"
 
+# rpc-server from the SAME vendored llama.cpp (the RPC protocol is versioned):
+# what this install runs to lend its cards to a GGUF loaded on another machine
+# (docs/cluster.md). Static ggml so the binary needs only the CUDA/Vulkan
+# runtime that is already in the image.
+COPY packaging/build-rpc-server.sh /tmp/build-rpc-server.sh
+RUN set -eux; \
+    uv pip install --python /opt/coderai/python/bin/python3 --find-links /opt/wheels --no-deps --no-index llama-cpp-python; \
+    PYTHON=/opt/coderai/python/bin/python3 CMAKE_CUDA_ARCHITECTURES="80;86;89;90;120" \
+      bash /tmp/build-rpc-server.sh --cuda --vulkan --dest /opt/coderai/python/bin \
+      || echo "WARNING: rpc-server not built; cluster RPC serving unavailable in this image"
+
 RUN set -eux; \
     pkg="stable-diffusion-cpp-python${SD_CPP_PYTHON_VERSION:+==${SD_CPP_PYTHON_VERSION}}"; \
     CMAKE_ARGS="-DSD_CUDA=ON -DSD_VULKAN=ON -DSD_WEBM=OFF" \

packaging/runpod/Dockerfile.capability-llama

@@ -19,11 +19,20 @@ RUN apt-get update && apt-get install -y --no-install-recommends \
     && rm -rf /var/lib/apt/lists/*
 # Ubuntu 24.04's python3 is 3.12, the same as the core: the wheel is cp312 either way.
 RUN python3 -m venv /bw && /bw/bin/pip install --no-cache-dir -U pip build scikit-build-core
-ENV CMAKE_ARGS="-DGGML_CUDA=on -DCMAKE_CUDA_ARCHITECTURES=80;86;89;90;120 -DGGML_NATIVE=off -DGGML_AVX2=on"
+ENV CMAKE_ARGS="-DGGML_CUDA=on -DGGML_RPC=on -DCMAKE_CUDA_ARCHITECTURES=80;86;89;90;120 -DGGML_NATIVE=off -DGGML_AVX2=on"
 RUN --mount=type=cache,target=/root/.cache/pip set -eux; \
     /bw/bin/pip wheel --no-cache-dir --no-deps --no-binary llama-cpp-python \
         "llama-cpp-python==${LLAMA_CPP_PYTHON}" -w /wheels; \
     ls -la /wheels
+# rpc-server from the same vendored llama.cpp, so this pod (or a host running
+# this image with CODERAI_RPC_SERVER_PORT) can lend its card to a GGUF loaded
+# elsewhere. Static ggml; needs only the CUDA runtime the pod already has.
+COPY packaging/build-rpc-server.sh /tmp/build-rpc-server.sh
+RUN --mount=type=cache,target=/root/.cache/pip set -eux; \
+    /bw/bin/pip install --no-cache-dir --no-deps /wheels/llama_cpp_python-*.whl; \
+    PYTHON=/bw/bin/python CMAKE_CUDA_ARCHITECTURES="80;86;89;90;120" \
+      bash /tmp/build-rpc-server.sh --cuda --dest /wheels; \
+    ls -la /wheels/rpc-server
 
 # -------------------------------------------------------------- runtime stage
 FROM ${CORE_BASE}
@@ -40,6 +49,7 @@ COPY --from=wheel /wheels/ /tmp/wheels/
 RUN --mount=type=cache,target=/root/.cache/pip set -eux; \
     python -m pip install --extra-index-url ${TORCH_INDEX} -r /tmp/profiles/${PROFILE}.txt; \
     python -m pip install --no-cache-dir /tmp/wheels/llama_cpp_python-*.whl; \
+    install -m 755 /tmp/wheels/rpc-server /usr/local/bin/rpc-server; \
     if python -c "import torch" 2>/dev/null; then \
         echo "ERROR: llama is a light-core image but torch reached its main venv" >&2; exit 1; \
     fi; echo "main venv: no torch (light core, as intended)"; \

packaging/runpod/boot.sh

@@ -69,6 +69,39 @@ if [ -n "$CODERAI_TLS_CERT" ] && [ -n "$CODERAI_TLS_KEY" ]; then
     phase "tls: serving https with the certificate the renting coderai sent"
 fi
 
+# A ray WORKER instead of a server: the vllm image joins a multi-node vLLM
+# launch whose head is the coderai that started this container
+# (codai/cluster/multinode.py). Nothing else runs; the head drives the GPUs.
+if [ -n "${CODERAI_RAY_ADDRESS:-}" ]; then
+    RAY_BIN="${CODERAI_VLLM_VENV:-/opt/coderai/venvs/vllm}/bin/ray"
+    [ -x "$RAY_BIN" ] || RAY_BIN="ray"
+    phase "ray worker: joining ${CODERAI_RAY_ADDRESS} with ${RAY_BIN}"
+    exec "$RAY_BIN" start --address="${CODERAI_RAY_ADDRESS}" --block \
+         ${CODERAI_RAY_NODE_IP:+--node-ip-address="$CODERAI_RAY_NODE_IP"} \
+         --disable-usage-stats
+fi
+
+# A llama.cpp rpc-server beside the app: this pod's/host's cards become
+# devices of a GGUF loaded elsewhere (codai/backends/ggml_rpc.py). The port
+# must be exposed (direct TCP on a pod, -p on docker). Unauthenticated
+# protocol — LAN, WireGuard or a pod's direct port with a firewall only.
+if [ -n "${CODERAI_RPC_SERVER_PORT:-}" ]; then
+    RPC_BIN="${CODERAI_RPC_SERVER_BIN:-/usr/local/bin/rpc-server}"
+    if [ -x "$RPC_BIN" ]; then
+        phase "rpc-server: listening on 0.0.0.0:${CODERAI_RPC_SERVER_PORT}${CODERAI_RPC_SERVER_DEVICE:+ device $CODERAI_RPC_SERVER_DEVICE}"
+        "$RPC_BIN" -H 0.0.0.0 -p "$CODERAI_RPC_SERVER_PORT" \
+            ${CODERAI_RPC_SERVER_DEVICE:+-d "$CODERAI_RPC_SERVER_DEVICE"} \
+            ${CODERAI_RPC_SERVER_MEM_MB:+-m "$CODERAI_RPC_SERVER_MEM_MB"} &
+        if [ "${CODERAI_RPC_SERVER_ONLY:-0}" = "1" ]; then
+            phase "rpc-server only: no API server on this pod"
+            wait
+            exit $?
+        fi
+    else
+        phase "rpc-server requested but $RPC_BIN is not in this image"
+    fi
+fi
+
 phase "starting uvicorn on 0.0.0.0:8000${TLS_ARGS:+ (https)}"
 exec python -m uvicorn codai.api.app:app --host 0.0.0.0 --port 8000 \
      --log-level "${CODERAI_LOG_LEVEL:-info}" $TLS_ARGS