ocr: a pipeline asks where the engine is, and a node is one of the answers

Everywhere else in coderai an engine is what RUNS a model. `ocr.default_engine`
borrows the word for two different kinds of thing: paddle and docTR are runners
with their own weights, while surya-2 and olmOCR are pipelines over a model that
vLLM runs. Only the second kind has to be told where a server is — and that is
the same question the model layer already answers for everything else.

So answer it in all the places a model can actually be served. A cluster node
was the one that was missing: a node is a whole coderai registered with the head
as a remote ENGINE, so nothing in the backend layer knows it exists, and a VLM
hosted there came back as "not served over HTTP" from an install that was
serving it perfectly well. The node's own catalogue is the authority, so it is
asked. An explicit backend pin still wins — discovery is the fallback, not an
override — and the nodes are not probed at all when clustering is off, which
would have added a timeout to the first page of every document.

Also: one test here was passing for the wrong reason. `from codai.models import
manager` reads the package attribute and never consults sys.modules, so a fake
installed there was simply not seen. Patched as attributes of the real module,
which is what made the precedence test fail honestly.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PxYHWbCAjjjD7A1rhFKtaW

codai/ocr/served_model.py

@@ -78,6 +78,10 @@ def _entry_backend(model_id: str):
             raise OcrError(f"model {model_id!r}: RunPod endpoint unusable: {exc}",
                            status=503)
 
+    node = _cluster_node_endpoint(model_id)
+    if node is not None:
+        return node
+
     if mm.resolve_engine_backend(model_id) == "vllm":
         try:
             from codai.api import runpod_worker, vllm_worker
@@ -98,6 +102,44 @@ def _entry_backend(model_id: str):
     )
 
 
+def _cluster_node_endpoint(model_id: str):
+    """A cluster node that holds ``model_id``, as a fixed endpoint, or None.
+
+    A node is a whole coderai, registered with the head as a remote ENGINE rather than
+    as a model-manager backend — so nothing in the backend layer knows about it, and a
+    pipeline asking "where is this model served" would have been told nowhere. The
+    node's own catalogue is the authority, so it is asked: whichever enabled node lists
+    the id serves it, with the cluster token as the key.
+    """
+    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 None
+        token = (getattr(ccfg, "token", "") or "").strip()
+        for spec in enabled_nodes(ccfg):
+            if token and not spec.api_key:
+                spec.api_key = token
+            base = spec.url.rstrip("/")
+            base = base if base.endswith("/v1") else base + "/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:
+                return _Fixed(base, spec.api_key, model_id)
+    except Exception:
+        return None
+    return None
+
+
 class _Fixed:
     """A backend-shaped answer for the cases that resolve to one fixed URL."""
 

docs/ocr.md

