cluster: work-parallel fan-out of one request over engines and nodes

Per-model 'distribute': n images/videos, embedding and rerank lists, TTS
sentences, transcriptions cut at silences, OCR batches are split over every
engine that has the model — here and on cluster nodes — and merged back.
Files a node generates are addressed through the head. Failed parts retry
once elsewhere. Models page block + tests.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FKm7Z56iKjHgNkHc6VhWKz

codai/admin/routes.py

@@ -2756,6 +2756,28 @@ async def api_model_configure(request: Request, username: str = Depends(require_
                 ):
         if key in data:
             entry[key] = data[key]
+    # Work-parallel fan-out over engines and nodes (codai/cluster/fanout.py).
+    if "distribute" in data:
+        src = data.get("distribute") if isinstance(data.get("distribute"), dict) else {}
+        blk = {}
+        if src.get("enabled"):
+            blk["enabled"] = True
+            nodes = src.get("nodes")
+            if isinstance(nodes, str):
+                nodes = [x.strip() for x in nodes.split(",") if x.strip()]
+            if nodes and nodes != ["auto"]:
+                blk["nodes"] = [str(x) for x in nodes]
+            for k in ("max_parts", "min_items", "chunk_seconds"):
+                v = src.get(k)
+                if v not in (None, ""):
+                    try:
+                        blk[k] = int(v)
+                    except (TypeError, ValueError):
+                        pass
+        if blk:
+            entry["distribute"] = blk
+        else:
+            entry.pop("distribute", None)
     # Per-model engine blocks laid over the global template: vLLM (parallelism,
     # ray nodes), SGLang/kt (nnodes, ranks' commands). A block with nothing set
     # is dropped so the entry stays clean.

codai/admin/templates/models.html

@@ -622,6 +622,21 @@ window.__DEFAULT_WHISPER_SERVER_PATH__ = {{ default_whisper_server_path|tojson }
         <input type="text" id="cfg-node-path" class="form-input" placeholder="blank = same path as here, else the HuggingFace id / URL the node can fetch">
         <span class="form-hint" style="font-size:11px">The node registers the model under this path when it does not share this machine's filesystem. Blank works when the path exists on the node too (NFS, same layout) or the model is a HuggingFace id.</span>
       </div>
+      <div class="form-row" id="cfg-dist-row" style="margin-top:.75rem;border:1px solid var(--border,#333);border-radius:8px;padding:.6rem">
+        <label style="display:flex;align-items:center;gap:.5rem;font-weight:600">
+          <input type="checkbox" id="cfg-dist-enabled" style="width:auto;margin:0" onchange="_updateDistVisibility()">
+          Distribute the work of one request across engines and nodes
+        </label>
+        <span class="form-hint" style="font-size:11px;line-height:1.5">Splits what a request carries — <b>n</b> images or videos, a list to embed or rerank, a long text to speak (by sentences), a long recording to transcribe (cut at silences), a batch of documents to OCR — over every engine that has this model, here and on cluster nodes, and merges the answers. One generation is never split; that is <em>Cards on other machines</em> / vLLM nodes above.</span>
+        <div id="cfg-dist-fields" style="display:none;margin-top:.5rem">
+          <div style="display:grid;grid-template-columns:2fr 1fr 1fr 1fr;gap:.6rem">
+            <div class="form-row" style="margin:0"><label class="form-label">Engines / nodes <span class="muted">(blank = every one that can)</span></label><input type="text" id="cfg-dist-nodes" class="form-input" placeholder="auto — or: local, box2, box3"></div>
+            <div class="form-row" style="margin:0"><label class="form-label">Max parts</label><input type="number" id="cfg-dist-max" class="form-input" min="0" placeholder="0 = no cap"></div>
+            <div class="form-row" style="margin:0"><label class="form-label">Min items to split</label><input type="number" id="cfg-dist-min" class="form-input" min="2" placeholder="2"></div>
+            <div class="form-row" style="margin:0"><label class="form-label">Audio chunk (s)</label><input type="number" id="cfg-dist-chunk" class="form-input" min="10" placeholder="120"></div>
+          </div>
+        </div>
+      </div>
       <div class="form-row" id="cfg-rpc-row" style="margin-top:.75rem;display:none;border:1px solid var(--border,#333);border-radius:8px;padding:.6rem">
         <label class="form-label" style="font-weight:600">Cards on other machines <span class="muted" style="font-weight:400">— llama.cpp RPC servers, one model over several machines</span></label>
         <input type="text" id="cfg-rpc-servers" class="form-input" list="cfg-rpc-known" placeholder="host:50052, host2:50052   (blank = this machine only)">
@@ -4311,9 +4326,18 @@ function _applyCompat(){
   const krow = document.getElementById('cfg-kt-row');
   if (krow) krow.style.display = (backend === 'kt') ? '' : 'none';
 }
+function _updateDistVisibility(){
+  const on = !!(document.getElementById('cfg-dist-enabled') || {}).checked;
+  const f = document.getElementById('cfg-dist-fields'); if (f) f.style.display = on ? '' : 'none';
+}
 function _populateCluster(m, s){
   window._cfgCurrentPath = (m && (m.path || m.id)) || '';
   const set = (id,v)=>{ const el=document.getElementById(id); if(el!=null) el.value=(v==null?'':v); };
+  const dz = s.distribute || {};
+  document.getElementById('cfg-dist-enabled').checked = !!dz.enabled;
+  set('cfg-dist-nodes', Array.isArray(dz.nodes) ? dz.nodes.join(', ') : (dz.nodes || ''));
+  set('cfg-dist-max', dz.max_parts || ''); set('cfg-dist-min', dz.min_items || ''); set('cfg-dist-chunk', dz.chunk_seconds || '');
+  _updateDistVisibility();
   set('cfg-node-path', s.node_path || '');
   set('cfg-rpc-servers', Array.isArray(s.rpc_servers) ? s.rpc_servers.join(', ') : (s.rpc_servers || ''));
   const dl = document.getElementById('cfg-rpc-known');
@@ -4341,6 +4365,11 @@ function _collectCluster(){
   const out = {
     node_path: val('cfg-node-path'),
     rpc_servers: val('cfg-rpc-servers'),
+    distribute: {
+      enabled: !!document.getElementById('cfg-dist-enabled').checked,
+      nodes: val('cfg-dist-nodes'), max_parts: num('cfg-dist-max'),
+      min_items: num('cfg-dist-min'), chunk_seconds: num('cfg-dist-chunk'),
+    },
   };
   out.vllm = backend === 'vllm' ? {
     tensor_parallel_size: num('cfg-vllm-tp'), pipeline_parallel_size: num('cfg-vllm-pp'),

codai/frontproxy/app.py

@@ -15,6 +15,7 @@ UI stays live even while an engine is busy loading a model.
 """
 
 import json
+import re
 import sys
 import time
 from typing import Optional
@@ -113,6 +114,7 @@ class FrontProxy:
         self._mls_cache: dict = {}
         self.registry = EngineRegistry()
         self.supervisor: Optional[EngineSupervisor] = None
+        self._node_files: dict = {}
         # Per-run secret shared only with the engines (passed via env at spawn). The
         # front stamps every engine request with it and engines reject requests that
         # lack it, so nothing on localhost can talk to an engine bypassing the front.
@@ -640,6 +642,97 @@ class FrontProxy:
     def _pin_for(self, model: Optional[str]) -> Optional[str]:
         return self._model_info(model).get("engine")
 
+    #: Generated files that live on a cluster node: name -> engine. A node's
+    #: answer names its files by ITS base URL; the head rewrites them to its
+    #: own and fetches from the node when the client comes for them.
+    _FILE_URL_RE = re.compile(rb'https?://[^"\s]*?/v1/files/([^"\s/?]+)')
+
+    def _rewrite_node_files(self, engine, raw: bytes, request: Request) -> bytes:
+        if not getattr(engine, "remote", False) or b"/v1/files/" not in (raw or b""):
+            return raw
+        proto = (request.headers.get("x-forwarded-proto") or request.url.scheme or "http").split(",")[0]
+        host = request.headers.get("host") or ""
+        prefix = (request.headers.get("x-forwarded-prefix") or "").rstrip("/")
+        base = f"{proto}://{host}{prefix}".encode("latin-1") if host else prefix.encode("latin-1")
+        files = self._node_files
+
+        def _sub(m):
+            files[m.group(1).decode("latin-1")] = engine
+            return base + b"/v1/files/" + m.group(1)
+        return self._FILE_URL_RE.sub(_sub, raw)
+
+    async def _serve_node_file(self, request: Request, name: str):
+        engine = self._node_files.get(name)
+        if engine is None or not engine.healthy:
+            return None
+        try:
+            rp_req = self._lc(engine).build_request(
+                "GET", engine.url + "/v1/files/" + name, headers=self._eh(engine, {}))
+            rp_resp = await self._lc(engine).send(rp_req, stream=True)
+        except Exception as exc:
+            return JSONResponse({"error": f"node {engine.name} unreachable: {exc}"},
+                                status_code=502)
+
+        async def _it():
+            try:
+                async for chunk in rp_resp.aiter_raw():
+                    yield chunk
+            finally:
+                await rp_resp.aclose()
+        return StreamingResponse(_it(), status_code=rp_resp.status_code,
+                                 headers=dict(self._filter_headers(rp_resp.headers, _DROP_RESP)))
+
+    async def _maybe_fanout(self, request: Request, path: str, model: Optional[str],
+                            body_bytes: Optional[bytes]):
+        """Split this request over several engines when the model asks for it
+        and the request carries more than one unit of work; None = serve as
+        one request. Never raises: a fan-out that cannot happen falls back."""
+        from codai.cluster import fanout as _fan
+        kind = _fan.kind_for(path)
+        if kind is None:
+            return None
+        ct = (request.headers.get("content-type") or "").lower()
+        if model is None and "multipart/form-data" in ct:
+            body_bytes = await request.body()
+            model = self._peek_model_multipart(body_bytes)
+        dcfg = _fan.parse_distribute(self._model_info(model).get("distribute"))
+        if not dcfg["enabled"]:
+            return None
+        cap = self._required_cap(path, model)
+        engines = _fan.choose_engines(self.registry, model or "", cap, dcfg["nodes"],
+                                      dcfg["max_parts"])
+        if len(engines) < 2:
+            return None
+        try:
+            if kind == "audio":
+                await request.body()          # cache it: form() re-reads, fallback forwards
+                res = await _fan.run_transcription(self, request, path, engines, model or "", dcfg)
+            elif kind == "files":
+                await request.body()
+                res = await _fan.run_ocr_batch(self, request, path, engines, model or "", dcfg)
+            else:
+                if "application/json" not in ct or not body_bytes:
+                    return None
+                try:
+                    body = json.loads(body_bytes)
+                except Exception:
+                    return None
+                if not isinstance(body, dict) or body.get("stream"):
+                    return None
+                if kind == "text:input":
+                    res = await _fan.run_speech(self, request, path, body, engines, model or "", dcfg)
+                else:
+                    res = await _fan.run_json(self, request, path, kind, body, engines,
+                                              model or "", dcfg)
+        except Exception as exc:
+            print(f"[fanout] {path} {model}: not split ({exc}); serving as one request",
+                  flush=True)
+            return None
+        if res is None:
+            return None
+        self._record_activity(model, self._task_kind(path), res.status, time.time())
+        return Response(content=res.content, status_code=res.status, media_type=res.media_type)
+
     def _load_pins(self) -> dict:
         import json as _json
         info: dict = {}
@@ -674,7 +767,10 @@ class FrontProxy:
                        # (enabled + triggers + serverless target). Used by the front
                        # to reroute when local can't serve.
                        "runpod_spillover": (m.get("runpod_spillover")
-                                            if isinstance(m.get("runpod_spillover"), dict) else None)}
+                                            if isinstance(m.get("runpod_spillover"), dict) else None),
+                       # Work-parallel fan-out over engines/nodes (cluster/fanout.py).
+                       "distribute": (m.get("distribute")
+                                      if isinstance(m.get("distribute"), dict) else None)}
                 for field_ in (m.get("path"), m.get("id"), m.get("alias")):
                     if not field_:
                         continue
@@ -2137,6 +2233,12 @@ class FrontProxy:
             if sysw is not None:
                 return await self._proxy_passthrough(request, sysw)
 
+        # A generated file that a cluster node holds is fetched from that node.
+        if method == "GET" and path.startswith("/v1/files/") and self._node_files:
+            served = await self._serve_node_file(request, path.rsplit("/", 1)[-1])
+            if served is not None:
+                return served
+
         # Inference JSON bodies are small: buffer so we can route by `model`, then
         # forward the buffered bytes. Everything else streams through unbuffered.
         body_bytes: Optional[bytes] = None
@@ -2145,6 +2247,15 @@ class FrontProxy:
             body_bytes = await request.body()
             model = self._peek_model(body_bytes, request.headers.get("content-type", ""))
 
+        # Work-parallel fan-out (codai/cluster/fanout.py): a model configured to
+        # distribute has its n images / input list / sentences / audio chunks /
+        # document batch spread over every engine that holds it, here and on the
+        # nodes, and the parts merged back into one answer.
+        if method == "POST":
+            fanned = await self._maybe_fanout(request, path, model, body_bytes)
+            if fanned is not None:
+                return fanned
+
         engine = _router.pick_engine(
             self.registry, path, method, model,
             required_cap=self._required_cap(path, model),
@@ -2290,6 +2401,20 @@ class FrontProxy:
                     self._record_activity(model, self._task_kind(path),
                                           rp_resp.status_code, _started)
 
+        # A node's JSON answer may name files by the node's own URL: rewrite
+        # them to this front (small bodies only; streams pass untouched).
+        _rct = (rp_resp.headers.get("content-type") or "").lower()
+        if getattr(engine, "remote", False) and "application/json" in _rct:
+            try:
+                raw = await rp_resp.aread()
+            finally:
+                await _release()
+            raw = self._rewrite_node_files(engine, raw, request)
+            hdrs = dict(self._filter_headers(rp_resp.headers, _DROP_RESP))
+            hdrs.pop("content-length", None)
+            return Response(content=raw, status_code=rp_resp.status_code, headers=hdrs,
+                            media_type=rp_resp.headers.get("content-type"))
+
         # Measure throughput from the SSE stream the front relays, and publish it on
         # the in-flight metadata. This gives the Tasks page a live it/s for the
         # synthesized task even while the engine is too busy to report its own — each