placement: where a model runs is configured, not discovered

Three corrections to the fan-out, all the same mistake in different places:
the code was deciding things the configuration had already decided.

Nodes are no longer discovered. Asking every cluster node what it holds and
using whoever answered is configuration by probe: it puts work on a machine
nobody chose and moves it the moment a different node replies first. A node
serves a model because the model is PINNED to it — `engine` on the entry, the
same hard constraint the front router already honours, comma-separated when
several nodes should share the work. A pinned node that does not hold the
model is reported, not silently replaced.

The decision is not an OCR decision, so it does not live in codai/ocr: any
model, any caller that needs an endpoint rather than a proxy hop, gets the
same answer from codai/models/placement.py. For requests that pass THROUGH
coderai this was already true — the router honours the pin, treats a node as
an engine by name, reaches RunPod through the backend. This is that decision
for a caller that must hold an address itself, which is why OCR needed it.

And "how busy is too busy" is the model's own scale_up_inflight_per_pod, on
the model page, instead of the parallel ocr.serve_spill_inflight knob I had
added — one number per model, already there, already meaning exactly this. A
model with no RunPod block has nothing to rent, so load alone never moves its
pages.

Finally, the caller: resolve_engine accepted the four pipeline names only, so
a client naming the model it can see in /v1/models got "unknown OCR engine".
Both now work — engine=surya and engine=datalab-to/surya-ocr-2 reach the same
place — because which pipeline reads a page, and where its model runs, are
ours to know and not the client's.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PxYHWbCAjjjD7A1rhFKtaW

codai/admin/routes.py

@@ -4141,7 +4141,6 @@ def build_settings_dict(c, gpu_cards):
             "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,
