ocr: a lane is a page in flight, not a process (v0.3.43)

The worker protocol was one request, one reply, strictly in order, so pages in
flight WAS the process count — and a process costs ~600 MB of imported torch.
That is the whole reason `surya_instances: 64` meant 64 processes, 37 GB and a
nine-minute stall with the rented GPU idle at 2% KV-cache usage.

In the served modes that memory buys nothing: the recognition VLM is on the pod,
the local predictor holds no weights (1.7 MB of tokenizer and config on disk —
surya-2 returns its own boxes, so the detection model is never even fetched),
and the process rasterises a page and then waits on a socket. So a lane is a
worker OBJECT inside the process that already exists, with its own predictor and
its own page, sharing the one torch import. Effective concurrency is
instances x lanes; `surya_lanes` defaults to 4, served modes only, built on
demand.

Three things make one process safe to drive concurrently:

  - every reply carries the id of the request it answers. Out-of-order replies
    are the point — a short page must not queue behind a dense one — and
    attributing one page's text to another request would be worse than losing
    it, so an unknown id is dropped rather than guessed at;
  - the token trace and the accent repairs are per THREAD. surya's hook is
    module-level, one _generate_one per process, and its sink was a list
    belonging to whichever worker object installed it last. Shared between
    lanes it would have filed one page's logprobs, and surya_repair_accents'
    rewrites, under a different page;
  - a worker that dies fails its pages instead of leaving them parked on a
    reply that is never coming, and killing the engine wakes everything
    waiting.

A single-lane engine sends no ids and reads replies in order exactly as before,
which is how paddle, docTR and surya-`local` are still driven. `lanes()` is
asked for politely (_engine_lanes), because an engine is anything with
recognize_image and a pool must not refuse to build over a capability an engine
simply does not have.

Tested against the REAL worker protocol with a stubbed engine: four 0.25s pages
finish in under 0.75s, on four distinct lane objects, and six pages over two
lanes never exceed two.
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.42"
+__version__ = "0.3.43"
 __tagline__ = "Complete orchestration, distribution and escalation of remotizable advanced inference."
 
 # Configure the CUDA caching allocator BEFORE torch is imported anywhere.

codai/admin/routes.py

@@ -4138,6 +4138,7 @@ def build_settings_dict(c, gpu_cards):
             "surya_enabled": c.ocr.surya_enabled,
             "surya_accept_license": c.ocr.surya_accept_license,
             "surya_instances": c.ocr.surya_instances,
+            "surya_lanes": getattr(c.ocr, "surya_lanes", 4),
             "surya_langs": c.ocr.surya_langs,
             "surya_token_trace": c.ocr.surya_token_trace,
             "surya_repair_accents": c.ocr.surya_repair_accents,