@@ -89,6 +89,21 @@ one in the vocabulary the rest of the server already speaks:
 "ocr": { "surya_serve": "model", "surya_model_id": "datalab-to/surya-ocr-2" }
 ```
 
+A word on vocabulary, because this subsystem borrows one. Everywhere else in coderai an
+**engine** is the thing that *runs* a model — vLLM, llama.cpp, ds4. In `ocr.default_engine`
+the same word covers two different kinds of thing: paddle and docTR really are runners
+(own weights, no server), while **surya-2 and olmOCR are pipelines over a model something
+else runs**. Their engine is vLLM. That is why they alone need to be told where a server
+is, and why the answer is the same question the rest of coderai already answers for every
+model: local, a cluster node, or a rented pod.
+
+| where the model is served | how `model` mode finds it |
+|---|---|
+| this box, coderai's own vLLM | the vLLM backend's instance for that entry |
+| a cluster node | the node that lists the id in its own `/v1/models`, with the cluster token |
+| RunPod | the pod pool for that entry, provisioning if it is cold |
+| any other server | the entry's `service_url` |
+
 coderai resolves that id through the backend that owns it — provisioning a pod if the
 pool is cold — and hands the worker `SURYA_INFERENCE_URL`, `SURYA_MODEL_CHECKPOINT` (the
 served name; Surya refuses a server whose `/v1/models` disagrees) and `VLLM_API_KEY` (pods

tests/test_ocr_served_model.py

@@ -70,6 +70,18 @@ class FakeBackend:
         return self.url, self.key
 
 
+def _models_layer(monkeypatch, runpod=False, engine=None):
+    """Steer the model layer. Patched as ATTRIBUTES of the real module: the resolver
+    does `from codai.models import manager`, which reads the package attribute and
+    never looks at sys.modules — a fake installed there is simply not seen (and a test
+    that thinks it is passes for the wrong reason)."""
+    from codai.models import manager as real
+    monkeypatch.setattr(real, "runpod_should_handle", lambda m: runpod)
+    monkeypatch.setattr(real, "resolve_engine_backend", lambda m: engine)
+    monkeypatch.setattr(real, "get_active_runpod_config", lambda: None, raising=False)
+    return real
+
+
 @pytest.fixture(autouse=True)
 def _clean_cache():
     served_model.forget()
@@ -106,12 +118,7 @@ def test_the_backend_is_resolved_once_not_per_page(monkeypatch):
 def test_a_model_nobody_serves_over_http_says_so(monkeypatch):
     """The alternative is a loopback to our own /v1, which would hide the
     misconfiguration behind an extra hop and drop Surya's guided decoding."""
-    mod = types.SimpleNamespace(
-        runpod_should_handle=lambda m: False,
-        resolve_engine_backend=lambda m: None,
-        get_active_runpod_config=lambda: None,
-    )
-    monkeypatch.setitem(sys.modules, "codai.models.manager", mod)
+    _models_layer(monkeypatch)
     with pytest.raises(OcrError) as caught:
         served_model.endpoint_for("local-only-model")
     assert caught.value.status == 503
@@ -337,3 +344,94 @@ def test_a_proxy_forwards_the_schema_it_was_given():
         src = (ROOT / mod).read_text()
         assert 'response_format.get("type") == "json_object"' not in src, mod
         assert src.count('payload["response_format"] = response_format') == 2, mod
+
+
+# ------------------------------------------------------------- the three places
+def _cluster(monkeypatch, nodes, listing, enabled=True, token="clustertok"):
+    """Stand in for the cluster config and the nodes' own catalogues."""
+    import types as _t
+
+    cfg = _t.SimpleNamespace(cluster=_t.SimpleNamespace(enabled=enabled, token=token))
+    monkeypatch.setitem(sys.modules, "codai.admin.routes",
+                        _t.SimpleNamespace(config_manager=_t.SimpleNamespace(config=cfg)))
+
+    class _Reply:
+        def __init__(self, ids):
+            self.status_code = 200 if ids is not None else 500
+            self._ids = ids or []
+
+        def json(self):
+            return {"data": [{"id": i} for i in self._ids]}
+
+    class _Client:
+        def __init__(self, spec):
+            self.spec = spec
+
+        def __enter__(self):
+            return self
+
+        def __exit__(self, *a):
+            return False
+
+        def get(self, url):
+            for node in nodes:
+                if node.url.rstrip("/") in url:
+                    return _Reply(listing.get(node.name))
+            return _Reply(None)
+
+    monkeypatch.setitem(sys.modules, "codai.cluster.nodes",
+                        _t.SimpleNamespace(enabled_nodes=lambda c: nodes,
+                                           sync_client=_Client))
+    return cfg
+
+
+class _Node:
+    def __init__(self, name, url, api_key=""):
+        self.name, self.url, self.api_key = name, url, api_key
+
+    def headers(self):
+        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):
+    """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."""
+    _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"
+
+
+def test_a_node_that_does_not_answer_is_skipped_not_fatal(monkeypatch):
+    _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"
+
+
+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."""
+    _models_layer(monkeypatch)
+    _cluster(monkeypatch, [_Node("alpha", "https://a.lan:8443")],
+             {"alpha": ["m"]}, enabled=False)
+    with pytest.raises(OcrError):
+        served_model.endpoint_for("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")