@@ -4941,11 +4940,6 @@ async def api_save_settings(request: Request, username: str = Depends(require_ad
         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

@@ -1205,10 +1205,9 @@
     </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>
+    <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 wherever that model is configured to run: this machine, the cluster node it is pinned to, or its RunPod pool, whichever is least busy. Without it the engine pins to a single pod, and the pool never sees the load so it never grows. <b>Where</b> a model runs, and how busy is too busy (<code>scale_up_inflight_per_pod</code>), are set per model on the <a href="/admin/models">Models</a> page — the same settings every other request already follows.</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>
@@ -2145,7 +2144,6 @@ async function loadSettings(){
     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;
@@ -2398,7 +2396,6 @@ async function saveSettings(silent){
       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/config.py

@@ -990,12 +990,10 @@ class OcrConfig:
     # 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).
+    # How busy is too busy comes from the MODEL (its own
+    # `scale_up_inflight_per_pod`), not from here — one knob per model, already on
+    # the model page, instead of a second OCR-shaped copy of it.
     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
@@ -1632,7 +1630,6 @@ class ConfigManager:
                 "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/ocr/dispatch.py → codai/models/placement.py

@@ -2,32 +2,40 @@
 # 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:
+"""Where should a request for this model run — asked per request, for any model.
+
+A caller names a model. Where it runs is coderai's business, and the answer comes
+from how the model is configured: on this machine, on a cluster node it is pinned
+to, on a rented pod, or on this machine until it is busy and then a rented pod.
+Nothing about that is specific to one kind of model or one endpoint.
+
+For requests that pass THROUGH coderai this has always been true — the front
+router honours the model's ``engine`` pin, treats a cluster node as an engine by
+name, reaches RunPod through the model's backend, and bursts to a pod when the
+model says to. This module is the same decision for a caller that cannot go
+through that path: a component that must hold an OpenAI **address** and speak to
+the model itself.
+
+OCR is why it exists. Surya-2 and olmOCR are pipelines, not models: they drive a
+model server directly, sending multipart images, a strict ``json_schema`` and
+``logprobs: true`` — all three of which were silently dropped the last time OCR
+went through the backend abstraction, which parses a request and re-emits it. So
+the pipeline keeps its direct address, and :mod:`codai.ocr.gateway` makes that
+address ours and forwards the bytes unchanged. What it cannot do is choose, and
+choosing is what this module does:
 
 * 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.
+  ``scale_up_inflight_per_pod`` in-flight requests, counted as a request passes
+  through the pool. A caller holding one pod's URL never passed through, so the
+  pool saw no load however hard it was driven — and had the caller wanted a
+  second pod, it had no address for one;
+* a model pinned to several cluster nodes went to whichever answered first, for
+  ever, however busy that node became;
+* a local model configured to burst to RunPod under load never did: that trigger
+  lives on the proxy path such a caller bypasses.
+
+Any future component that needs "an endpoint for model X, placed as configured"
+should ask here rather than grow its own answer.
 """
 
 import threading
@@ -162,13 +170,53 @@ def _local_candidate(model_id: str) -> Optional[Candidate]:
                      base=base.rstrip("/") + "/v1", priority=0, primary=True)
 
 
+def model_entry(model_id: str) -> dict:
+    """The model's own entry in models.json, or {}."""
+    try:
+        from codai.admin.routes import config_manager
+        md = getattr(config_manager, "models_data", {}) or {}
+        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:
+                    return m
+    except Exception:
+        return {}
+    return {}
+
+
+def _pinned_engines(model_id: str) -> set:
+    """The engines this model is PINNED to, lower-cased, or an empty set.
+
+    ``engine`` on a model entry is what the front router already treats as a hard
+    constraint — the model runs there or not at all. A cluster node is an engine
+    by that name, so this is also how an operator says "serve it on box2".
+    """
+    pin = str((model_entry(model_id) or {}).get("engine") or "").strip().lower()
+    if not pin:
+        return set()
+    # A comma-separated pin names several (so OCR can spread over them); the
+    # front treats the whole string as one engine name, so a single pin — the
+    # common case — behaves identically in both places.
+    return {part.strip() for part in pin.split(",") if part.strip()}
+
+
 def _node_candidates(model_id: str) -> List[Candidate]:
-    """Every enabled cluster node whose own catalogue lists the model.
+    """The cluster nodes this model is PINNED to, that do hold it.
 
-    All of them, not the first one that answers: which node serves a given page
-    is decided per request, by load.
+    Nodes are never discovered. Asking every node what it has, and using whoever
+    answered, is configuration by probe: it would send pages to a machine nobody
+    chose, and silently move them when a different node happened to reply first.
+    A node serves OCR because the operator pinned the model to it.
+
+    All the pinned ones are returned, not the first that answers: which of them
+    takes a given page is decided per request, by load.
     """
     out: List[Candidate] = []
+    pinned = _pinned_engines(model_id)
+    if not pinned:
+        return out
     try:
         from codai.admin.routes import config_manager
         from codai.cluster.nodes import enabled_nodes, sync_client
@@ -177,6 +225,8 @@ def _node_candidates(model_id: str) -> List[Candidate]:
             return out
         token = (getattr(ccfg, "token", "") or "").strip()
         for spec in enabled_nodes(ccfg):
+            if str(getattr(spec, "name", "") or "").strip().lower() not in pinned:
+                continue
             if token and not spec.api_key:
                 spec.api_key = token
             root = spec.url.rstrip("/")
@@ -195,7 +245,13 @@ def _node_candidates(model_id: str) -> List[Candidate]:
             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))
+                                     node_url=root, primary=True))
+            else:
+                # Pinned to a node that does not hold it: say so, rather than
+                # quietly serving the page from somewhere the operator did not
+                # choose (or renting a card for it).
+                print(f"[ocr] {model_id!r} is pinned to node {spec.name!r}, which "
+                      f"does not list it — skipping that node", flush=True)
     except Exception:
         return out
     return out
@@ -239,16 +295,8 @@ def _spill_pod_candidate(model_id: str) -> Optional[Candidate]:
     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:
+        entry = model_entry(model_id)
+        if not entry:
             return None
         sp = entry.get("runpod_spillover")
         if not isinstance(sp, dict) or not sp.get("enabled"):
@@ -417,6 +465,32 @@ def _take(cand: Candidate) -> Optional[Lease]:
     return Lease(base, key, cand.key, cand.kind, release)
 
 
+def spill_threshold(model_id: str) -> int:
+    """In-flight requests at which a server we own is busy enough to rent instead.
+
+    This is the model's own ``scale_up_inflight_per_pod`` — the number already
+    configured, per model, to mean "one server should not hold more than this
+    before another is needed". Reusing it keeps one knob per model instead of a
+    parallel OCR-shaped one, and it is already in the model page.
+
+    0 when the model has no RunPod block to burst to: nothing to rent, so load
+    alone never changes where a page goes.
+    """
+    entry = model_entry(model_id)
+    for block in (entry.get("runpod"),
+                  (entry.get("runpod_spillover") or {}).get("target")):
+        if isinstance(block, dict):
+            try:
+                n = int(block.get("scale_up_inflight_per_pod") or 0)
+            except (TypeError, ValueError):
+                n = 0
+            if n > 0:
+                return n
+            if block is entry.get("runpod") or block:
+                return 4          # the documented default for a pool that exists
+    return 0
+
+
 def any_candidate(model_id: str) -> bool:
     """Whether anything at all can serve the model — without renting a thing."""
     try:

codai/ocr/gateway.py

@@ -122,8 +122,8 @@ async def health(token: str):
     got = lease(token)
     if got is None:
         return _unknown_lease()
-    from codai.ocr import dispatch
-    if not dispatch.any_candidate(got["model_id"]):
+    from codai.models import placement
+    if not placement.any_candidate(got["model_id"]):
         return JSONResponse(status_code=503,
                             content={"error": {"message": "no server for "
                                                           + got["model_id"]}})
@@ -165,14 +165,15 @@ async def chat_completions(token: str, request: Request):
 
     import httpx
 
-    from codai.admin.routes import config_manager
-    from codai.ocr import dispatch
+    from codai.models import placement
     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)
+        # How busy is too busy is the MODEL's own setting, the same one the pool
+        # scales on — not a second, OCR-shaped copy of it.
+        chosen = placement.choose(
+            got["model_id"],
+            spill_after=placement.spill_threshold(got["model_id"]))
     except Exception as exc:
         status = int(getattr(exc, "status", 503) or 503)
         return JSONResponse(status_code=status,

codai/ocr/manager.py

@@ -410,11 +410,36 @@ class OcrManager:
             getattr(cfg, "vlm_gpu_memory_utilization", 0.0),
         )
 
+    def _engine_serving_model(self, wanted: str) -> str:
+        """The engine configured to read pages with this MODEL, or "".
+
+        A client asks for OCR with a model it knows about — the id it sees in
+        /v1/models — not with the name of the pipeline we happen to drive it with.
+        Both are accepted: "surya" names the engine, and the Surya-2 checkpoint
+        names the same thing from the caller's side. Where that model then runs
+        (here, a pinned cluster node, a rented pod) is ours to decide and nothing
+        the caller should have to know.
+        """
+        cfg = self._require_cfg()
+        for engine, fields in (("surya", ("surya_model_id", "surya_model")),
+                               ("olmocr", ("olmocr_model_id", "olmocr_model"))):
+            for field in fields:
+                got = str(getattr(cfg, field, "") or "").strip().lower()
+                if got and got == wanted:
+                    return engine
+        return ""
+
     def resolve_engine(self, requested: Optional[str]) -> str:
         cfg = self._require_cfg()
         name = (requested or cfg.default_engine or "paddle").strip().lower()
         if name not in _ENGINE_FACTORIES:
-            raise OcrError(f"unknown OCR engine '{name}'", status=400)
+            by_model = self._engine_serving_model(name)
+            if by_model:
+                name = by_model
+        if name not in _ENGINE_FACTORIES:
+            raise OcrError(
+                f"unknown OCR engine or model '{name}' (engines: "
+                f"{', '.join(sorted(_ENGINE_FACTORIES))})", status=400)
         if not _engine_enabled(cfg, name):
             raise OcrError(
                 f"OCR engine '{name}' is not enabled (or license not accepted). "

codai/ocr/olmocr.py

@@ -353,9 +353,9 @@ class OlmOcrEngine(OcrEngine):
         """
         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)
+            from codai.models import placement
+            lease = placement.choose(
+                model_id, spill_after=placement.spill_threshold(model_id))
         except Exception:
             return self._ask_local(messages, max_tokens, temperature)
         try:

codai/ocr/served_model.py

@@ -186,8 +186,9 @@ def _fanout_endpoint(model_id: str):
     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)
+    from codai.models import placement
+    from codai.ocr import gateway
+    cands = placement.candidates(model_id)
     if not cands:
         return None
     served = model_id

codai/ocr/surya.py

@@ -190,12 +190,13 @@ class SuryaEngine(SubprocessOcrEngine):
                 raise
             print(f"[ocr] surya endpoint {self._server_url} is gone "
                   f"({first}) — re-resolving", flush=True)
-            from codai.ocr import dispatch, served_model
+            from codai.models import placement
+            from codai.ocr import 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()
+            placement.forget()
             self.cleanup()
             self.load()
             self._loaded = True

docs/ocr.md

@@ -313,17 +313,32 @@ address of coderai's own, and each request is forwarded — **verbatim**, body b
 in and body bytes out, so nothing is decided about it — to whichever server is
 least busy:
 
-| the model is… | a page goes to |
+| the model is configured… | 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.
+| to run on this machine | the local server, until it is busier than the model's own `scale_up_inflight_per_pod`, then a pod (the burst `runpod_spillover` describes) |
+| pinned to cluster nodes (`engine`) | the least busy of them, counting what each node reports for its **whole** install, so a node loaded with other work is left alone |
+| as several of these at once | its configured home first — an explicit `backend` or `engine` is a decision, not a hint — then the rest as overflow |
+
+Nothing here is discovered. A cluster node serves a model because the model is
+**pinned** to it, with the same `engine` pin the front router already treats as a
+hard constraint (a comma-separated pin names several nodes, so OCR can spread over
+them). Asking every node what it happens to hold, and using whoever answers first,
+would put work on a machine nobody chose and move it when a different node replied
+first.
+
+How busy is too busy is the model's own `scale_up_inflight_per_pod`, on the model
+page — the same number the pool scales on, not a second OCR-shaped copy of it. A
+model with no RunPod block has nothing to rent, so load alone never moves its
+pages. Turning `ocr.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.
+
+**None of this is visible to the caller.** A client asks for OCR with a model it
+has heard of — the id it sees in `/v1/models` — or with an engine name, and both
+work: `engine=surya` and `engine=datalab-to/surya-ocr-2` reach the same place.
+Where that model runs is read from its configuration, exactly as it is for every
+other request coderai serves.
 
 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

tests/test_ocr_fanout.py

@@ -27,7 +27,8 @@ 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.models import placement as dispatch
+from codai.ocr import gateway                            # noqa: E402
 from codai.ocr.base import OcrError                                # noqa: E402
 
 
@@ -632,3 +633,79 @@ def test_the_engine_reaches_an_engine_only_process(monkeypatch):
     assert reached == []
     assert any(m.get("status") == 403 for m in sent if m.get("type")
                == "http.response.start")
+
+
+# ------------------------------------------------- seamless from the client in #
+# A client asks for OCR with a model it has heard of — the id it sees in
+# /v1/models — and should not have to know that we read pages with it by driving
+# a pipeline called "surya", nor where that model is currently running.
+
+
+def _mgr(**ocr):
+    import types
+    from codai.ocr.manager import OcrManager
+    mgr = OcrManager.__new__(OcrManager)
+    fields = dict(default_engine="paddle", surya_model_id="", surya_model="",
+                  olmocr_model_id="", olmocr_model="", surya_enabled=True,
+                  surya_accept_license=True, olmocr_enabled=True,
+                  olmocr_accept_license=True, enabled=True)
+    fields.update(ocr)
+    cfg = types.SimpleNamespace(**fields)
+    mgr._require_cfg = lambda: cfg
+    return mgr
+
+
+def test_a_client_can_name_the_model_instead_of_the_engine():
+    mgr = _mgr(surya_model_id="datalab-to/surya-ocr-2")
+    assert mgr.resolve_engine("datalab-to/surya-ocr-2") == "surya"
+
+
+def test_the_olmocr_checkpoint_names_olmocr():
+    mgr = _mgr(olmocr_model_id="allenai/olmOCR-2-7B-1025-FP8")
+    assert mgr.resolve_engine("allenai/olmOCR-2-7B-1025-FP8") == "olmocr"
+
+
+def test_naming_the_engine_still_works():
+    """The existing clients say "surya", and must keep working unchanged."""
+    mgr = _mgr(surya_model_id="datalab-to/surya-ocr-2")
+    assert mgr.resolve_engine("surya") == "surya"
+
+
+def test_the_checkpoint_field_counts_too_not_only_the_model_id():
+    """An install serving surya locally sets surya_model, not surya_model_id."""
+    mgr = _mgr(surya_model="datalab-to/surya-ocr-2")
+    assert mgr.resolve_engine("datalab-to/surya-ocr-2") == "surya"
+
+
+def test_a_model_we_do_not_read_pages_with_is_still_a_clear_400():
+    from codai.ocr.base import OcrError
+    mgr = _mgr(surya_model_id="datalab-to/surya-ocr-2")
+    try:
+        mgr.resolve_engine("meta-llama/Llama-3.3-70B")
+    except OcrError as exc:
+        assert exc.status == 400
+        assert "engines:" in str(exc)
+    else:
+        raise AssertionError("expected an OcrError")
+
+
+# ------------------------------------- how busy is too busy is the MODEL's own #
+def test_the_spill_threshold_comes_from_the_model(monkeypatch):
+    """Not from a second OCR-shaped copy of a number the model already carries."""
+    from codai.models import placement
+    monkeypatch.setattr(placement, "model_entry",
+                        lambda m: {"id": m, "runpod": {"scale_up_inflight_per_pod": 9}})
+    assert placement.spill_threshold("m") == 9
+
+
+def test_a_pool_with_no_threshold_set_uses_the_documented_default(monkeypatch):
+    from codai.models import placement
+    monkeypatch.setattr(placement, "model_entry", lambda m: {"id": m, "runpod": {}})
+    assert placement.spill_threshold("m") == 4
+
+
+def test_a_model_with_nothing_to_rent_never_spills_on_load(monkeypatch):
+    """No RunPod block means no pod to burst to, so load alone cannot move a page."""
+    from codai.models import placement
+    monkeypatch.setattr(placement, "model_entry", lambda m: {"id": m})
+    assert placement.spill_threshold("m") == 0

tests/test_ocr_served_model.py

@@ -347,13 +347,19 @@ def test_a_proxy_forwards_the_schema_it_was_given():
 
 
 # ------------------------------------------------------------- the three places
-def _cluster(monkeypatch, nodes, listing, enabled=True, token="clustertok"):
-    """Stand in for the cluster config and the nodes' own catalogues."""
+def _cluster(monkeypatch, nodes, listing, enabled=True, token="clustertok",
+             pin=None, model="m"):
+    """Stand in for the cluster config, the model entry and the nodes' catalogues.
+
+    ``pin`` is the model entry's ``engine`` — what makes a node eligible at all.
+    """
     import types as _t
 
     cfg = _t.SimpleNamespace(cluster=_t.SimpleNamespace(enabled=enabled, token=token))
+    entries = [{"id": model, **({"engine": pin} if pin else {})}]
     monkeypatch.setitem(sys.modules, "codai.admin.routes",
-                        _t.SimpleNamespace(config_manager=_t.SimpleNamespace(config=cfg)))
+                        _t.SimpleNamespace(config_manager=_t.SimpleNamespace(
+                            config=cfg, models_data={"text_models": entries})))
 
     class _Reply:
         def __init__(self, ids):
@@ -401,11 +407,14 @@ def test_a_model_held_by_a_cluster_node_is_found(monkeypatch):
     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
+    from codai.models import placement as dispatch
+    from codai.models import placement as 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"]})
+    _cluster(monkeypatch, nodes, {"alpha": ["qwen"], "beta": ["datalab-to/surya-ocr-2"]},
+             pin="beta", model="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
@@ -413,31 +422,67 @@ def test_a_model_held_by_a_cluster_node_is_found(monkeypatch):
 
 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
+    from codai.models import placement as dispatch
+    from codai.models import placement as 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"]})
+    _cluster(monkeypatch, nodes, {"alpha": ["m"], "beta": ["m"]}, pin="alpha, beta")
     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
+    from codai.models import placement as dispatch
+    from codai.models import placement as 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"]})
+    _cluster(monkeypatch, nodes, {"down": None, "up": ["m"]}, pin="down, up")
     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
+    from codai.models import placement as dispatch
+    from codai.models import placement as dispatch
     dispatch.forget()
     _models_layer(monkeypatch)
     _cluster(monkeypatch, [_Node("alpha", "https://a.lan:8443")],
-             {"alpha": ["m"]}, enabled=False)
+             {"alpha": ["m"]}, enabled=False, pin="alpha")
+    assert dispatch._node_candidates("m") == []
+
+
+def test_a_node_is_never_used_unless_the_model_is_pinned_to_it(monkeypatch):
+    """Asking every node what it holds, and using whoever answers, is
+    configuration by probe: it puts work on a machine nobody chose and moves it
+    the moment a different node replies first. A node serves a model because the
+    operator pinned the model to it — the same `engine` pin the front router
+    already treats as a hard constraint."""
+    from codai.models import placement as dispatch
+    dispatch.forget()
+    _models_layer(monkeypatch)
+    _cluster(monkeypatch, [_Node("alpha", "https://a.lan:8443")], {"alpha": ["m"]})
+    assert dispatch._node_candidates("m") == []
+
+
+def test_a_pin_to_one_node_does_not_enlist_the_others(monkeypatch):
+    from codai.models import placement as 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"]}, pin="beta")
+    assert [c.base for c in dispatch._node_candidates("m")] == ["https://b.lan:8443/v1"]
+
+
+def test_a_pinned_node_that_does_not_hold_the_model_is_not_invented(monkeypatch):
+    """Better an honest "nothing serves this" than pages sent to a node that
+    cannot read them."""
+    from codai.models import placement as dispatch
+    dispatch.forget()
+    _models_layer(monkeypatch)
+    _cluster(monkeypatch, [_Node("alpha", "https://a.lan:8443")], {"alpha": ["other"]},
+             pin="alpha")
     assert dispatch._node_candidates("m") == []
 
 
@@ -448,7 +493,7 @@ def test_runpod_wins_over_a_node_that_also_lists_it(monkeypatch):
     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
+    from codai.models import placement as dispatch
     dispatch.forget()
     pool_taken = []