ocr: one address for the engine, every server behind it (v0.3.9)

An OCR pipeline is handed one server address and drives it itself, which is
what keeps its page images, its strict json_schema and its logprobs intact.
One address cannot spread load, and three things followed from that:

  * a pool grows on in-flight requests counted as they pass through it. An
    attached engine never passed through, so the pool saw no load however hard
    OCR ran, never grew, and the engine had one pod's URL anyway;
  * several cluster nodes serving the model resolved to whichever answered
    first, for ever, however busy it became;
  * a local model configured to burst to RunPod never did: that trigger lives
    on the proxy path an attached engine bypasses.

So the engine gets a lease on an address of ours, and codai/ocr/dispatch.py
chooses per request: a pool that is the model's configured home takes every
page (picking the least-loaded pod, renting when they are all at the
threshold); otherwise this machine and the nodes holding the model compete on
load — including the load a node reports for its whole install, so a node busy
with someone else's work is left alone — and the overflow past
ocr.serve_spill_inflight rents instead of queueing. An explicit `backend` pin
still wins over discovery: that was a decision, and load-aware picking must
not quietly repeal it.

The hop is allowed to exist only because it decides nothing: body bytes in,
body bytes out. The losses it is there to avoid came from PARSING a request
and re-emitting it through the backend abstraction.

Two gates stood in the way and neither could be satisfied the usual way: the
engine has no API token, and surya's /health and /v1/models probes send no
headers at all (bare httpx GETs), while refusing to attach if /health is not
200. The lease is therefore read from the URL, and the engine-only internal
gate exempts that path alone — tested to be no wider than that.

olmOCR's model mode asks the same dispatcher, and still takes the in-process
route when the answer is local, where the model manager's loading, eviction
and queueing belong.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PxYHWbCAjjjD7A1rhFKtaW

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.3.8"
+__version__ = "0.3.9"
 __tagline__ = "Complete orchestration, distribution and escalation of remotizable advanced inference."
 
 # Configure the CUDA caching allocator BEFORE torch is imported anywhere.

codai/admin/routes.py

@@ -4140,6 +4140,8 @@ def build_settings_dict(c, gpu_cards):
             "surya_serve": c.ocr.surya_serve,
             "surya_model": c.ocr.surya_model,
             "surya_model_id": c.ocr.surya_model_id,
+            "serve_fanout": c.ocr.serve_fanout,
+            "serve_spill_inflight": c.ocr.serve_spill_inflight,
             "surya_server_url": c.ocr.surya_server_url,
             "build_retry_cooldown_s": c.ocr.build_retry_cooldown_s,
             "evict_drain_timeout_s": c.ocr.evict_drain_timeout_s,