@@ -4961,6 +4962,7 @@ async def api_save_settings(request: Request, username: str = Depends(require_ad
         if "surya_enabled" in d: o.surya_enabled = bool(d["surya_enabled"])
         if "surya_accept_license" in d: o.surya_accept_license = bool(d["surya_accept_license"])
         if "surya_instances" in d: o.surya_instances = _i("surya_instances", o.surya_instances, 1)
+        if "surya_lanes" in d: o.surya_lanes = _i("surya_lanes", getattr(o, "surya_lanes", 4), 1)
         if "surya_token_trace" in d: o.surya_token_trace = bool(d["surya_token_trace"])
         if "surya_repair_accents" in d: o.surya_repair_accents = bool(d["surya_repair_accents"])
         if "surya_langs" in d: o.surya_langs = (d.get("surya_langs") or "it").strip()

codai/admin/templates/settings.html

@@ -1203,7 +1203,8 @@
     <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-surya-enabled"><span>Enable</span></label>
       <label style="display:flex;align-items:center;gap:.4rem;cursor:pointer"><input type="checkbox" id="s-ocr-surya-license"><span>Accept license</span></label>
-      <div class="form-group" style="flex:1;min-width:110px"><label class="form-label">Instances</label><input type="number" id="s-ocr-surya-instances" class="form-input" min="1" placeholder="1"></div>
+      <div class="form-group" style="flex:1;min-width:110px"><label class="form-label" title="Worker PROCESSES, ~600 MB each. Capped at one per CPU core and half of free RAM.">Instances</label><input type="number" id="s-ocr-surya-instances" class="form-input" min="1" placeholder="1"></div>
+      <div class="form-group" style="flex:1;min-width:110px"><label class="form-label" title="Pages in flight per worker process (served modes only — in Local mode a lane would be a second copy of the model on the card). Effective concurrency is Instances x Lanes, and a lane costs almost nothing because the recognition runs on the serving model.">Lanes / inst.</label><input type="number" id="s-ocr-surya-lanes" class="form-input" min="1" placeholder="4"></div>
       <label style="display:flex;align-items:center;gap:.4rem;cursor:pointer" title="Local mode only. On, Surya runs on this machine's GPU or refuses to load — it will not quietly fall back to the CPU, where a page takes minutes. Off, the CPU is used deliberately."><input type="checkbox" id="s-ocr-surya-use-gpu" checked><span>Require GPU (Local)</span></label>
       <div class="form-group" style="flex:1;min-width:120px"><label class="form-label">Langs</label><input type="text" id="s-ocr-surya-langs" class="form-input" placeholder="it"></div>
       <label style="display:flex;align-items:center;gap:.4rem;cursor:pointer" title="Return the model's own per-token bytes wherever a page came back unreadable. A diagnostic — leave off unless you are chasing lost characters."><input type="checkbox" id="s-ocr-surya-token-trace"><span>Trace unreadable tokens</span></label>
@@ -2202,6 +2203,7 @@ async function loadSettings(){
     document.getElementById('s-ocr-surya-enabled').checked = !!ocr.surya_enabled;
     document.getElementById('s-ocr-surya-license').checked = !!ocr.surya_accept_license;
     document.getElementById('s-ocr-surya-instances').value = ocr.surya_instances ?? 1;
+    document.getElementById('s-ocr-surya-lanes').value = ocr.surya_lanes ?? 4;
     document.getElementById('s-ocr-surya-use-gpu').checked = ocr.surya_use_gpu !== false;
     document.getElementById('s-ocr-surya-langs').value = ocr.surya_langs ?? 'it';
     document.getElementById('s-ocr-surya-token-trace').checked = !!ocr.surya_token_trace;
@@ -2487,6 +2489,7 @@ async function saveSettings(silent){
       surya_enabled: document.getElementById('s-ocr-surya-enabled').checked,
       surya_accept_license: document.getElementById('s-ocr-surya-license').checked,
       surya_instances: parseInt(document.getElementById('s-ocr-surya-instances').value) || 1,
+      surya_lanes: parseInt(document.getElementById('s-ocr-surya-lanes').value) || 4,
       surya_langs: document.getElementById('s-ocr-surya-langs').value.trim() || 'it',
       surya_token_trace: document.getElementById('s-ocr-surya-token-trace').checked,
       surya_repair_accents: document.getElementById('s-ocr-surya-repair-accents').checked,

codai/config.py

@@ -1028,7 +1028,16 @@ class OcrConfig:
     # --- Surya (LICENSE-GATED) --- runs in an ISOLATED venv subprocess (pillow<11 clash)
     surya_enabled: bool = False
     surya_accept_license: bool = False   # MUST be explicitly set (GPL — compatible with coderai GPLv3)
-    surya_instances: int = 1
+    surya_instances: int = 1             # WORKER PROCESSES (~600 MB each)
+    #: Pages in flight PER worker process, in the served modes only. The
+    #: protocol used to be one page per process, so concurrency was the process
+    #: count: `surya_instances: 64` meant 64 processes and 37 GB, and on
+    #: 2026-10-12 it meant a nine-minute stall with the GPU idle. In a served
+    #: mode the process only rasterises a page and waits on a socket, so lanes
+    #: buy the same concurrency for one torch import instead of N. Effective
+    #: concurrency is instances x lanes. In `local` mode this is ignored (a lane
+    #: would be a second copy of the model on the card).
+    surya_lanes: int = 4
     surya_langs: str = "it"
     # Return the model's own per-token bytes for the places a page came back
     # unreadable (see base.UNREADABLE). Off by default: it is a diagnostic, it
@@ -1778,6 +1787,7 @@ class ConfigManager:
                 "surya_enabled": self.config.ocr.surya_enabled,
                 "surya_accept_license": self.config.ocr.surya_accept_license,
                 "surya_instances": self.config.ocr.surya_instances,
+                "surya_lanes": getattr(self.config.ocr, "surya_lanes", 4),
                 "surya_langs": self.config.ocr.surya_langs,
                 "surya_token_trace": self.config.ocr.surya_token_trace,
                 "surya_repair_accents": self.config.ocr.surya_repair_accents,

codai/ocr/base.py

@@ -192,6 +192,16 @@ class OcrEngine(ABC):
         manager's VRAM-eviction thread). Overridden per engine."""
         self._loaded = False
 
+    def lanes(self) -> int:
+        """How many pages ONE instance may serve at once.
+
+        One by default: an engine is only safe to drive concurrently if it says
+        so. The in-process engines (chandra, olmOCR) get their concurrency from
+        the instance count instead, each instance being a slot rather than a
+        process, so they have no reason to raise this.
+        """
+        return 1
+
     def vram_gb(self) -> float:
         """Rough VRAM footprint (GB) of one loaded instance, for eviction accounting."""
         return 1.0

codai/ocr/manager.py

@@ -276,6 +276,20 @@ def _engine_instances(cfg, name: str) -> int:
     return want
 
 
+def _engine_lanes(eng) -> int:
+    """How many pages one instance of ``eng`` serves at once, asked politely.
+
+    `OcrEngine` has `lanes()`, but an engine is anything with `recognize_image` —
+    the fallback chain and several tests drive duck-typed ones — and a pool must
+    not refuse to build because an engine does not know about a capability it
+    simply does not have. Unknown means one.
+    """
+    try:
+        return max(1, int(eng.lanes() or 1))
+    except Exception:
+        return 1
+
+
 def _evict_for_ocr(needed_gb: float) -> None:
     """Ask the model manager to free ``needed_gb`` of VRAM before an OCR pool loads,
     so OCR instances contend for VRAM on equal footing with LLM/diffusion models
@@ -424,7 +438,12 @@ class _Pool:
                     if first and need <= 0:
                         await asyncio.to_thread(_evict_for_ocr, self.size * eng.vram_gb())
                     self._instances.append(eng)
-                    await self._q.put(eng)
+                    # A lane is a page this instance can serve CONCURRENTLY, so
+                    # the instance is offered once per lane. The queue is what
+                    # bounds concurrency, and handing out the same engine twice
+                    # is exactly what a multi-lane worker is for.
+                    for _ in range(_engine_lanes(eng)):
+                        await self._q.put(eng)
             except Exception as e:
                 self.release_sync()
                 self._fail_at = time.time()
@@ -433,7 +452,12 @@ class _Pool:
                 raise self._fail_err
             self._fail_err = None
             self._built = True
-            print(f"[ocr] engine '{self.name}': {self.size} instance(s) ready")
+            lanes = _engine_lanes(self._instances[0]) if self._instances else 1
+            if lanes > 1:
+                print(f"[ocr] engine '{self.name}': {self.size} instance(s) ready, "
+                      f"{lanes} lane(s) each — {self.size * lanes} pages at once")
+            else:
+                print(f"[ocr] engine '{self.name}': {self.size} instance(s) ready")
 
     def release_sync(self) -> float:
         """Tear down all instances (free VRAM). SYNC — safe to call from the model

codai/ocr/subprocess_engine.py

@@ -65,7 +65,12 @@ class SubprocessOcrEngine(OcrEngine):
     def __init__(self, cfg):
         super().__init__(cfg)
         self._proc = None
-        self._lock = threading.Lock()
+        self._lock = threading.Lock()        # guards _pending
+        self._wlock = threading.Lock()       # one writer on the worker's stdin
+        self._pending = {}                   # request id -> {"event", "reply"}
+        self._next_id = 0
+        self._mux = False                    # replies carry ids and arrive out of order
+        self._reader = None
 
     # -- subclass hooks ----------------------------------------------------
 
@@ -79,6 +84,15 @@ class SubprocessOcrEngine(OcrEngine):
     def _auto_build(self) -> bool:
         return False
 
+    def lanes(self) -> int:
+        """How many pages this engine's worker may have in flight at once.
+
+        One, unless a subclass knows better. A lane costs a worker OBJECT inside
+        the existing process, not a process — which is only cheap when the
+        engine's heavy part lives somewhere else (see SuryaEngine.lanes).
+        """
+        return 1
+
     def _worker_opts(self) -> dict:
         return {}
 
@@ -142,11 +156,25 @@ class SubprocessOcrEngine(OcrEngine):
         ready = self._readline()
         if not ready.get("ready"):
             raise OcrError(f"{self.name}: worker did not signal ready", status=500)
-        resp = self._rpc({"cmd": "load", "opts": self._worker_opts()})
+        opts = dict(self._worker_opts())
+        want = max(1, int(self.lanes() or 1))
+        opts["lanes"] = want
+        resp = self._rpc({"cmd": "load", "opts": opts})
         if not resp.get("ok"):
             self._kill()
             raise OcrError(resp.get("error") or f"{self.name}: load failed",
                            status=int(resp.get("status", 500)))
+        # The worker echoes the lanes it will actually serve. Only then does the
+        # parent start reading replies out of order: a worker that did not answer
+        # with a lane count is driven exactly as before, one page at a time.
+        got = int(resp.get("lanes") or 1)
+        self._mux = got > 1
+        if self._mux:
+            self._reader = threading.Thread(target=self._read_replies, daemon=True,
+                                            name=f"ocr-{self.name}-reader")
+            self._reader.start()
+            print(f"[ocr] {self.name}: worker serves {got} page(s) at once "
+                  f"in one process", flush=True)
 
     def _readline(self) -> dict:
         line = self._proc.stdout.readline()
@@ -164,6 +192,75 @@ class SubprocessOcrEngine(OcrEngine):
         self._proc.stdin.flush()
         return self._readline()
 
+    def _read_replies(self) -> None:
+        """Hand each reply to the page that asked for it, until the pipe closes.
+
+        A reply with no id belongs to a sequential RPC (load, ping) and cannot
+        reach here: those run before the reader starts. Anything unrecognised is
+        dropped rather than guessed at — attributing a page's text to the wrong
+        request is the one failure worse than losing it.
+        """
+        proc = self._proc
+        while True:
+            try:
+                line = proc.stdout.readline()
+            except Exception:
+                break
+            if not line:
+                break
+            try:
+                msg = json.loads(line)
+            except Exception:
+                continue
+            rid = msg.get("id")
+            if rid is None:
+                continue
+            with self._lock:
+                slot = self._pending.pop(rid, None)
+            if slot is not None:
+                slot["reply"] = msg
+                slot["event"].set()
+        # The worker is gone. Everyone still waiting would otherwise wait for ever.
+        self._fail_pending(f"{self.name}: worker closed the pipe")
+
+    def _fail_pending(self, why: str) -> None:
+        with self._lock:
+            waiting = list(self._pending.values())
+            self._pending.clear()
+        for slot in waiting:
+            slot["reply"] = {"ok": False, "error": why, "status": 500}
+            slot["event"].set()
+
+    def _rpc_mux(self, msg: dict) -> dict:
+        """One page, tagged, answered by the reader thread whenever it lands."""
+        if self._proc is None or self._proc.poll() is not None:
+            raise OcrError(f"{self.name}: worker not running", status=500)
+        slot = {"event": threading.Event(), "reply": None}
+        with self._lock:
+            self._next_id += 1
+            rid = self._next_id
+            self._pending[rid] = slot
+        try:
+            line = json.dumps(dict(msg, id=rid)) + "\n"
+            with self._wlock:
+                self._proc.stdin.write(line)
+                self._proc.stdin.flush()
+        except Exception as e:
+            with self._lock:
+                self._pending.pop(rid, None)
+            raise OcrError(f"{self.name}: worker not running ({e})", status=500)
+        # No deadline on the page itself — a dense scan legitimately takes
+        # minutes, and the engine above has its own timeout. But a worker that
+        # DIED must not read as a page still being worked on, so the wait is
+        # broken up to check that it is still there.
+        while not slot["event"].wait(5.0):
+            if self._proc is None or self._proc.poll() is not None:
+                with self._lock:
+                    self._pending.pop(rid, None)
+                raise OcrError(f"{self.name}: worker died while reading a page",
+                               status=500)
+        return slot["reply"] or {"ok": False, "error": f"{self.name}: no reply"}
+
     def _kill(self):
         try:
             if self._proc:
@@ -171,6 +268,8 @@ class SubprocessOcrEngine(OcrEngine):
         except Exception:
             pass
         self._proc = None
+        self._mux = False
+        self._fail_pending(f"{self.name}: worker stopped")
 
     def cleanup(self) -> None:
         """Stop the worker subprocess (frees its VRAM). SYNC."""
@@ -188,8 +287,11 @@ class SubprocessOcrEngine(OcrEngine):
         buf = io.BytesIO()
         image.convert("RGB").save(buf, "PNG")
         b64 = base64.b64encode(buf.getvalue()).decode("ascii")
-        with self._lock:                       # one request at a time per subprocess
-            resp = self._rpc({"cmd": "ocr", "image": b64})
+        if self._mux:
+            resp = self._rpc_mux({"cmd": "ocr", "image": b64})
+        else:
+            with self._lock:                   # one request at a time per subprocess
+                resp = self._rpc({"cmd": "ocr", "image": b64})
         if not resp.get("ok"):
             raise OcrError(resp.get("error") or f"{self.name}: ocr failed",
                            status=int(resp.get("status", 500)))

codai/ocr/surya.py

@@ -76,6 +76,25 @@ class SuryaEngine(SubprocessOcrEngine):
     #: Modes where the recognition VLM lives in a model server we only talk to.
     SERVED_MODES = ("model", "vllm", "llamacpp", "server")
 
+    def lanes(self) -> int:
+        """Pages in flight per worker process.
+
+        Only in the SERVED modes, and that is the whole point: there the
+        recognition VLM is somewhere else, the predictor holds no weights (1.7 MB
+        of tokenizer and config on disk) and this process rasterises a page and
+        then waits on a socket. Several pages can wait at once for the price of
+        the torch import that is already paid.
+
+        In `local` mode a lane would be a second copy of the model on this
+        machine's GPU, so there is exactly one.
+        """
+        if self._serve_mode() not in self.SERVED_MODES:
+            return 1
+        try:
+            return max(1, int(getattr(self.cfg, "surya_lanes", 4) or 1))
+        except (TypeError, ValueError):
+            return 4
+
     def _serve_mode(self) -> str:
         mode = (getattr(self.cfg, "surya_serve", "model") or "model").strip().lower()
         return "llamacpp" if mode == "server" else mode

codai/ocr/workers/ocr_worker.py

@@ -19,12 +19,48 @@ import base64
 import os
 import io
 import json
+import queue
 import sys
+import threading
+
+
+#: stdout is now written by several threads (one per page in flight), and a
+#: half-written JSON line is an unparseable reply for a page that actually
+#: succeeded. One lock, held for exactly one line.
+_OUT_LOCK = threading.Lock()
 
 
 def _emit(obj):
-    sys.stdout.write(json.dumps(obj) + "\n")
-    sys.stdout.flush()
+    line = json.dumps(obj) + "\n"
+    with _OUT_LOCK:
+        sys.stdout.write(line)
+        sys.stdout.flush()
+
+
+#: Per-PAGE trace state, kept per THREAD.
+#:
+#: surya's hook is module-level — there is one ``_generate_one`` per process —
+#: and the sink used to be a list belonging to whichever worker object installed
+#: the hook last. With one page per process that was exact. With several pages in
+#: flight in one process it would attribute one page's logprobs, and its accent
+#: repairs, to another page entirely: a diagnostic that lies is worse than none,
+#: and `surya_repair_accents` would rewrite the wrong text.
+_TRACE = threading.local()
+
+
+def _trace_begin():
+    """Start a page's trace. Returns its (calls, repairs) — this thread's own."""
+    _TRACE.calls = []
+    _TRACE.repairs = []
+    return _TRACE.calls, _TRACE.repairs
+
+
+def _trace_calls():
+    return getattr(_TRACE, "calls", None)
+
+
+def _trace_repairs():
+    return getattr(_TRACE, "repairs", None)
 
 
 def _html_to_text(html: str) -> str:
@@ -639,15 +675,14 @@ class SuryaWorker:
         self._rec_proc = None
         self._run_ocr = None
         self._langs = [l.strip() for l in str(opts.get("langs", "it")).split(",") if l.strip()] or ["it"]
-        #: Per-call token logprobs, collected by the hook installed in load().
-        self._calls = []
         self._trace = bool(opts.get("token_trace", False))
         #: Rebuild a reply's text from the model's own bytes when the text we are
         #: handed has lost characters. On by default: the alternative is shipping
         #: text we have MEASURED to be corrupt (see repair_content).
         self._repair = bool(opts.get("repair_accents", True))
-        #: [{"block": i, "recovered": n}] for this page.
-        self._repairs = []
+        #: The page's token logprobs and accent repairs live in _TRACE, per
+        #: thread — see _trace_begin(). They were per-worker-object until lanes
+        #: put several pages in one process.
 
     def load(self):
         """Load surya, matching the API of whichever release is installed.
@@ -701,24 +736,20 @@ class SuryaWorker:
         except Exception as e:
             print(f"[surya] token trace unavailable: {e}", file=sys.stderr)
             return
+        _oc._coderai_repair = self._repair
         if getattr(_oc, "_coderai_traced", False):
-            # The module is patched once per process; re-point it at THIS worker.
-            _oc._coderai_sink = self._calls
-            _oc._coderai_repairs = self._repairs
-            _oc._coderai_repair = self._repair
+            # Patched once per process. Nothing to re-point: the sink is the
+            # calling THREAD's, so every lane lands in its own page's trace.
             return
         inner = getattr(_oc, "_generate_one", None)
         if inner is None:
             print("[surya] token trace: no _generate_one to hook", file=sys.stderr)
             return
-        _oc._coderai_sink = self._calls
-        _oc._coderai_repairs = self._repairs
-        _oc._coderai_repair = self._repair
 
         def _wrapped(*args, **kwargs):
             result = inner(*args, **kwargs)
             try:
-                sink = getattr(_oc, "_coderai_sink", None)
+                sink = _trace_calls()
                 if sink is not None:
                     sink.append(getattr(result, "logprobs", None))
                 # Repaired HERE, before surya parses the reply, so the blocks, the
@@ -730,7 +761,7 @@ class SuryaWorker:
                                               getattr(result, "logprobs", None))
                     if fixed is not None:
                         result.raw = fixed
-                        reps = getattr(_oc, "_coderai_repairs", None)
+                        reps = _trace_repairs()
                         if reps is not None:
                             reps.append({"recovered": n})
             except Exception:
@@ -776,11 +807,10 @@ class SuryaWorker:
 
     def ocr(self, img):
         img = img.convert("RGB")
-        # One page per call, so the trace is cleared here rather than accumulated:
+        # One page per call, so the trace is started here rather than accumulated:
         # a long document would otherwise report the first page's anomalies on the
-        # hundredth.
-        del self._calls[:]
-        del self._repairs[:]
+        # hundredth. Per-thread, because several pages share this process.
+        calls, repairs = _trace_begin()
         if self._mode == "predictor":
             preds = self._predict(img)
         else:
@@ -827,19 +857,19 @@ class SuryaWorker:
                 "text": "\n".join(l["text"] for l in lines)}
         # Only when something actually went unread: a clean page stays byte-identical
         # to what it has always returned.
-        if self._repairs:
+        if repairs:
             # Said out loud rather than silently: the text the caller gets is not
             # byte-for-byte what the server sent, and anyone comparing the two
             # needs to know why.
             page["accents_repaired"] = {
-                "blocks": len(self._repairs),
-                "recovered": sum(int(r.get("recovered", 0)) for r in self._repairs),
+                "blocks": len(repairs),
+                "recovered": sum(int(r.get("recovered", 0)) for r in repairs),
                 "how": ("rebuilt from the model's own per-token bytes, which "
                         "decoded cleanly where message.content did not"),
             }
         if self._trace and UNREADABLE in page["text"]:
             try:
-                page["unreadable_trace"] = _trace_report(self._calls)
+                page["unreadable_trace"] = _trace_report(calls)
             except Exception as e:
                 # Always a report, never an absence: "the analysis broke" and "the
                 # bytes were fine" must not look the same.
@@ -855,12 +885,88 @@ class SuryaWorker:
 # Main loop
 # ---------------------------------------------------------------------------
 
+class _Lanes:
+    """Several pages in flight in ONE worker process.
+
+    The protocol was one request, one reply, strictly in order — so pages in
+    flight WAS the process count, and a process costs ~600 MB. In the served
+    modes (`surya_serve: model`) that memory buys nothing: the page is rasterised
+    here and recognised on a GPU somewhere else, so the worker spends its life
+    waiting on a socket. Raising concurrency by forking more of them is how
+    Aruba ended up with 64 processes, 37 GB and a nine-minute stall.
+
+    So a lane is a worker OBJECT, not a process: each has its own predictor and
+    its own page, and they share the one torch import. In the served modes a
+    predictor holds no weights (1.7 MB of tokenizer and config on disk), which is
+    what makes this nearly free. Lanes are built on demand, so a quiet install
+    pays for exactly one.
+    """
+
+    def __init__(self, make, lanes: int, first):
+        self._make = make
+        self._limit = max(1, int(lanes or 1))
+        self._idle = queue.Queue()
+        self._idle.put(first)
+        self._built = 1
+        self._lock = threading.Lock()
+        self._slots = threading.Semaphore(self._limit)
+
+    @property
+    def limit(self) -> int:
+        return self._limit
+
+    def borrow(self):
+        """A lane of our own, waiting for one if every lane is busy."""
+        self._slots.acquire()
+        try:
+            try:
+                return self._idle.get_nowait()
+            except queue.Empty:
+                pass
+            with self._lock:
+                worker = self._make()
+                self._built += 1
+            return worker
+        except BaseException:
+            self._slots.release()      # never strand a slot on a failed build
+            raise
+
+    def give_back(self, worker) -> None:
+        if worker is not None:
+            self._idle.put(worker)
+        self._slots.release()
+
+
+def _build(engine: str, opts: dict):
+    worker = PaddleWorker(opts) if engine == "paddle" else SuryaWorker(opts)
+    worker.load()
+    return worker
+
+
+def _serve_page(lanes: "_Lanes", msg: dict, engine: str) -> None:
+    """One page, on a lane of its own, answered with the id it was asked under."""
+    rid = msg.get("id")
+    worker = None
+    try:
+        worker = lanes.borrow()
+        from PIL import Image
+        data = base64.b64decode(msg["image"])
+        img = Image.open(io.BytesIO(data)); img.load()
+        _emit({"ok": True, "id": rid, "page": worker.ocr(img)})
+    except Exception as e:
+        _emit({"ok": False, "id": rid, "error": f"{engine} ocr failed: {e}",
+               "status": 500})
+    finally:
+        if worker is not None:
+            lanes.give_back(worker)
+
+
 def main():
     engine = sys.argv[1] if len(sys.argv) > 1 else "paddle"
     from PIL import Image  # noqa: F401 (ensure Pillow present before READY)
     _emit({"ready": True})
 
-    worker = None
+    lanes = None
     for line in sys.stdin:
         line = line.strip()
         if not line:
@@ -875,22 +981,38 @@ def main():
         if cmd == "load":
             try:
                 opts = msg.get("opts") or {}
-                worker = PaddleWorker(opts) if engine == "paddle" else SuryaWorker(opts)
-                worker.load()
-                _emit({"ok": True})
+                # The first lane is loaded HERE, synchronously, so a missing
+                # dependency is still a load failure and not a failed page.
+                first = _build(engine, opts)
+                lanes = _Lanes(lambda: _build(engine, opts),
+                               int(opts.get("lanes") or 1), first)
+                _emit({"ok": True, "lanes": lanes.limit})
             except Exception as e:
                 _emit({"ok": False, "error": f"{engine} load failed: {e}", "status": 503})
             continue
         if cmd == "ocr":
-            if worker is None:
-                _emit({"ok": False, "error": "not loaded", "status": 500}); continue
-            try:
-                from PIL import Image
-                data = base64.b64decode(msg["image"])
-                img = Image.open(io.BytesIO(data)); img.load()
-                _emit({"ok": True, "page": worker.ocr(img)})
-            except Exception as e:
-                _emit({"ok": False, "error": f"{engine} ocr failed: {e}", "status": 500})
+            if lanes is None:
+                _emit({"ok": False, "id": msg.get("id"), "error": "not loaded",
+                       "status": 500}); continue
+            if msg.get("id") is None:
+                # No id: the caller is reading replies in order, one at a time.
+                worker = None
+                try:
+                    worker = lanes.borrow()
+                    from PIL import Image
+                    data = base64.b64decode(msg["image"])
+                    img = Image.open(io.BytesIO(data)); img.load()
+                    _emit({"ok": True, "page": worker.ocr(img)})
+                except Exception as e:
+                    _emit({"ok": False, "error": f"{engine} ocr failed: {e}",
+                           "status": 500})
+                finally:
+                    if worker is not None:
+                        lanes.give_back(worker)
+                continue
+            # Answered out of order, under its id, while the loop reads the next.
+            threading.Thread(target=_serve_page, args=(lanes, msg, engine),
+                             daemon=True).start()
             continue
         if cmd == "shutdown":
             _emit({"ok": True}); return

docs/ocr.md

@@ -976,6 +976,7 @@ The two numbers look interchangeable on the OCR card and are not:
 |---|---|---|
 | `runpod.scale_up_inflight_per_pod`, `runpod.max_pods` | concurrent pages **per rented GPU**, and how many GPUs | the pod |
 | `ocr.surya_instances` (also `paddle`, `doctr`) | **worker processes on this machine** | ~600 MB RSS and a core's rendering, each |
+| `ocr.surya_lanes` | pages in flight **per worker process** (served modes) | almost nothing — see below |
 | `ocr.chandra_instances`, `ocr.olmocr_instances` | concurrency slots, in-process | a slot |
 
 Only surya, paddle and docTR spawn a process per instance — they run in their own
@@ -1004,9 +1005,48 @@ Both halves are now bounded:
   is down, since the replacement asks that same front for its lease. A 503 means
   the pool has nothing rented this second, which the lease does not care about.
 
-To raise OCR throughput, raise the pod-side numbers, not these. The local worker
-pool is a hard ceiling on how many pods surya can keep fed at all — which is the
-standing argument for running it in-process, as chandra and olmOCR already do.
+#### Lanes: concurrency without another process
+
+The worker protocol was one request, one reply, strictly in order, so **pages in
+flight was the process count** — and that is the whole reason 64 pages meant 64
+processes and 37 GB.
+
+In a served mode that memory buys nothing. The recognition VLM is on the pod; the
+local predictor holds **no weights** (1.7 MB of tokenizer and config on disk, and
+surya-2 returns its own boxes so the detection model is never fetched); and the
+process spends a page rasterising it and then waiting on a socket. So a **lane**
+is a worker *object* inside the process that already exists, with its own
+predictor and its own page, sharing the one torch import:
+
+```
+effective concurrency = <engine>_instances x <engine>_lanes
+```
+
+`surya_lanes` defaults to **4** and applies to the served modes only — in `local`
+mode a lane would be a second copy of the model on the card, so there is exactly
+one. Lanes are built on demand, so a quiet install still pays for one.
+
+Three details make it safe to drive one process concurrently:
+
+* **every reply carries the id of the request it answers.** Out-of-order replies
+  are the point — a short page must not wait behind a dense one — and attributing
+  one page's text to another request would be worse than losing it. A reply whose
+  id is unknown is dropped, never guessed at;
+* **the token trace and the accent repairs are per THREAD.** surya's hook is
+  module-level, one `_generate_one` per process, and its sink used to be a list
+  belonging to whichever worker object installed it last. Shared between lanes it
+  would have filed one page's logprobs — and `surya_repair_accents`' rewrites —
+  under a different page;
+* **a worker that dies fails its pages** instead of leaving them parked on a
+  reply that is never coming: the wait is broken up to check the process is still
+  there, and killing the engine wakes everything waiting.
+
+A single-lane engine sends no ids and reads replies in order, exactly as before,
+which is how paddle, docTR and surya-`local` are still driven.
+
+To raise OCR throughput, raise lanes and the pod-side numbers, not the process
+count. Instances still bound how much of this host OCR may take; lanes are what
+turn one process into many pages.
 
 ```bash
 curl -F file=@sentenza.pdf -F structured=true -F detect=both -F schema=italian_sentenza \

tests/ocr_lane_stub_worker.py

+# CoderAI - OpenAI-compatible API server
+# Copyright (C) 2026 Stefy Lanza <stefy@nexlab.net>
+# GPLv3 - see the project LICENSE.
+
+"""Stands in for an OCR engine's venv python, running the REAL worker protocol.
+
+The point of the lane tests is the protocol — ids, out-of-order replies, one page
+per lane object — so the protocol must be the shipped code and only the engine
+stubbed. ``codai/ocr/workers/ocr_worker.py`` is loaded by path (it is a script,
+not a module on sys.path), its two worker classes are replaced with a sleepy stub
+that reports which lane served the page, and its own ``main()`` is run.
+
+Driven by tests/test_ocr_worker_lanes.py; not a test itself.
+"""
+
+import importlib.util
+import os
+import time
+
+
+class _Stub:
+    """One object per lane, so a crossed reply shows up as a repeated lane."""
+
+    made = 0
+
+    def __init__(self, opts):
+        type(self).made += 1
+        self.lane = type(self).made
+
+    def load(self):
+        pass
+
+    def ocr(self, img):
+        # Long enough that four sequential pages could not fit in the window the
+        # test allows, so overlapping is the only way to pass it.
+        time.sleep(float(os.environ.get("OCR_STUB_PAGE_S") or 0.25))
+        return {"index": 0, "text": f"lane{self.lane}", "width": 1, "height": 1,
+                "lines": [], "regions": []}
+
+
+def main():
+    path = os.environ["OCR_WORKER_PATH"]
+    spec = importlib.util.spec_from_file_location("ocr_worker_under_test", path)
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+    mod.SuryaWorker = _Stub
+    mod.PaddleWorker = _Stub
+    mod.main()
+
+
+if __name__ == "__main__":
+    main()

tests/test_ocr_accent_repair.py

@@ -178,9 +178,10 @@ def test_a_repaired_page_says_so(monkeypatch):
     w._mode = "predictor"
 
     def _predict(img):
-        # The hook registers the repair DURING the call; ocr() clears the list
-        # first, so pre-populating it would prove nothing.
-        w._repairs.append({"recovered": 4})
+        # The hook registers the repair DURING the call, into the calling
+        # THREAD's list; ocr() starts a fresh one, so pre-populating would prove
+        # nothing.
+        W._trace_repairs().append({"recovered": 4})
         return [{"blocks": [{"html": "<p>società</p>",
                              "polygon": [[0, 0], [1, 0], [1, 1], [0, 1]],
                              "confidence": 0.96, "reading_order": 0}]}]
@@ -208,9 +209,21 @@ def test_repairs_are_cleared_between_pages(monkeypatch):
     monkeypatch.setattr(w, "_predict", lambda img: [
         {"blocks": [{"html": "<p>ok</p>", "polygon": [[0, 0], [1, 0], [1, 1], [0, 1]],
                      "confidence": 0.9, "reading_order": 0}]}])
-    w._repairs.append({"recovered": 1})
-    w.ocr(Image.new("RGB", (10, 10)))
-    assert w._repairs == []
+    pages = []
+
+    def _repairing(img):
+        if not pages:
+            W._trace_repairs().append({"recovered": 1})
+        pages.append(1)
+        return [{"blocks": [{"html": "<p>ok</p>",
+                             "polygon": [[0, 0], [1, 0], [1, 1], [0, 1]],
+                             "confidence": 0.9, "reading_order": 0}]}]
+    monkeypatch.setattr(w, "_predict", _repairing)
+    first = w.ocr(Image.new("RGB", (10, 10)))
+    second = w.ocr(Image.new("RGB", (10, 10)))
+    assert "accents_repaired" in first
+    assert "accents_repaired" not in second       # page 2 does not inherit it
+    assert W._trace_repairs() == []
 
 
 # ---------------------------------------------------------------- and the wiring #

tests/test_ocr_token_trace.py

@@ -197,10 +197,14 @@ def test_a_clean_page_reports_nothing(monkeypatch):
     """The whole response shape has to stay byte-identical when nothing is wrong."""
     w = W.SuryaWorker({"token_trace": True})
     w._mode = "predictor"
-    w._calls.append([_tok("Ã", [195])])          # an anomaly is present...
-    monkeypatch.setattr(w, "_predict", lambda img: [
-        {"blocks": [{"html": "<p>tutto bene</p>", "polygon": [[0, 0], [1, 0], [1, 1], [0, 1]],
-                     "confidence": 0.9, "reading_order": 0}]}])
+
+    def _predict(img):
+        # An anomaly IS present in what the model sent...
+        W._trace_calls().append([_tok("Ã", [195])])
+        return [{"blocks": [{"html": "<p>tutto bene</p>",
+                             "polygon": [[0, 0], [1, 0], [1, 1], [0, 1]],
+                             "confidence": 0.9, "reading_order": 0}]}]
+    monkeypatch.setattr(w, "_predict", _predict)
     from PIL import Image
     page = w.ocr(Image.new("RGB", (10, 10)))
     # ...but the text has no marker, so there is nothing to diagnose.
@@ -213,8 +217,10 @@ def test_an_unreadable_page_carries_the_diagnosis(monkeypatch):
     from PIL import Image
 
     def _predict(img):
-        # The hook would have filled this during the real call.
-        w._calls.append([_tok("p", [112]), _tok("i", [105]), _tok("Ã", [195])])
+        # The hook would have filled this during the real call. It is the
+        # calling THREAD's trace, not the worker's: one process now serves
+        # several pages at once (lanes).
+        W._trace_calls().append([_tok("p", [112]), _tok("i", [105]), _tok("Ã", [195])])
         return [{"blocks": [{"html": "<p>pi�</p>",
                              "polygon": [[0, 0], [1, 0], [1, 1], [0, 1]],
                              "confidence": 0.4, "reading_order": 0}]}]
@@ -230,12 +236,22 @@ def test_the_trace_is_cleared_between_pages(monkeypatch):
     w = W.SuryaWorker({"token_trace": True})
     w._mode = "predictor"
     from PIL import Image
-    monkeypatch.setattr(w, "_predict", lambda img: [
-        {"blocks": [{"html": "<p>ok</p>", "polygon": [[0, 0], [1, 0], [1, 1], [0, 1]],
-                     "confidence": 0.9, "reading_order": 0}]}])
-    w._calls.append([_tok("Ã", [195])])
-    w.ocr(Image.new("RGB", (10, 10)))
-    assert w._calls == []
+    pages = []
+
+    def _predict(img):
+        # Only the first page's model output is anomalous.
+        if not pages:
+            W._trace_calls().append([_tok("Ã", [195])])
+        pages.append(1)
+        return [{"blocks": [{"html": "<p>pi\ufffd</p>",
+                             "polygon": [[0, 0], [1, 0], [1, 1], [0, 1]],
+                             "confidence": 0.4, "reading_order": 0}]}]
+    monkeypatch.setattr(w, "_predict", _predict)
+    first = w.ocr(Image.new("RGB", (10, 10)))
+    second = w.ocr(Image.new("RGB", (10, 10)))
+    assert first["unreadable_trace"]["findings"]          # page 1 diagnosed
+    assert not second["unreadable_trace"]["findings"]     # page 2 carries none of it
+    assert W._trace_calls() == []                         # and the page left it clean
 
 
 def test_analysis_failure_is_reported_not_raised(monkeypatch):

tests/test_ocr_worker_lanes.py

+"""Pages in flight must stop being the process count.
+
+The OCR worker protocol was one request, one reply, strictly in order — so the
+only way to read two pages at once was to run two processes, and a process costs
+~600 MB of imported torch. `surya_instances: 64` therefore meant 64 processes:
+on 2026-10-12 that was 37 GB of RSS, swap exhausted and load 25 on a 16-core
+host, and a batch that gave one 19-second burst and then stalled for nine
+minutes with the rented GPU idle at 2% KV-cache usage.
+
+In the served modes that memory buys nothing — the recognition VLM is on the pod,
+the predictor holds no weights, and the process rasterises a page and then waits
+on a socket. So a lane is a worker OBJECT inside the one process, and these tests
+pin the protocol that makes that safe: ids, out-of-order replies, one page per
+lane object, and a trace that stays with its own page.
+"""
+
+import json
+import os
+import pathlib
+import sys
+import threading
+import time
+
+import pytest
+
+ROOT = pathlib.Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT))
+
+from codai.ocr.base import OcrError                              # noqa: E402
+from codai.ocr.subprocess_engine import SubprocessOcrEngine      # noqa: E402
+
+WORKER = ROOT / "codai" / "ocr" / "workers" / "ocr_worker.py"
+#: Runs the real protocol with a stubbed engine — see its docstring.
+FAKE = ROOT / "tests" / "ocr_lane_stub_worker.py"
+
+
+#: A real 1x1 PNG: the worker decodes what it is sent before handing it on, so
+#: a plausible-looking header is not enough.
+def _png_bytes():
+    import io
+    from PIL import Image
+    buf = io.BytesIO()
+    Image.new("RGB", (1, 1), (255, 255, 255)).save(buf, "PNG")
+    return buf.getvalue()
+
+
+class _Img:
+    """The minimum of a PIL image the engine touches before sending it."""
+
+    def convert(self, mode):
+        return self
+
+    def save(self, buf, fmt):
+        buf.write(_png_bytes())
+
+
+class _Engine(SubprocessOcrEngine):
+    """The real engine, driving the real worker in this interpreter's venv."""
+
+    name = "surya"
+    _worker_engine = "surya"
+
+    def __init__(self, lanes):
+        super().__init__(cfg=None)
+        self._lanes = lanes
+
+    def lanes(self):
+        return self._lanes
+
+    def _venv_dir(self):
+        return "unused"
+
+    def _requirements(self):
+        return ""
+
+    def _venv_python(self):
+        return sys.executable
+
+    def _ensure_venv(self):
+        return sys.executable
+
+    def _worker_opts(self):
+        return {}
+
+    def _worker_env(self):
+        return {"OCR_WORKER_PATH": str(WORKER)}
+
+
+@pytest.fixture
+def engine(monkeypatch):
+    """A live worker subprocess, torn down with the test."""
+    import codai.ocr.subprocess_engine as se
+    monkeypatch.setattr(se, "_WORKER", str(FAKE))
+    made = []
+
+    def _make(lanes):
+        eng = _Engine(lanes)
+        eng.load()
+        made.append(eng)
+        return eng
+
+    yield _make
+    for eng in made:
+        eng.cleanup()
+
+
+def test_four_lanes_read_four_pages_at_once(engine):
+    """The whole point: four pages of 0.25s each take about a quarter of a
+    second, not a second. Sequentially this is the same code path."""
+    eng = engine(4)
+    assert eng._mux is True
+    out = {}
+
+    def _one(i):
+        out[i] = eng.recognize_image(_Img()).text
+
+    started = time.time()
+    threads = [threading.Thread(target=_one, args=(i,)) for i in range(4)]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join(30)
+    took = time.time() - started
+
+    assert len(out) == 4                      # every page answered
+    assert took < 0.75                        # overlapped, not queued (4 x 0.25s)
+    # Four pages at once means four lane objects, each used by one page.
+    assert len(set(out.values())) == 4
+
+
+def test_one_lane_still_speaks_the_old_protocol(engine):
+    """A single-lane engine sends no ids and reads replies in order, exactly as
+    before — paddle, docTR and surya-local are driven this way."""
+    eng = engine(1)
+    assert eng._mux is False
+    assert eng.recognize_image(_Img()).text.startswith("lane")
+
+
+def test_a_lane_is_built_only_when_a_page_needs_it(engine):
+    """A quiet install pays for one. The 37 GB came from building for the peak."""
+    eng = engine(8)
+    first = eng.recognize_image(_Img()).text
+    second = eng.recognize_image(_Img()).text
+    assert first == second == "lane1"          # reused, never a second lane
+
+
+def test_pages_beyond_the_lane_count_wait_rather_than_pile_up(engine):
+    """Lanes are a bound, not a hint: six pages over two lanes is three rounds."""
+    eng = engine(2)
+    out = []
+    lock = threading.Lock()
+
+    def _one():
+        page = eng.recognize_image(_Img())
+        with lock:
+            out.append(page.text)
+
+    threads = [threading.Thread(target=_one) for _ in range(6)]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join(30)
+    assert len(out) == 6
+    assert len(set(out)) <= 2                  # never more lanes than allowed
+
+
+def test_a_reply_goes_to_the_page_that_asked_for_it(engine):
+    """Attributing one page's text to another request is the one failure worse
+    than losing the page, so each request carries an id and the reader matches
+    on it. The stub answers with its lane, so a crossed reply would show up as
+    two pages claiming the same lane while both were in flight."""
+    eng = engine(3)
+    seen = {}
+
+    def _one(i):
+        seen[i] = eng.recognize_image(_Img()).text
+
+    threads = [threading.Thread(target=_one, args=(i,)) for i in range(3)]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join(30)
+    assert len(seen) == 3
+    assert len(set(seen.values())) == 3
+
+
+def test_a_dead_worker_fails_the_pages_instead_of_hanging(engine):
+    """With no reply ever coming, a page used to block on readline for ever. The
+    wait is broken up so a worker that died is noticed."""
+    eng = engine(4)
+    eng.recognize_image(_Img())            # prove it works first
+    eng._proc.kill()
+    eng._proc.wait(timeout=10)
+    with pytest.raises(OcrError) as got:
+        eng.recognize_image(_Img())
+    assert "worker" in str(got.value)
+
+
+def test_killing_the_engine_wakes_everything_waiting(engine):
+    """cleanup() during a batch must not leave threads parked on an Event."""
+    eng = engine(2)
+    errs = []
+
+    def _one():
+        try:
+            eng.recognize_image(_Img())
+        except OcrError as e:
+            errs.append(str(e))
+
+    threads = [threading.Thread(target=_one) for _ in range(4)]
+    for t in threads:
+        t.start()
+    time.sleep(0.05)
+    eng.cleanup()
+    for t in threads:
+        t.join(20)
+        assert not t.is_alive()
+
+
+# ----------------------------------------------------- the trace is per page #
+def test_the_token_trace_is_per_thread_not_per_worker():
+    """surya's hook is module-level: one `_generate_one` per process. The sink
+    used to belong to whichever worker object installed it last, which with
+    lanes would file one page's logprobs — and its accent repairs — under
+    another page."""
+    import importlib.util
+    spec = importlib.util.spec_from_file_location("ocr_worker_mod", str(WORKER))
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+
+    seen = {}
+
+    def _in_thread(name):
+        calls, repairs = mod._trace_begin()
+        calls.append(name)
+        time.sleep(0.05)                   # overlap with the other thread
+        seen[name] = list(mod._trace_calls())
+
+    a = threading.Thread(target=_in_thread, args=("A",))
+    b = threading.Thread(target=_in_thread, args=("B",))
+    a.start(); b.start(); a.join(10); b.join(10)
+
+    assert seen["A"] == ["A"]              # no cross-talk
+    assert seen["B"] == ["B"]
+
+
+def test_lanes_default_to_one_for_an_engine_that_has_not_opted_in():
+    """Concurrency is opt-in per engine: driving one that is not safe for it
+    concurrently would corrupt pages rather than slow them down."""
+    from codai.ocr.base import OcrEngine
+
+    class _Plain(OcrEngine):
+        def load(self):
+            pass
+
+        def recognize_image(self, image):
+            raise NotImplementedError
+
+    assert _Plain(cfg=None).lanes() == 1
+
+
+def test_surya_takes_lanes_from_config_in_served_mode_only():
+    from codai.ocr.surya import SuryaEngine
+
+    class _Cfg:
+        surya_serve = "model"
+        surya_lanes = 6
+        surya_langs = "it"
+
+    eng = SuryaEngine.__new__(SuryaEngine)
+    eng.cfg = _Cfg()
+    assert eng.lanes() == 6
+
+    # Local mode: a lane would be a second copy of the model on the card.
+    eng.cfg.surya_serve = "local"
+    assert eng.lanes() == 1
+
+
+def test_the_pool_offers_an_instance_once_per_lane():
+    """What turns lanes into concurrency: the queue is the bound, so a 4-lane
+    instance has to be in it four times."""
+    import asyncio
+    from codai.ocr import manager
+
+    class _Eng:
+        def __init__(self):
+            self.loaded = 0
+
+        def lanes(self):
+            return 4
+
+        def ensure_loaded(self):
+            self.loaded += 1
+
+        def prelaunch_vram_gb(self):
+            return 0.0
+
+        def vram_gb(self):
+            return 0.0
+
+    made = []
+
+    def _factory(cfg):
+        eng = _Eng()
+        made.append(eng)
+        return eng
+
+    manager._ENGINE_FACTORIES["_lanetest"] = _factory
+    try:
+        pool = manager._Pool("_lanetest", cfg=None, size=2)
+        asyncio.run(pool._ensure_built())
+        assert len(made) == 2                      # two processes
+        assert pool._q.qsize() == 8                # eight pages at once
+    finally:
+        manager._ENGINE_FACTORIES.pop("_lanetest", None)