@@ -4938,6 +4940,12 @@ async def api_save_settings(request: Request, username: str = Depends(require_ad
         if "surya_serve" in d: o.surya_serve = (d.get("surya_serve") or "local").strip()
         if "surya_model" in d: o.surya_model = (d.get("surya_model") or "").strip()
         if "surya_model_id" in d: o.surya_model_id = (d.get("surya_model_id") or "").strip()
+        if "serve_fanout" in d: o.serve_fanout = bool(d.get("serve_fanout"))
+        if "serve_spill_inflight" in d:
+            try:
+                o.serve_spill_inflight = max(0, int(d.get("serve_spill_inflight") or 0))
+            except (TypeError, ValueError):
+                pass
         if "surya_server_url" in d: o.surya_server_url = (d.get("surya_server_url") or "").strip()
         if "build_retry_cooldown_s" in d: o.build_retry_cooldown_s = _f("build_retry_cooldown_s", o.build_retry_cooldown_s, 0.0)
         if "vlm_gpu_memory_utilization" in d: o.vlm_gpu_memory_utilization = _f("vlm_gpu_memory_utilization", o.vlm_gpu_memory_utilization, 0.0)

codai/admin/templates/settings.html

@@ -1204,6 +1204,13 @@
       <div class="form-group" style="flex:2;min-width:200px"><label class="form-label">Server URL (External)</label><input type="text" id="s-ocr-surya-server-url" class="form-input" placeholder="http://127.0.0.1:8080/v1"></div>
     </div>
 
+    <div class="card-title" style="font-size:13px;margin-top:.4rem">Scaling in <b>Model</b> mode <span class="muted" style="font-weight:400">(both VLM engines)</span></div>
+    <p class="form-hint" style="margin:0 0 .4rem">In <b>Model</b> mode the OCR engine drives the model server itself, which is what keeps its page images, strict JSON schema and <b>confidence scores</b> intact. <b>Spread the load</b> gives it one address here instead, forwarded per request — unchanged, byte for byte — to the least busy of: this machine's server, the cluster nodes that hold the model, and the RunPod pool. Without it the engine pins to a single pod, and the pool never sees the load so it never grows.</p>
+    <div class="form-row" style="display:flex;gap:1rem;flex-wrap:wrap;align-items:center">
+      <label style="display:flex;align-items:center;gap:.4rem;cursor:pointer"><input type="checkbox" id="s-ocr-serve-fanout"><span>Spread the load across servers</span></label>
+      <div class="form-group" style="flex:1;min-width:200px"><label class="form-label">Rent when busier than <span class="muted">(requests in flight; 0 = never on load)</span></label><input type="number" id="s-ocr-serve-spill" class="form-input" min="0" placeholder="4"></div>
+    </div>
+
     <div class="card-title" style="font-size:13px;margin-top:.4rem">olmOCR-2 <span class="muted" style="font-weight:400">(document VLM)</span></div>
     <p class="form-hint" style="margin:0 0 .4rem"><a href="https://huggingface.co/allenai/olmOCR-2-7B-1025" target="_blank" rel="noopener">allenai/olmOCR-2</a> (Apache-2.0) reads a whole page at once — equations as LaTeX, tables as HTML, reading order sorted — and is the one to reach for on old scans, maths and multi-column pages. It is a <b>vision model</b>, so it returns <b>no bounding boxes</b> and no layout regions: use paddle or Surya when you need those.</p>
     <div class="form-row" style="display:flex;gap:1rem;flex-wrap:wrap;align-items:center">
@@ -2137,6 +2144,8 @@ async function loadSettings(){
     document.getElementById('s-ocr-surya-model').value = ocr.surya_model ?? '';
     document.getElementById('s-ocr-surya-model-id').value = ocr.surya_model_id ?? '';
     document.getElementById('s-ocr-surya-server-url').value = ocr.surya_server_url ?? '';
+    document.getElementById('s-ocr-serve-fanout').checked = ocr.serve_fanout !== false;
+    document.getElementById('s-ocr-serve-spill').value = ocr.serve_spill_inflight ?? 4;
     document.getElementById('s-ocr-olmocr-enabled').checked = !!ocr.olmocr_enabled;
     document.getElementById('s-ocr-olmocr-serve').value = ocr.olmocr_serve ?? 'model';
     document.getElementById('s-ocr-olmocr-instances').value = ocr.olmocr_instances ?? 1;
@@ -2388,6 +2397,8 @@ async function saveSettings(silent){
       surya_serve: document.getElementById('s-ocr-surya-serve').value,
       surya_model: document.getElementById('s-ocr-surya-model').value.trim(),
       surya_model_id: document.getElementById('s-ocr-surya-model-id').value.trim(),
+      serve_fanout: document.getElementById('s-ocr-serve-fanout').checked,
+      serve_spill_inflight: parseInt(document.getElementById('s-ocr-serve-spill').value || '4', 10),
       surya_server_url: document.getElementById('s-ocr-surya-server-url').value.trim(),
       olmocr_enabled: document.getElementById('s-ocr-olmocr-enabled').checked,
       olmocr_serve: document.getElementById('s-ocr-olmocr-serve').value || 'model',

codai/api/app.py

@@ -256,6 +256,17 @@ class _InternalAuthMiddleware:
 
     async def __call__(self, scope, receive, send):
         if self._token and scope.get("type") == "http":
+            # The OCR fan-out gateway is the one exception, and it has to be: the
+            # client is an OCR engine running as a subprocess of THIS process, and
+            # it speaks plain OpenAI — surya's client sends a bearer token and
+            # nothing else, and its /health and /v1/models probes send no headers
+            # at all, so it cannot present the internal token however much we
+            # would like it to. The path is not open: every route under it demands
+            # a live lease (codai.ocr.gateway), a secret minted per attachment and
+            # held only in memory, and BearerAuthMiddleware checks it too.
+            if str(scope.get("path", "")).startswith("/v1/ocr/engine/"):
+                await self._app(scope, receive, send)
+                return
             headers = dict(scope.get("headers", []))
             got = headers.get(b"x-coderai-internal", b"").decode("latin-1")
             if got != self._token:
@@ -725,6 +736,11 @@ app.include_router(loras_router, tags=["LoRAs"])
 app.include_router(environments_router, tags=["Environments"])
 app.include_router(spatial_router, tags=["Spatial / 3D"])
 app.include_router(ocr_router, tags=["OCR"])
+# The address an attached OCR engine drives: one stable URL of ours, forwarded
+# per request to the least-busy server that holds the model. Not in the schema —
+# it is machine-to-machine, authenticated by a lease rather than an API token.
+from codai.ocr.gateway import router as ocr_gateway_router      # noqa: E402
+app.include_router(ocr_gateway_router, include_in_schema=False)
 app.include_router(rerank_router, tags=["Rerank"])
 from codai.api.models_transfer import router as models_transfer_router
 app.include_router(models_transfer_router, tags=["Models"])

codai/api/ratelimit.py

@@ -107,6 +107,24 @@ class BearerAuthMiddleware(BaseHTTPMiddleware):
         if _env_token and auth_header.lower().startswith("bearer "):
             if hmac.compare_digest(auth_header[7:].strip(), _env_token):
                 return await call_next(request)
+        # An OCR engine attached to the fan-out gateway. It is a subprocess of
+        # this install with no account and no API token; the credential it was
+        # given is a lease, minted per attachment and held only in memory, and it
+        # authorises nothing but its own gateway path.
+        #
+        # The lease is taken from the PATH, not the header, because the engine's
+        # probes carry no header: surya checks /health and /v1/models with a bare
+        # httpx GET (see surya/inference/backends/spawn.py) and refuses to attach
+        # at all if /health is not 200. A bearer is still accepted and still has
+        # to be a valid lease — it is what its chat requests send.
+        if path.startswith("/v1/ocr/engine/"):
+            try:
+                from codai.ocr.gateway import token_is_valid
+                lease_tok = path[len("/v1/ocr/engine/"):].split("/", 1)[0]
+                if token_is_valid(lease_tok):
+                    return await call_next(request)
+            except Exception:
+                pass
         # The shared cluster token (Settings → Cluster, or CODERAI_CLUSTER_TOKEN
         # on a capability image): what a head that discovered this install
         # presents on every request it forwards here.

codai/config.py

@@ -981,6 +981,22 @@ class OcrConfig:
     surya_model_id: str = ""           # ("model" mode) the models.json id serving Surya2
     surya_server_url: str = ""         # external OpenAI server URL (llamacpp/manual); blank = auto
 
+    # --- how a served engine reaches its model ("model" mode, both engines) --- #
+    # An OCR pipeline drives ONE server address itself, which is what keeps its
+    # images, json_schema and logprobs intact. One address cannot spread load,
+    # though: an attached engine pins to the pod it was given, and a pool that
+    # counts its own in-flight requests never sees OCR traffic at all, so it never
+    # scales. With fan-out on, the engine is handed an address of OURS that
+    # forwards each request — byte for byte, deciding nothing — to the least busy
+    # of the local server, the cluster nodes holding the model, and the pod pool.
+    # Turn it off to go back to one direct address (one pod, no growth).
+    serve_fanout: bool = True
+    # In-flight requests at which a FIXED server (local, or a cluster node) is
+    # considered busy enough to rent instead: the local-plus-spillover policy, in
+    # the one place an attached OCR engine can act on it. 0 = never rent on load
+    # alone (a pool is then used only when there is nothing else).
+    serve_spill_inflight: int = 4
+
     # --- olmOCR-2 (AllenAI, Apache-2.0) --- a document VLM: whole-page transcription in
     # reading order, equations as LaTeX, tables as HTML, no bounding boxes. Driven over an
     # OpenAI chat endpoint with olmOCR's own prompt; see codai/ocr/olmocr.py.
@@ -1615,6 +1631,8 @@ class ConfigManager:
                 "surya_model": self.config.ocr.surya_model,
                 "surya_model_id": self.config.ocr.surya_model_id,
                 "surya_server_url": self.config.ocr.surya_server_url,
+                "serve_fanout": self.config.ocr.serve_fanout,
+                "serve_spill_inflight": self.config.ocr.serve_spill_inflight,
                 "build_retry_cooldown_s": self.config.ocr.build_retry_cooldown_s,
                 "evict_drain_timeout_s": self.config.ocr.evict_drain_timeout_s,
                 "vlm_gpu_memory_utilization": self.config.ocr.vlm_gpu_memory_utilization,

codai/main.py

@@ -1529,6 +1529,13 @@ def main():
         print(f"API docs: http://{bind_host}:{bind_port}/docs")
         print(f"Admin UI: http://{bind_host}:{bind_port}/admin")
 
+    # Where this process can be reached on loopback. An OCR engine runs as a
+    # subprocess of THIS process and is handed a URL back into it (the fan-out
+    # gateway), so it has to know the port it actually bound — which for an engine
+    # is an internal port, not the configured public one.
+    os.environ["CODERAI_SELF_BASE"] = (
+        f"http{'s' if bind_https else ''}://127.0.0.1:{bind_port}")
+
     if model_manager.backend is not None:
         actual_backend = model_manager.backend_type
         if hasattr(model_manager.backend, 'force_cuda') and model_manager.backend.force_cuda:

codai/ocr/dispatch.py

+# CoderAI - OpenAI-compatible API server
+# Copyright (C) 2026 Stefy Lanza <stefy@nexlab.net>
+# GPLv3 - see the project LICENSE.
+
+"""Which server should serve THIS OCR page — asked once per request.
+
+:mod:`codai.ocr.served_model` answers "where is this model served" once and hands
+the answer to the OCR engine, which then speaks to that one address for as long as
+it lives. That is the right shape for fidelity (the engine drives the server
+directly, so nothing rewrites its images, its json_schema or its logprobs) and the
+wrong shape for load:
+
+* a RunPod pool grows when its least-loaded pod reaches
+  ``scale_up_inflight_per_pod`` in-flight requests, and in-flight is counted in
+  :meth:`RunpodBackend._acquire_endpoint`. An attached engine never goes through
+  it, so the pool saw zero load however hard OCR was running, never grew, and the
+  engine would not have used a second pod anyway — it had one URL;
+* a model served on several cluster nodes resolved to whichever node answered
+  first, for ever, however busy that node became;
+* a local model that was configured to burst to RunPod under load never did: the
+  spillover trigger lives on the proxy path the engine bypasses.
+
+So the engine is given a stable address that belongs to US, and this module picks
+the upstream per request: the least loaded of the local server, the cluster nodes
+that hold the model, and the pod pool — borrowing a pod through
+:meth:`PodPool.acquire` so the pool sees the real concurrency and grows on it.
+
+The fidelity argument survives because :mod:`codai.ocr.gateway` forwards bytes.
+The earlier losses came from coderai *parsing* a request and re-emitting it through
+the backend abstraction; a hop that copies the body verbatim decides nothing.
+"""
+
+import threading
+import time
+from typing import Callable, List, Optional
+
+from codai.ocr.base import OcrError
+
+#: How long a discovered candidate set is reused before it is rebuilt. Short,
+#: because a node coming back or a model being unloaded should take effect
+#: without restarting coderai, and discovery is a handful of local HTTP calls.
+CANDIDATES_TTL_S = 30.0
+
+#: How long a node's reported load is trusted. A node reports the load of its
+#: WHOLE install, not just our OCR traffic, so this is the only way to see a node
+#: that is busy with something else.
+NODE_LOAD_TTL_S = 5.0
+
+_lock = threading.Lock()
+_candidates: dict = {}          # model id -> (expires_at, [Candidate])
+_inflight: dict = {}            # candidate key -> our own in-flight count
+_node_load: dict = {}           # node url -> (expires_at, inflight)
+
+
+class Candidate:
+    """One place that can serve the model.
+
+    ``acquire`` exists for a pool, where choosing is not free: it picks the
+    least-loaded pod, counts the request against it (which is what makes the pool
+    grow) and may rent a new pod. Everything else is a fixed address.
+    """
+
+    def __init__(self, key: str, kind: str, base: str, api_key: str = "",
+                 priority: int = 0, acquire: Optional[Callable] = None,
+                 node_url: str = "", primary: bool = False):
+        self.key = key
+        self.kind = kind                # "local" | "node" | "pod"
+        self.base = base                # ends /v1 (empty for a lazy pool)
+        self.api_key = api_key
+        self.priority = priority        # lower wins ties
+        self.acquire = acquire
+        self.node_url = node_url
+        #: This is where the operator SAID the model runs — its own `backend`.
+        #: An explicit pin is a decision, not a hint: discovery may add places to
+        #: overflow to, and must never override the configured home.
+        self.primary = primary
+
+    def __repr__(self):                 # pragma: no cover - debugging aid
+        return f"<Candidate {self.kind} {self.key}>"
+
+
+class Lease:
+    """One chosen upstream for one request. Always released."""
+
+    def __init__(self, base: str, api_key: str, key: str, kind: str,
+                 release: Optional[Callable] = None):
+        self.base = base
+        self.api_key = api_key
+        self.key = key
+        self.kind = kind
+        self._release = release
+        self._done = False
+
+    def release(self) -> None:
+        if self._done:
+            return
+        self._done = True
+        _dec(self.key)
+        if self._release is not None:
+            try:
+                self._release()
+            except Exception:
+                pass
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *exc):
+        self.release()
+        return False
+
+
+def _inc(key: str) -> None:
+    with _lock:
+        _inflight[key] = _inflight.get(key, 0) + 1
+
+
+def _dec(key: str) -> None:
+    with _lock:
+        n = _inflight.get(key, 0) - 1
+        if n > 0:
+            _inflight[key] = n
+        else:
+            _inflight.pop(key, None)
+
+
+def inflight(key: str = "") -> int:
+    """Our in-flight count for one candidate, or the total."""
+    with _lock:
+        if key:
+            return int(_inflight.get(key, 0))
+        return sum(_inflight.values())
+
+
+def forget(model_id: str = "") -> None:
+    """Drop discovered candidates (config changed, a node came or went)."""
+    with _lock:
+        if model_id:
+            _candidates.pop(model_id.strip(), None)
+        else:
+            _candidates.clear()
+        _node_load.clear()
+
+
+# ------------------------------------------------------------------ discovery #
+def _local_candidate(model_id: str) -> Optional[Candidate]:
+    """This install's own vLLM, when it is what serves the model."""
+    try:
+        from codai.models import manager as mm
+        if mm.resolve_engine_backend(model_id) != "vllm":
+            return None
+        from codai.api import runpod_worker, vllm_worker
+        from codai.models.manager import get_active_vllm_config
+        vcfg = get_active_vllm_config()
+        path = (runpod_worker._model_path_for(model_id) or "").strip() or model_id
+        base = vllm_worker.ensure_service(vcfg, model_path=path, served_name=model_id)
+    except Exception:
+        return None
+    if not base:
+        return None
+    return Candidate(key=f"local:{model_id}", kind="local",
+                     base=base.rstrip("/") + "/v1", priority=0, primary=True)
+
+
+def _node_candidates(model_id: str) -> List[Candidate]:
+    """Every enabled cluster node whose own catalogue lists the model.
+
+    All of them, not the first one that answers: which node serves a given page
+    is decided per request, by load.
+    """
+    out: List[Candidate] = []
+    try:
+        from codai.admin.routes import config_manager
+        from codai.cluster.nodes import enabled_nodes, sync_client
+        ccfg = getattr(getattr(config_manager, "config", None), "cluster", None)
+        if ccfg is None or not getattr(ccfg, "enabled", False):
+            return out
+        token = (getattr(ccfg, "token", "") or "").strip()
+        for spec in enabled_nodes(ccfg):
+            if token and not spec.api_key:
+                spec.api_key = token
+            root = spec.url.rstrip("/")
+            base = root if root.endswith("/v1") else root + "/v1"
+            try:
+                # sync_client carries the node's token and trust settings but no
+                # base_url, so the address is spelled out.
+                with sync_client(spec) as client:
+                    reply = client.get(base + "/models")
+                    if reply.status_code != 200:
+                        continue
+                    ids = {str(m.get("id") or "")
+                           for m in (reply.json().get("data") or [])}
+            except Exception:
+                continue
+            if model_id in ids:
+                out.append(Candidate(key=f"node:{spec.name or root}", kind="node",
+                                     base=base, api_key=spec.api_key, priority=1,
+                                     node_url=root))
+    except Exception:
+        return out
+    return out
+
+
+def _pod_candidate(model_id: str) -> Optional[Candidate]:
+    """The model's RunPod pool, as a lazy candidate.
+
+    Nothing is rented here: a pool that exists is a candidate, and the renting
+    happens in :meth:`PodPool.acquire` only when a request actually picks it.
+    """
+    try:
+        from codai.models import manager as mm
+        if not mm.runpod_should_handle(model_id):
+            return None
+        from codai.backends.runpod import RunpodBackend
+        backend = RunpodBackend(mm.get_active_runpod_config())
+        backend.load_model(model_id)
+    except Exception:
+        return None
+
+    def _acquire():
+        base, release = backend._acquire_endpoint()
+        auth = (backend._headers or {}).get("Authorization", "")
+        key = auth[7:].strip() if auth[:7].lower() == "bearer " else ""
+        return base, key, release
+
+    # `backend: runpod` on the entry means the pool IS this model's home, so it
+    # comes before a node that happens to list the same id.
+    return Candidate(key=f"pod:{model_id}", kind="pod", base="", priority=2,
+                     acquire=_acquire, primary=True)
+
+
+def _spill_pod_candidate(model_id: str) -> Optional[Candidate]:
+    """A LOCAL model's burst-to-RunPod target, as a pod candidate.
+
+    ``runpod_spillover`` on a model entry is honoured by the front proxy for
+    requests passing through it. An attached OCR engine bypasses the proxy, so the
+    same intent is read here: the operator said "when the local one is busy, rent".
+    """
+    try:
+        from codai.admin.routes import config_manager
+        from codai.api.runpod_worker import (get_model_pod_pool, parse_model_runpod)
+        md = getattr(config_manager, "models_data", {}) or {}
+        entry = None
+        for lst in md.values():
+            if not isinstance(lst, list):
+                continue
+            for m in lst:
+                if isinstance(m, dict) and str(m.get("id") or "") == model_id:
+                    entry = m
+                    break
+        if entry is None:
+            return None
+        sp = entry.get("runpod_spillover")
+        if not isinstance(sp, dict) or not sp.get("enabled"):
+            return None
+        # on_busy is exactly this situation. A spillover configured only for
+        # on_no_gpu / on_local_error is not a load policy, so it is not used here.
+        if not (sp.get("on_busy") or sp.get("on_concurrency_full")):
+            return None
+        target = sp.get("target") if isinstance(sp.get("target"), dict) else {}
+        if (target.get("mode") or "serverless").lower() != "pods":
+            return None
+        rp = getattr(getattr(config_manager, "config", None), "runpod", None)
+        if rp is None or not getattr(rp, "enabled", False):
+            return None
+        mcfg = parse_model_runpod(target)
+        pool = get_model_pod_pool(model_id, entry, target)
+    except Exception:
+        return None
+    if pool is None:
+        return None
+
+    def _acquire():
+        pod, url = pool.acquire()
+        key = getattr(pool, "api_key", "") or ""
+        return (url.rstrip("/") + "/v1", key, lambda: pool.release(pod))
+
+    return Candidate(key=f"spill:{model_id}", kind="pod", base="", priority=3,
+                     acquire=_acquire)
+
+
+def candidates(model_id: str) -> List[Candidate]:
+    """Everywhere this model can be served right now, cheapest hop first."""
+    model_id = (model_id or "").strip()
+    now = time.time()
+    with _lock:
+        hit = _candidates.get(model_id)
+        if hit and hit[0] > now:
+            return list(hit[1])
+
+    found: List[Candidate] = []
+    local = _local_candidate(model_id)
+    if local is not None:
+        found.append(local)
+    found.extend(_node_candidates(model_id))
+    pod = _pod_candidate(model_id)
+    if pod is not None:
+        found.append(pod)
+    spill = _spill_pod_candidate(model_id)
+    if spill is not None:
+        found.append(spill)
+
+    with _lock:
+        _candidates[model_id] = (now + CANDIDATES_TTL_S, list(found))
+    return found
+
+
+# -------------------------------------------------------------------- picking #
+def _node_reported_load(cand: Candidate) -> int:
+    """What a node says its whole install is doing, cached briefly.
+
+    A node serves more than our OCR pages; counting only what we sent it would
+    route onto a node that is saturated with someone else's work.
+    """
+    if cand.kind != "node" or not cand.node_url:
+        return 0
+    now = time.time()
+    with _lock:
+        hit = _node_load.get(cand.node_url)
+        if hit and hit[0] > now:
+            return int(hit[1])
+    load = 0
+    try:
+        from codai.admin.routes import config_manager
+        from codai.cluster.nodes import enabled_nodes, sync_client
+        ccfg = getattr(getattr(config_manager, "config", None), "cluster", None)
+        for spec in enabled_nodes(ccfg):
+            if spec.url.rstrip("/") != cand.node_url:
+                continue
+            with sync_client(spec) as client:
+                reply = client.get(cand.node_url + "/cluster/state")
+                if reply.status_code == 200:
+                    load = int(reply.json().get("inflight") or 0)
+            break
+    except Exception:
+        load = 0
+    with _lock:
+        _node_load[cand.node_url] = (now + NODE_LOAD_TTL_S, load)
+    return load
+
+
+def _load_of(cand: Candidate) -> int:
+    return inflight(cand.key) + _node_reported_load(cand)
+
+
+def choose(model_id: str, spill_after: int = 0) -> Lease:
+    """A lease on the least-loaded server for the model.
+
+    ``spill_after`` is the in-flight count at which a fixed server (the local one,
+    a node) is considered busy enough to prefer renting instead — the
+    local-plus-spillover policy, expressed where OCR can act on it. 0 means never
+    spill on load alone; a pool is then used only when it is all there is.
+    """
+    cands = candidates(model_id)
+    if not cands:
+        raise OcrError(
+            f"model {model_id!r} is not served over HTTP by this install, so an OCR "
+            f"engine cannot attach to it. Pin the model to a backend that has an "
+            f"endpoint (backend: \"runpod\", a cluster node that holds it, or a "
+            f"service_url), or pick a serve mode that fits this box.",
+            status=503)
+
+    last_exc = None
+
+    # 1. A pool that is this model's configured home. `backend: runpod` means
+    #    "run it on RunPod", so that is where it runs — and because every request
+    #    goes through acquire(), the pool sees the real concurrency, spreads the
+    #    pages over the pods it has and grows when they are all at their
+    #    threshold. This is "on runpod only", scaling.
+    for cand in [c for c in cands if c.primary and c.acquire is not None]:
+        lease = _take(cand)
+        if lease is not None:
+            return lease
+        last_exc = "pool unavailable"
+
+    # 2. Fixed servers — this machine and the cluster nodes that hold the model —
+    #    compete on load, so a head with two nodes sends each page to whichever is
+    #    least busy, counting what the node itself reports so a node loaded with
+    #    someone else's work is left alone.
+    fixed = sorted((c for c in cands if c.acquire is None),
+                   key=lambda c: (_load_of(c), 0 if c.primary else 1, c.priority))
+    pools = sorted((c for c in cands if c.acquire is not None),
+                   key=lambda c: (c.priority, _load_of(c)))
+    if fixed:
+        best = fixed[0]
+        if not pools or spill_after <= 0 or _load_of(best) < spill_after:
+            _inc(best.key)
+            return Lease(best.base, best.api_key, best.key, best.kind)
+
+    # 3. Everything we own is busy: rent. This is "local + spillover" — the
+    #    overflow goes to a pool instead of queueing behind a saturated GPU.
+    for cand in pools:
+        lease = _take(cand)
+        if lease is not None:
+            return lease
+        last_exc = "pool unavailable"
+
+    if fixed:
+        # Busy, and nothing could be rented (no budget, no capacity): the busy
+        # server still serves, queued, rather than failing the page.
+        best = fixed[0]
+        _inc(best.key)
+        return Lease(best.base, best.api_key, best.key, best.kind)
+
+    raise OcrError(f"model {model_id!r}: no OCR server could be reached "
+                   f"({last_exc})", status=503)
+
+
+def _take(cand: Candidate) -> Optional[Lease]:
+    """Acquire a pool candidate, or None when it cannot provide a server."""
+    try:
+        base, key, release = cand.acquire()
+    except Exception as exc:
+        print(f"[ocr] {cand.key} cannot serve this page: {exc}", flush=True)
+        return None
+    _inc(cand.key)
+    return Lease(base, key, cand.key, cand.kind, release)
+
+
+def any_candidate(model_id: str) -> bool:
+    """Whether anything at all can serve the model — without renting a thing."""
+    try:
+        return bool(candidates(model_id))
+    except Exception:
+        return False

codai/ocr/gateway.py

+# CoderAI - OpenAI-compatible API server
+# Copyright (C) 2026 Stefy Lanza <stefy@nexlab.net>
+# GPLv3 - see the project LICENSE.
+
+"""One stable OpenAI address for an OCR engine, fanned out per request.
+
+An OCR pipeline (surya-2, olmOCR) is given ONE server address and drives it
+itself. That is deliberate: the engine sends multipart images, a strict
+``json_schema`` and ``logprobs: true``, and every one of those was quietly lost
+the last time OCR went through coderai's backend abstraction, which parses a
+request and re-emits it.
+
+This module keeps the engine's single address while letting the work spread, by
+being a hop that *decides nothing about the request*: the body goes upstream as
+the bytes that arrived, and the reply comes back as the bytes that were sent. The
+only decision is WHERE, taken per request by :mod:`codai.ocr.dispatch`, which can
+pick the local server, the least-busy cluster node, or a pod from the pool —
+counting the request against it so the pool grows under real load.
+
+The three routes are exactly what surya probes on attach (see
+``surya/inference/backends/spawn.py``):
+
+* ``GET  <root>/health``                 — 200, or it refuses to attach at all;
+* ``GET  <root>/v1/models``              — ``data[0].id`` must EQUAL the engine's
+  configured checkpoint, or it raises a model-mismatch error;
+* ``POST <root>/v1/chat/completions``    — the work.
+
+The lease in the path is a secret generated per engine attachment, and it is also
+the bearer token the engine is given, so the route is not reachable by anything
+that was not handed both.
+"""
+
+import os
+import secrets
+import threading
+from typing import Optional
+
+from fastapi import APIRouter, Request
+from fastapi.responses import JSONResponse, Response
+
+router = APIRouter()
+
+#: Path the engine attaches under. The OpenAI base it is given is
+#: ``<self>/v1/ocr/engine/<lease>/v1`` so that stripping ``/v1`` — which is how
+#: surya derives the health URL — lands on ``<self>/v1/ocr/engine/<lease>``.
+PREFIX = "/v1/ocr/engine"
+
+#: How long one page may take upstream. A dense scan on a cold pod legitimately
+#: takes minutes; a timeout here would look to the engine like a failed page.
+UPSTREAM_TIMEOUT_S = 3600.0
+
+_lock = threading.Lock()
+_leases: dict = {}              # lease token -> {"model_id", "served_name"}
+
+
+def issue(model_id: str, served_name: str) -> str:
+    """Mint a lease for one engine attachment and return its token."""
+    token = secrets.token_urlsafe(24)
+    with _lock:
+        _leases[token] = {"model_id": model_id, "served_name": served_name}
+    return token
+
+
+def revoke(model_id: str = "") -> None:
+    with _lock:
+        if not model_id:
+            _leases.clear()
+            return
+        for tok in [t for t, v in _leases.items() if v["model_id"] == model_id]:
+            _leases.pop(tok, None)
+
+
+def lease(token: str) -> Optional[dict]:
+    with _lock:
+        got = _leases.get((token or "").strip())
+        return dict(got) if got else None
+
+
+def token_is_valid(token: str) -> bool:
+    """Whether a bearer token is a live OCR lease.
+
+    The auth middleware asks this: an attached engine has no account and no API
+    token, and the lease is the credential it was given.
+    """
+    return lease(token) is not None
+
+
+def self_base() -> str:
+    """This install's own base URL, for an address we hand to a local engine.
+
+    Set at bind time (``CODERAI_SELF_BASE``). The engine runs as a subprocess on
+    this machine, so loopback is both correct and the only thing that needs no
+    certificate.
+    """
+    base = (os.environ.get("CODERAI_SELF_BASE") or "").strip().rstrip("/")
+    if base:
+        return base
+    port = (os.environ.get("CODERAI_PORT") or "").strip()
+    return f"http://127.0.0.1:{port or '8000'}"
+
+
+def engine_url(token: str) -> str:
+    """The OpenAI base an engine is given for this lease (ends ``/v1``)."""
+    return f"{self_base()}{PREFIX}/{token}/v1"
+
+
+# --------------------------------------------------------------------- routes #
+def _unknown_lease() -> JSONResponse:
+    return JSONResponse(status_code=404,
+                        content={"error": {"message": "unknown OCR engine lease",
+                                           "type": "invalid_request_error"}})
+
+
+@router.get(PREFIX + "/{token}/health", include_in_schema=False)
+async def health(token: str):
+    """200 while something can serve the model — without renting anything.
+
+    Surya refuses to attach to a URL whose ``/health`` is not 200, so this must
+    answer before any pod exists; a scale-to-zero pool is healthy precisely
+    because it can rent on demand.
+    """
+    got = lease(token)
+    if got is None:
+        return _unknown_lease()
+    from codai.ocr import dispatch
+    if not dispatch.any_candidate(got["model_id"]):
+        return JSONResponse(status_code=503,
+                            content={"error": {"message": "no server for "
+                                                          + got["model_id"]}})
+    return Response(status_code=200, content=b"OK", media_type="text/plain")
+
+
+@router.get(PREFIX + "/{token}/v1/models", include_in_schema=False)
+async def models(token: str):
+    """The id the engine expects, which is also what we will send upstream.
+
+    Surya compares ``data[0].id`` with its configured checkpoint and refuses on a
+    mismatch. Answering from the lease — rather than relaying whatever one
+    upstream happens to call it — is also what makes several upstreams
+    interchangeable to the engine.
+    """
+    got = lease(token)
+    if got is None:
+        return _unknown_lease()
+    return JSONResponse(content={
+        "object": "list",
+        "data": [{"id": got["served_name"], "object": "model",
+                  "owned_by": "coderai"}],
+    })
+
+
+@router.post(PREFIX + "/{token}/v1/chat/completions", include_in_schema=False)
+async def chat_completions(token: str, request: Request):
+    """Forward the request verbatim to whichever server is least busy.
+
+    Verbatim is the whole point. The body is not parsed, so the multipart image
+    stays an image, the ``json_schema`` stays strict, and ``logprobs`` is asked
+    for and answered — the three things a request-editing proxy dropped. The one
+    field that is rewritten is ``model``, and only when an upstream serves it
+    under another name.
+    """
+    got = lease(token)
+    if got is None:
+        return _unknown_lease()
+
+    import httpx
+
+    from codai.admin.routes import config_manager
+    from codai.ocr import dispatch
+    body = await request.body()
+    ocfg = getattr(getattr(config_manager, "config", None), "ocr", None)
+    spill_after = int(getattr(ocfg, "serve_spill_inflight", 0) or 0)
+
+    try:
+        chosen = dispatch.choose(got["model_id"], spill_after=spill_after)
+    except Exception as exc:
+        status = int(getattr(exc, "status", 503) or 503)
+        return JSONResponse(status_code=status,
+                            content={"error": {"message": str(exc),
+                                               "type": "server_error"}})
+
+    try:
+        headers = {"Content-Type": request.headers.get("content-type",
+                                                       "application/json")}
+        if chosen.api_key:
+            headers["Authorization"] = f"Bearer {chosen.api_key}"
+        url = chosen.base.rstrip("/") + "/chat/completions"
+        async with httpx.AsyncClient(timeout=UPSTREAM_TIMEOUT_S) as client:
+            reply = await client.post(url, content=body, headers=headers)
+        # The reply goes back as it came: a re-serialised response is how
+        # logprobs got lost in the first place.
+        media = reply.headers.get("content-type", "application/json")
+        return Response(status_code=reply.status_code, content=reply.content,
+                        media_type=media)
+    except Exception as exc:
+        return JSONResponse(status_code=502,
+                            content={"error": {
+                                "message": f"OCR upstream {chosen.kind} failed: {exc}",
+                                "type": "server_error"}})
+    finally:
+        chosen.release()

codai/ocr/olmocr.py

@@ -337,9 +337,36 @@ class OlmOcrEngine(OcrEngine):
         temperature = float(getattr(self.cfg, "olmocr_temperature", 0.1) or 0.0)
 
         if self._mode() == "model":
-            return self._ask_local(messages, max_tokens, temperature)
+            return self._ask_model(messages, max_tokens, temperature)
         return self._ask_http(messages, max_tokens, temperature)
 
+    def _ask_model(self, messages, max_tokens, temperature) -> str:
+        """Model mode: ask the dispatcher which server should take this page.
+
+        The in-process path below is the right one for a model this machine
+        serves — it goes through the model manager, so loading, VRAM eviction and
+        queueing all apply. It is the WRONG one for a model served elsewhere: it
+        cannot see which cluster node is least busy (nodes live in the front's
+        registry, not here), and it bypasses the front, where burst-to-RunPod
+        lives. So the choice is made per page, and only a local answer takes the
+        in-process route.
+        """
+        model_id = (getattr(self.cfg, "olmocr_model_id", "") or "").strip()
+        try:
+            from codai.ocr import dispatch
+            spill = int(getattr(self.cfg, "serve_spill_inflight", 0) or 0)
+            lease = dispatch.choose(model_id, spill_after=spill)
+        except Exception:
+            return self._ask_local(messages, max_tokens, temperature)
+        try:
+            if lease.kind == "local":
+                return self._ask_local(messages, max_tokens, temperature)
+            return self._ask_http(messages, max_tokens, temperature,
+                                  base=lease.base, api_key=lease.api_key,
+                                  model_name=model_id)
+        finally:
+            lease.release()
+
     def _ask_local(self, messages, max_tokens, temperature) -> str:
         """Through coderai's own in-process chat handler (model mode)."""
         import asyncio
@@ -365,22 +392,27 @@ class OlmOcrEngine(OcrEngine):
             raise OcrError(f"olmOCR model '{model_id}' returned no text", status=500)
         return content
 
-    def _ask_http(self, messages, max_tokens, temperature) -> str:
-        """Against an OpenAI-compatible server (vllm/server modes)."""
+    def _ask_http(self, messages, max_tokens, temperature, base: str = "",
+                  api_key: str = "", model_name: str = "") -> str:
+        """Against an OpenAI-compatible server (vllm/server modes).
+
+        ``base``/``api_key``/``model_name`` override the configured server for one
+        page, which is how model mode reaches the upstream the dispatcher chose.
+        """
         import requests
 
         payload = {
-            "model": self._model_name(),
+            "model": model_name or self._model_name(),
             "messages": messages,
             "max_tokens": max_tokens,
             "temperature": temperature,
             "stream": False,
         }
         headers = {"Content-Type": "application/json"}
-        key = (getattr(self.cfg, "olmocr_api_key", "") or "").strip()
+        key = api_key or (getattr(self.cfg, "olmocr_api_key", "") or "").strip()
         if key:
             headers["Authorization"] = f"Bearer {key}"
-        url = self._base.rstrip("/") + "/chat/completions"
+        url = (base or self._base).rstrip("/") + "/chat/completions"
         try:
             r = requests.post(url, json=payload, headers=headers,
                               timeout=float(getattr(self.cfg, "olmocr_timeout", 300.0) or 300.0))

codai/ocr/served_model.py

@@ -26,6 +26,15 @@ scaling, the budget ledger and the idle reaper stay where they belong.
 
 The endpoint is a lease, not an address: a scale-to-zero pod is reaped when OCR goes
 quiet, and the caller is expected to re-resolve rather than to trust it forever.
+
+One address, however, cannot spread load: an engine handed a pod's URL keeps using
+that pod however busy it gets, and the pool — which grows on in-flight requests it
+counts itself — never learns there is load at all. So unless the install turns it
+off (``ocr.serve_fanout``), what the engine is handed is a lease on OUR OpenAI
+address (:mod:`codai.ocr.gateway`), which forwards each request verbatim to
+whichever server is least busy (:mod:`codai.ocr.dispatch`). The direct address
+remains the behaviour with fan-out disabled, and is still what the gateway resolves
+to when there is exactly one place to go.
 """
 
 import threading
@@ -162,6 +171,30 @@ def _served_name_of(backend, model_id: str) -> str:
     return model_id
 
 
+def _fanout_enabled() -> bool:
+    try:
+        from codai.admin.routes import config_manager
+        ocfg = getattr(getattr(config_manager, "config", None), "ocr", None)
+        return bool(getattr(ocfg, "serve_fanout", True))
+    except Exception:
+        return True
+
+
+def _fanout_endpoint(model_id: str):
+    """``(gateway base, lease token, served name)``, or None if fan-out cannot be used.
+
+    The served name is resolved the same way as for a direct attachment, because the
+    engine checks it against its own checkpoint before sending a single page.
+    """
+    from codai.ocr import dispatch, gateway
+    cands = dispatch.candidates(model_id)
+    if not cands:
+        return None
+    served = model_id
+    token = gateway.issue(model_id, served)
+    return gateway.engine_url(token), token, served
+
+
 def endpoint_for(model_id: str):
     """``(base_url ending /v1, api_key, served_name)`` for ``model_id``.
 
@@ -173,6 +206,14 @@ def endpoint_for(model_id: str):
     if not model_id:
         raise OcrError("no model id configured for the served OCR engine", status=400)
 
+    # One address of ours, many servers behind it. Discovery happens now (so a
+    # misconfiguration is still a clear error at load time) but nothing is rented:
+    # a pod is picked, and paid for, per request.
+    if _fanout_enabled():
+        fan = _fanout_endpoint(model_id)
+        if fan is not None:
+            return fan
+
     with _lock:
         backend = _backends.get(model_id)
     if backend is None:

codai/ocr/surya.py

@@ -190,8 +190,12 @@ class SuryaEngine(SubprocessOcrEngine):
                 raise
             print(f"[ocr] surya endpoint {self._server_url} is gone "
                   f"({first}) — re-resolving", flush=True)
-            from codai.ocr import served_model
+            from codai.ocr import dispatch, served_model
             served_model.forget()
+            # The candidate set is discovered, so a vanished pod or node must be
+            # re-discovered too; otherwise the re-resolve hands back the same
+            # dead address it just rejected.
+            dispatch.forget()
             self.cleanup()
             self.load()
             self._loaded = True

docs/ocr.md

@@ -293,6 +293,42 @@ orchestrator: surya attaches straight to the endpoint the pool hands it. A pod r
 only the code baked into its image, so an install whose pods predate this needs its
 capability image rebuilt before the numbers appear.
 
+### Scaling in `model` mode
+
+An OCR pipeline drives one server address itself — that is what keeps the page
+images, the strict schema and the confidence scores intact. One address cannot
+spread load, though, and three things used to follow from that:
+
+* a RunPod pool grows when its least-loaded pod reaches
+  `scale_up_inflight_per_pod` **in-flight** requests, and in-flight is counted
+  when a request passes through the pool. An attached engine never did, so the
+  pool saw no load however hard OCR ran — and the engine had one pod's URL anyway;
+* several cluster nodes serving the model resolved to whichever answered first,
+  for ever, however busy it became;
+* a local model set to burst to RunPod never did: that trigger lives on the proxy
+  path an attached engine bypasses.
+
+With **`ocr.serve_fanout`** on (the default) the engine is instead handed one
+address of coderai's own, and each request is forwarded — **verbatim**, body bytes
+in and body bytes out, so nothing is decided about it — to whichever server is
+least busy:
+
+| the model is… | a page goes to |
+|---|---|
+| `backend: runpod` | the pool, every request: it picks the least-loaded pod and rents another when they are all at the threshold |
+| served on this machine | the local server, until it is busier than `ocr.serve_spill_inflight`, then a pod (the burst `runpod_spillover` describes) |
+| held by cluster nodes | the least busy of them, counting what each node reports for its **whole** install, so a node loaded with other work is left alone |
+| several of these at once | its configured home first — an explicit `backend` is a decision, not a hint — then the rest as overflow |
+
+`ocr.serve_spill_inflight: 0` means "never rent because of load": a pool is then
+used only when there is nothing else. Turning `serve_fanout` off restores the
+single direct address (one pod, no growth), which is one hop fewer and loses
+nothing on an install with exactly one server.
+
+The address is a lease: a secret minted per attachment, carried in the URL (the
+engine's `/health` and `/v1/models` probes send no headers, so it cannot be a
+bearer alone) and valid for nothing but that one gateway path.
+
 ```bash
 curl -F file=@sentenza.pdf -F structured=true -F detect=both -F schema=italian_sentenza \
      http://localhost:8776/v1/ocr

tests/test_ocr_fanout.py

+"""OCR has to scale like everything else, without losing what made it accurate.
+
+An OCR pipeline is handed ONE server address and drives it itself, because that is
+what keeps its page images, its strict json_schema and its logprobs intact — a hop
+that parses and re-emits the request lost all three, twice. But one address cannot
+spread load:
+
+  * a RunPod pool grows when its least-loaded pod reaches
+    `scale_up_inflight_per_pod` IN-FLIGHT requests, counted in
+    `RunpodBackend._acquire_endpoint`. `openai_endpoint()` releases the slot at
+    once (by design — an attachment must not pin a scaling slot), so the pool saw
+    zero load however hard OCR ran, and the engine had only one pod's URL anyway;
+  * several cluster nodes serving the model resolved to whichever answered first,
+    for ever, however busy it got;
+  * a local model configured to burst to RunPod never did: that trigger lives on
+    the proxy path an attached engine bypasses.
+
+So the engine now gets a lease on an address of ours that forwards each request
+verbatim to the least busy server. These tests pin both halves: that the choosing
+is load-aware and makes the pool grow, and that the forwarding still decides
+nothing about the request.
+"""
+
+import pathlib
+import sys
+
+ROOT = pathlib.Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT))
+
+from codai.ocr import dispatch, gateway                            # noqa: E402
+from codai.ocr.base import OcrError                                # noqa: E402
+
+
+def _reset():
+    dispatch.forget()
+    gateway.revoke()
+
+
+def _cand(key, kind="node", base="http://n/v1", priority=1, node_url="",
+          acquire=None):
+    return dispatch.Candidate(key=key, kind=kind, base=base, priority=priority,
+                              node_url=node_url, acquire=acquire)
+
+
+def _fixed_candidates(monkeypatch, cands):
+    monkeypatch.setattr(dispatch, "candidates", lambda m: list(cands))
+
+
+def _no_node_load(monkeypatch):
+    """Node-reported load is a network call; default it to quiet in unit tests."""
+    monkeypatch.setattr(dispatch, "_node_reported_load", lambda c: 0)
+
+
+# ------------------------------------------------------------ least busy first #
+def test_the_least_busy_node_gets_the_page(monkeypatch):
+    _reset()
+    _no_node_load(monkeypatch)
+    a, b = _cand("node:a", base="http://a/v1"), _cand("node:b", base="http://b/v1")
+    _fixed_candidates(monkeypatch, [a, b])
+    # Put two requests on A; B is idle.
+    held = [dispatch.choose("m"), dispatch.choose("m")]
+    assert {h.key for h in held} == {"node:a", "node:b"}       # one each, then
+    third = dispatch.choose("m")
+    assert dispatch.inflight(third.key) >= 1
+    for h in held + [third]:
+        h.release()
+    assert dispatch.inflight() == 0
+
+
+def test_releasing_frees_the_slot(monkeypatch):
+    _reset()
+    _no_node_load(monkeypatch)
+    only = _cand("node:a")
+    _fixed_candidates(monkeypatch, [only])
+    lease = dispatch.choose("m")
+    assert dispatch.inflight("node:a") == 1
+    lease.release()
+    assert dispatch.inflight("node:a") == 0
+    lease.release()                                            # idempotent
+    assert dispatch.inflight("node:a") == 0
+
+
+def test_a_node_busy_with_someone_elses_work_is_avoided(monkeypatch):
+    """A node serves more than our OCR pages. Counting only what WE sent it would
+    route onto a node already saturated by its own traffic."""
+    _reset()
+    quiet = _cand("node:quiet", base="http://quiet/v1", node_url="http://quiet")
+    busy = _cand("node:busy", base="http://busy/v1", node_url="http://busy")
+    _fixed_candidates(monkeypatch, [busy, quiet])
+    monkeypatch.setattr(dispatch, "_node_reported_load",
+                        lambda c: 9 if c.key == "node:busy" else 0)
+    lease = dispatch.choose("m")
+    assert lease.base == "http://quiet/v1"
+    lease.release()
+
+
+def test_a_local_server_is_preferred_while_it_is_not_busy(monkeypatch):
+    """It is the cheapest hop and already paid for."""
+    _reset()
+    _no_node_load(monkeypatch)
+    local = _cand("local:m", kind="local", base="http://127.0.0.1:8000/v1", priority=0)
+    node = _cand("node:a")
+    _fixed_candidates(monkeypatch, [node, local])
+    lease = dispatch.choose("m", spill_after=4)
+    assert lease.kind == "local"
+    lease.release()
+
+
+# ----------------------------------------------------------------- pods & growth #
+class _FakePool:
+    """Stands in for PodPool: records acquire/release, which is what growth rides on."""
+
+    def __init__(self):
+        self.acquired = 0
+        self.released = 0
+        self.api_key = "podkey"
+
+    def acquire(self):
+        self.acquired += 1
+        return (f"pod{self.acquired}", "http://pod.runpod/")
+
+    def release(self, pod):
+        self.released += 1
+
+
+def _pod_cand(pool, key="pod:m", priority=2):
+    def _acquire():
+        pod, url = pool.acquire()
+        return url.rstrip("/") + "/v1", pool.api_key, lambda: pool.release(pod)
+    return dispatch.Candidate(key=key, kind="pod", base="", priority=priority,
+                              acquire=_acquire)
+
+
+def test_a_runpod_only_model_goes_through_the_pool_every_request(monkeypatch):
+    """This is what makes the pool scale: it counts in-flight in acquire(), so the
+    request has to pass through it rather than reuse a remembered URL."""
+    _reset()
+    pool = _FakePool()
+    _fixed_candidates(monkeypatch, [_pod_cand(pool)])
+    for _ in range(3):
+        dispatch.choose("m").release()
+    assert pool.acquired == 3
+    assert pool.released == 3
+
+
+def test_the_pod_slot_is_returned_even_when_the_caller_forgets_nothing(monkeypatch):
+    _reset()
+    pool = _FakePool()
+    _fixed_candidates(monkeypatch, [_pod_cand(pool)])
+    with dispatch.choose("m"):
+        assert pool.released == 0
+    assert pool.released == 1
+
+
+def test_a_busy_local_server_spills_to_a_pod(monkeypatch):
+    """local + spillover: the local server keeps the first requests, and the
+    overflow rents instead of queueing behind them."""
+    _reset()
+    _no_node_load(monkeypatch)
+    local = _cand("local:m", kind="local", base="http://127.0.0.1:8000/v1", priority=0)
+    pool = _FakePool()
+    _fixed_candidates(monkeypatch, [local, _pod_cand(pool)])
+    held = [dispatch.choose("m", spill_after=2) for _ in range(2)]
+    assert [h.kind for h in held] == ["local", "local"]
+    overflow = dispatch.choose("m", spill_after=2)
+    assert overflow.kind == "pod"
+    assert pool.acquired == 1
+    for h in held + [overflow]:
+        h.release()
+
+
+def test_zero_threshold_never_spills_on_load_alone(monkeypatch):
+    """0 means "do not rent because of load" — an install that wants its own GPU
+    used and nothing billed must be able to say so."""
+    _reset()
+    _no_node_load(monkeypatch)
+    local = _cand("local:m", kind="local", base="http://l/v1", priority=0)
+    pool = _FakePool()
+    _fixed_candidates(monkeypatch, [local, _pod_cand(pool)])
+    held = [dispatch.choose("m", spill_after=0) for _ in range(5)]
+    assert {h.kind for h in held} == {"local"}
+    assert pool.acquired == 0
+    for h in held:
+        h.release()
+
+
+def test_a_pool_is_still_used_when_it_is_all_there_is(monkeypatch):
+    _reset()
+    pool = _FakePool()
+    _fixed_candidates(monkeypatch, [_pod_cand(pool)])
+    lease = dispatch.choose("m", spill_after=0)
+    assert lease.kind == "pod"
+    lease.release()
+
+
+def test_an_ordinary_pod_pool_is_tried_before_a_spillover_one(monkeypatch):
+    """Spillover is the burst of last resort: it exists because the first choice
+    was busy."""
+    _reset()
+    _no_node_load(monkeypatch)
+    main, spill = _FakePool(), _FakePool()
+    _fixed_candidates(monkeypatch, [_pod_cand(spill, key="spill:m", priority=3),
+                                    _pod_cand(main, key="pod:m", priority=2)])
+    dispatch.choose("m").release()
+    assert (main.acquired, spill.acquired) == (1, 0)
+
+
+def test_a_pool_that_cannot_rent_falls_through_to_the_next(monkeypatch):
+    _reset()
+    _no_node_load(monkeypatch)
+
+    def _broken():
+        raise RuntimeError("no capacity in region")
+    broken = dispatch.Candidate(key="pod:m", kind="pod", base="", priority=2,
+                                acquire=_broken)
+    good = _FakePool()
+    _fixed_candidates(monkeypatch, [broken, _pod_cand(good, key="spill:m", priority=3)])
+    lease = dispatch.choose("m")
+    assert good.acquired == 1
+    lease.release()
+
+
+def test_a_busy_fixed_server_still_serves_when_nothing_can_be_rented(monkeypatch):
+    """Queued on a busy GPU beats failing the page."""
+    _reset()
+    _no_node_load(monkeypatch)
+
+    def _broken():
+        raise RuntimeError("budget exhausted")
+    local = _cand("local:m", kind="local", base="http://l/v1", priority=0)
+    _fixed_candidates(monkeypatch, [
+        local, dispatch.Candidate(key="pod:m", kind="pod", base="", priority=2,
+                                  acquire=_broken)])
+    held = [dispatch.choose("m", spill_after=1) for _ in range(2)]
+    assert [h.kind for h in held] == ["local", "local"]
+    for h in held:
+        h.release()
+
+
+def test_nothing_anywhere_is_a_clear_503(monkeypatch):
+    _reset()
+    _fixed_candidates(monkeypatch, [])
+    try:
+        dispatch.choose("datalab-to/surya-ocr-2")
+    except OcrError as exc:
+        assert exc.status == 503
+        assert "not served over HTTP" in str(exc)
+    else:
+        raise AssertionError("expected an OcrError")
+
+
+# ------------------------------------------------------------------- the lease #
+def test_the_engine_url_is_what_surya_can_attach_to(monkeypatch):
+    """Surya derives the health URL by stripping a trailing /v1, so the address it
+    is given must end in /v1 and leave a working root behind."""
+    _reset()
+    monkeypatch.setenv("CODERAI_SELF_BASE", "http://127.0.0.1:8777")
+    token = gateway.issue("m", "m")
+    url = gateway.engine_url(token)
+    assert url == f"http://127.0.0.1:8777/v1/ocr/engine/{token}/v1"
+    assert url.endswith("/v1")
+    assert url[: -len("/v1")] == f"http://127.0.0.1:8777/v1/ocr/engine/{token}"
+
+
+def test_a_lease_is_a_credential_and_nothing_else():
+    _reset()
+    token = gateway.issue("m", "m")
+    assert gateway.token_is_valid(token)
+    assert not gateway.token_is_valid("guessed")
+    assert not gateway.token_is_valid("")
+    gateway.revoke("m")
+    assert not gateway.token_is_valid(token)
+
+
+def test_two_engines_get_two_leases():
+    _reset()
+    a = gateway.issue("surya", "surya")
+    b = gateway.issue("olmocr", "olmocr")
+    assert a != b
+    assert gateway.lease(a)["model_id"] == "surya"
+    assert gateway.lease(b)["model_id"] == "olmocr"
+
+
+def test_revoking_one_model_leaves_the_other():
+    _reset()
+    a = gateway.issue("surya", "surya")
+    b = gateway.issue("olmocr", "olmocr")
+    gateway.revoke("surya")
+    assert not gateway.token_is_valid(a)
+    assert gateway.token_is_valid(b)
+
+
+# ------------------------------------------------------- the resolver uses it #
+def test_model_mode_hands_the_engine_the_gateway(monkeypatch):
+    _reset()
+    monkeypatch.setenv("CODERAI_SELF_BASE", "http://127.0.0.1:8777")
+    from codai.ocr import served_model
+    monkeypatch.setattr(served_model, "_fanout_enabled", lambda: True)
+    _fixed_candidates(monkeypatch, [_cand("node:a")])
+    base, key, served = served_model.endpoint_for("datalab-to/surya-ocr-2")
+    assert "/v1/ocr/engine/" in base and base.endswith("/v1")
+    assert gateway.token_is_valid(key)
+    assert served == "datalab-to/surya-ocr-2"
+
+
+def test_the_served_name_is_what_the_engine_expects(monkeypatch):
+    """probe_model_id compares data[0].id with SURYA_MODEL_CHECKPOINT and refuses
+    on a mismatch, so the gateway answers with the name the engine was told —
+    which is also what makes several upstreams interchangeable to it."""
+    _reset()
+    token = gateway.issue("datalab-to/surya-ocr-2", "datalab-to/surya-ocr-2")
+    assert gateway.lease(token)["served_name"] == "datalab-to/surya-ocr-2"
+
+
+def test_fanout_off_goes_back_to_one_direct_address(monkeypatch):
+    """The direct attachment stays available: it is one hop fewer, and an install
+    with a single server loses nothing by it."""
+    _reset()
+    from codai.ocr import served_model
+    monkeypatch.setattr(served_model, "_fanout_enabled", lambda: False)
+
+    class _B:
+        def openai_endpoint(self):
+            return "http://pod/v1", "k"
+        served_name = "m"
+    monkeypatch.setattr(served_model, "_entry_backend", lambda m: _B())
+    base, key, served = served_model.endpoint_for("m")
+    assert base == "http://pod/v1" and key == "k"
+
+
+def test_a_blank_model_id_is_still_a_400(monkeypatch):
+    _reset()
+    from codai.ocr import served_model
+    try:
+        served_model.endpoint_for("   ")
+    except OcrError as exc:
+        assert exc.status == 400
+    else:
+        raise AssertionError("expected an OcrError")
+
+
+# --------------------------------------------------------------- health probe #
+def test_health_is_200_before_any_pod_exists(monkeypatch):
+    """A scale-to-zero pool is HEALTHY: it can rent on demand. Surya refuses to
+    attach to a URL whose /health is not 200, so answering only once a pod is up
+    would mean renting an A40 just to pass a probe."""
+    _reset()
+    import asyncio
+    pool = _FakePool()
+    _fixed_candidates(monkeypatch, [_pod_cand(pool)])
+    token = gateway.issue("m", "m")
+    reply = asyncio.run(gateway.health(token))
+    assert reply.status_code == 200
+    assert pool.acquired == 0          # nothing was rented to answer a probe
+
+
+def test_health_refuses_when_nothing_serves_the_model(monkeypatch):
+    _reset()
+    import asyncio
+    _fixed_candidates(monkeypatch, [])
+    token = gateway.issue("m", "m")
+    reply = asyncio.run(gateway.health(token))
+    assert reply.status_code == 503
+
+
+def test_an_unknown_lease_is_a_404():
+    _reset()
+    import asyncio
+    reply = asyncio.run(gateway.health("nope"))
+    assert reply.status_code == 404
+
+
+# ------------------------------------------- the hop must decide NOTHING at all #
+#: A real surya request: an inline page image, a strict json_schema, logprobs on.
+#: Every one of these was dropped by a proxy that parsed and re-emitted it, which
+#: is why this hop copies bytes instead.
+SURYA_BODY = (b'{"model":"datalab-to/surya-ocr-2","messages":[{"role":"user",'
+              b'"content":[{"type":"image_url","image_url":{"url":"data:image/png;'
+              b'base64,iVBORw0KGgo="}},{"type":"text","text":"OCR this page"}]}],'
+              b'"logprobs":true,"response_format":{"type":"json_schema",'
+              b'"json_schema":{"name":"structured_output","strict":true,'
+              b'"schema":{"type":"object"}}},"max_tokens":4096}')
+
+#: And the answer, with the logprobs surya turns into each line's confidence.
+UPSTREAM_REPLY = (b'{"choices":[{"index":0,"message":{"role":"assistant",'
+                  b'"content":"text"},"finish_reason":"stop","logprobs":'
+                  b'{"content":[{"token":"A","logprob":-0.25}]}}],'
+                  b'"usage":{"prompt_tokens":1,"completion_tokens":1}}')
+
+
+class _CapturingClient:
+    """Captures what the gateway sends upstream and answers verbatim."""
+
+    sent = {}
+
+    def __init__(self, *a, **kw):
+        self.timeout = kw.get("timeout")
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *exc):
+        return False
+
+    async def post(self, url, content=None, headers=None):
+        type(self).sent = {"url": url, "content": content, "headers": headers or {}}
+
+        class _R:
+            status_code = 200
+            headers = {"content-type": "application/json"}
+            content = UPSTREAM_REPLY
+        return _R()
+
+
+class _Req:
+    """The minimum of a Starlette request the route touches."""
+
+    def __init__(self, body, content_type="application/json"):
+        self._body = body
+        self.headers = {"content-type": content_type}
+
+    async def body(self):
+        return self._body
+
+
+def _forward(monkeypatch, body=SURYA_BODY, **kw):
+    import asyncio
+    import httpx
+    monkeypatch.setattr(httpx, "AsyncClient", _CapturingClient)
+    _no_node_load(monkeypatch)
+    _fixed_candidates(monkeypatch, kw.pop("cands", [_cand("node:a", base="http://a/v1")]))
+    token = gateway.issue("datalab-to/surya-ocr-2", "datalab-to/surya-ocr-2")
+    reply = asyncio.run(gateway.chat_completions(token, _Req(body)))
+    return reply, _CapturingClient.sent
+
+
+def test_the_request_body_arrives_upstream_unchanged(monkeypatch):
+    """Byte-identical: the image stays an image, the schema stays strict, and
+    logprobs stays asked for. This is the whole licence for adding a hop."""
+    _reset()
+    _, sent = _forward(monkeypatch)
+    assert sent["content"] == SURYA_BODY
+
+
+def test_the_reply_comes_back_unchanged(monkeypatch):
+    """Including the logprobs. Re-serialising the response is how they were lost."""
+    _reset()
+    reply, _ = _forward(monkeypatch)
+    assert reply.body == UPSTREAM_REPLY
+    assert b'"logprobs"' in reply.body
+    assert reply.status_code == 200
+
+
+def test_it_goes_to_the_chosen_upstream(monkeypatch):
+    _reset()
+    _, sent = _forward(monkeypatch, cands=[_cand("node:b", base="http://chosen/v1")])
+    assert sent["url"] == "http://chosen/v1/chat/completions"
+
+
+def test_a_locked_upstream_is_given_its_bearer(monkeypatch):
+    _reset()
+    cand = _cand("node:a", base="http://a/v1")
+    cand.api_key = "nodetoken"
+    _, sent = _forward(monkeypatch, cands=[cand])
+    assert sent["headers"]["Authorization"] == "Bearer nodetoken"
+
+
+def test_an_unlocked_upstream_gets_no_authorization_header(monkeypatch):
+    """A local vLLM with no key must not receive an empty bearer, which some
+    servers reject outright."""
+    _reset()
+    _, sent = _forward(monkeypatch)
+    assert "Authorization" not in sent["headers"]
+
+
+def test_the_slot_is_released_after_forwarding(monkeypatch):
+    """Otherwise the first page would make its upstream look permanently busy and
+    every later page would route elsewhere — or rent."""
+    _reset()
+    _forward(monkeypatch)
+    assert dispatch.inflight() == 0
+
+
+def test_the_pod_slot_is_released_even_if_the_upstream_errors(monkeypatch):
+    _reset()
+    import asyncio
+    import httpx
+
+    class _Failing(_CapturingClient):
+        async def post(self, url, content=None, headers=None):
+            raise RuntimeError("connection reset")
+    monkeypatch.setattr(httpx, "AsyncClient", _Failing)
+    pool = _FakePool()
+    _fixed_candidates(monkeypatch, [_pod_cand(pool)])
+    token = gateway.issue("m", "m")
+    reply = asyncio.run(gateway.chat_completions(token, _Req(SURYA_BODY)))
+    assert reply.status_code == 502
+    assert pool.released == 1
+    assert dispatch.inflight() == 0
+
+
+def test_an_unknown_lease_cannot_forward_anything(monkeypatch):
+    _reset()
+    import asyncio
+    reply = asyncio.run(gateway.chat_completions("forged", _Req(SURYA_BODY)))
+    assert reply.status_code == 404
+
+
+def test_no_server_available_is_reported_as_the_resolver_said(monkeypatch):
+    _reset()
+    import asyncio
+    _fixed_candidates(monkeypatch, [])
+    token = gateway.issue("m", "m")
+    reply = asyncio.run(gateway.chat_completions(token, _Req(SURYA_BODY)))
+    assert reply.status_code == 503
+
+
+def test_models_answers_the_name_the_engine_checks(monkeypatch):
+    _reset()
+    import asyncio
+    token = gateway.issue("datalab-to/surya-ocr-2", "datalab-to/surya-ocr-2")
+    reply = asyncio.run(gateway.models(token))
+    import json
+    data = json.loads(reply.body)
+    assert data["data"][0]["id"] == "datalab-to/surya-ocr-2"
+
+
+# ------------------------------------------------- the engine must get THROUGH #
+# Two gates stand in front of this route on a real install, and the engine can
+# satisfy neither the usual way: it has no API token, and surya's probes send no
+# headers at all (`probe_health` / `probe_model_id` are bare httpx GETs). If
+# either gate refuses, surya does not degrade — it refuses to attach, and every
+# page fails with "SURYA_INFERENCE_URL is not reachable at /health".
+
+
+class _Probe:
+    """A request shaped like surya's probe: no Authorization header at all."""
+
+    def __init__(self, path, headers=None):
+        import types
+        self.url = types.SimpleNamespace(path=path)
+        self.headers = headers or {}
+        self.scope = {"server": ("127.0.0.1", 8777), "type": "http"}
+
+
+def _through_bearer_gate(path, headers=None):
+    """True when BearerAuthMiddleware lets the request reach the route."""
+    import asyncio
+    from codai.api.ratelimit import BearerAuthMiddleware
+    mw = BearerAuthMiddleware(lambda scope, receive, send: None)
+    passed = []
+
+    async def _call_next(_req):
+        passed.append(True)
+        return "reached-the-route"
+
+    reply = asyncio.run(mw.dispatch(_Probe(path, headers), _call_next))
+    return bool(passed) and reply == "reached-the-route"
+
+
+def test_a_probe_with_no_headers_at_all_gets_through(monkeypatch):
+    """surya checks /health with a bare httpx GET and refuses to attach if it is
+    not 200. The lease is therefore read from the URL, not from a header."""
+    _reset()
+    monkeypatch.setenv("CODERAI_API_TOKEN", "some-api-token")
+    token = gateway.issue("m", "m")
+    assert _through_bearer_gate(f"/v1/ocr/engine/{token}/health")
+    assert _through_bearer_gate(f"/v1/ocr/engine/{token}/v1/models")
+    assert _through_bearer_gate(f"/v1/ocr/engine/{token}/v1/chat/completions",
+                                {"authorization": f"Bearer {token}"})
+
+
+def test_a_forged_lease_is_not_a_way_in(monkeypatch):
+    """Otherwise the exemption would be a hole in the API gate rather than a
+    credential the engine happens to hold."""
+    _reset()
+    monkeypatch.setenv("CODERAI_API_TOKEN", "some-api-token")
+    gateway.issue("m", "m")
+    assert not _through_bearer_gate("/v1/ocr/engine/forged/health")
+
+
+def test_a_revoked_lease_stops_working(monkeypatch):
+    _reset()
+    monkeypatch.setenv("CODERAI_API_TOKEN", "some-api-token")
+    token = gateway.issue("m", "m")
+    assert _through_bearer_gate(f"/v1/ocr/engine/{token}/health")
+    gateway.revoke("m")
+    assert not _through_bearer_gate(f"/v1/ocr/engine/{token}/health")
+
+
+def test_the_lease_does_not_unlock_the_rest_of_the_api(monkeypatch):
+    """A lease authorises its own gateway path and nothing else."""
+    _reset()
+    monkeypatch.setenv("CODERAI_API_TOKEN", "some-api-token")
+    token = gateway.issue("m", "m")
+    assert not _through_bearer_gate("/v1/chat/completions",
+                                    {"authorization": f"Bearer {token}"})
+    assert not _through_bearer_gate("/v1/models",
+                                    {"authorization": f"Bearer {token}"})
+
+
+def test_the_engine_reaches_an_engine_only_process(monkeypatch):
+    """With CODERAI_INTERNAL_TOKEN set, the engine app rejects every request that
+    does not carry the front's internal header — and the OCR worker cannot carry
+    it. The gateway path is exempted, which is safe because the lease is still
+    required behind it."""
+    import asyncio
+    from codai.api.app import _InternalAuthMiddleware
+    monkeypatch.setenv("CODERAI_INTERNAL_TOKEN", "internal-secret")
+    reached = []
+
+    async def _inner(scope, receive, send):
+        reached.append(scope["path"])
+
+    mw = _InternalAuthMiddleware(_inner)
+    token = gateway.issue("m", "m")
+
+    sent = []
+
+    async def _send(msg):
+        sent.append(msg)
+
+    # No internal header, as the OCR worker sends it.
+    asyncio.run(mw({"type": "http", "path": f"/v1/ocr/engine/{token}/health",
+                    "headers": []}, None, _send))
+    assert reached == [f"/v1/ocr/engine/{token}/health"]
+
+    # Everything else is still refused without the header.
+    reached.clear()
+    asyncio.run(mw({"type": "http", "path": "/v1/chat/completions",
+                    "headers": []}, None, _send))
+    assert reached == []
+    assert any(m.get("status") == 403 for m in sent if m.get("type")
+               == "http.response.start")

tests/test_ocr_served_model.py

@@ -393,45 +393,76 @@ class _Node:
         return {"Authorization": f"Bearer {self.api_key}"} if self.api_key else {}
 
 
-def test_a_model_held_by_a_cluster_node_resolves_to_that_node(monkeypatch):
+def test_a_model_held_by_a_cluster_node_is_found(monkeypatch):
     """A node is a whole coderai registered as a remote ENGINE, so no backend in the
     model layer knows it exists. Without asking the nodes, a node-hosted VLM answered
-    "not served over HTTP" — on an install that was serving it perfectly well."""
+    "not served over HTTP" — on an install that was serving it perfectly well.
+
+    Which node serves a given page is now decided per request by load
+    (codai.ocr.dispatch), so this asserts the discovery: the node that holds the
+    model is a candidate, carrying the shared cluster token."""
+    from codai.ocr import dispatch
+    dispatch.forget()
     _models_layer(monkeypatch)
     nodes = [_Node("alpha", "https://a.lan:8443"), _Node("beta", "https://b.lan:8443")]
     _cluster(monkeypatch, nodes, {"alpha": ["qwen"], "beta": ["datalab-to/surya-ocr-2"]})
-    base, key, served = served_model.endpoint_for("datalab-to/surya-ocr-2")
-    assert base == "https://b.lan:8443/v1"
-    assert key == "clustertok"          # the shared token fills a blank per-node key
-    assert served == "datalab-to/surya-ocr-2"
+    cands = dispatch._node_candidates("datalab-to/surya-ocr-2")
+    assert [c.base for c in cands] == ["https://b.lan:8443/v1"]
+    assert cands[0].api_key == "clustertok"   # shared token fills a blank node key
+
+
+def test_every_node_that_holds_it_is_a_candidate_not_just_the_first(monkeypatch):
+    """The first node that answered used to win for ever, however busy it got."""
+    from codai.ocr import dispatch
+    dispatch.forget()
+    _models_layer(monkeypatch)
+    nodes = [_Node("alpha", "https://a.lan:8443"), _Node("beta", "https://b.lan:8443")]
+    _cluster(monkeypatch, nodes, {"alpha": ["m"], "beta": ["m"]})
+    assert len(dispatch._node_candidates("m")) == 2
 
 
 def test_a_node_that_does_not_answer_is_skipped_not_fatal(monkeypatch):
+    from codai.ocr import dispatch
+    dispatch.forget()
     _models_layer(monkeypatch)
     nodes = [_Node("down", "https://d.lan:8443"), _Node("up", "https://u.lan:8443")]
     _cluster(monkeypatch, nodes, {"down": None, "up": ["m"]})
-    base, _, _ = served_model.endpoint_for("m")
-    assert base == "https://u.lan:8443/v1"
+    assert [c.base for c in dispatch._node_candidates("m")] == ["https://u.lan:8443/v1"]
 
 
 def test_the_nodes_are_not_asked_when_clustering_is_off(monkeypatch):
     """Probing every configured node on an install that does not cluster would add a
     timeout to the first page of every document."""
+    from codai.ocr import dispatch
+    dispatch.forget()
     _models_layer(monkeypatch)
     _cluster(monkeypatch, [_Node("alpha", "https://a.lan:8443")],
              {"alpha": ["m"]}, enabled=False)
-    with pytest.raises(OcrError):
-        served_model.endpoint_for("m")
+    assert dispatch._node_candidates("m") == []
 
 
 def test_runpod_wins_over_a_node_that_also_lists_it(monkeypatch):
     """An explicit backend pin on the entry is the operator's decision and is
-    authoritative; node discovery is the fallback, not an override."""
-    backend = FakeBackend(url="https://pod/v1", key="podtok")
-    _models_layer(monkeypatch, runpod=True, engine="runpod")
-    import codai.backends.runpod as rp
-    backend.load_model = lambda mid: None
-    monkeypatch.setattr(rp, "RunpodBackend", lambda cfg: backend)
-    _cluster(monkeypatch, [_Node("alpha", "https://a.lan:8443")], {"alpha": ["m"]})
-    base, key, _ = served_model.endpoint_for("m")
-    assert (base, key) == ("https://pod/v1", "podtok")
+    authoritative; node discovery is the fallback, not an override.
+
+    Load-aware picking must not quietly repeal that: a pool that is the model's
+    configured home is marked primary and is taken first, even though a node is
+    the cheaper hop and would otherwise win on load."""
+    from codai.ocr import dispatch
+    dispatch.forget()
+    pool_taken = []
+
+    node = dispatch.Candidate(key="node:alpha", kind="node",
+                              base="https://a.lan:8443/v1", priority=1)
+
+    def _acquire():
+        pool_taken.append(1)
+        return "https://pod/v1", "podtok", lambda: None
+    pod = dispatch.Candidate(key="pod:m", kind="pod", base="", priority=2,
+                             acquire=_acquire, primary=True)
+    monkeypatch.setattr(dispatch, "candidates", lambda m: [node, pod])
+    monkeypatch.setattr(dispatch, "_node_reported_load", lambda c: 0)
+    lease = dispatch.choose("m")
+    assert (lease.base, lease.api_key) == ("https://pod/v1", "podtok")
+    assert pool_taken == [1]
+    lease.release()