ocr: the batch was not stalled, it was re-reading pages (v0.3.45)

Read from the pod's own drained log, ten minutes of batch OCR on 2026-10-13:

  Running: 15-19 reqs, Waiting: 0 reqs, GPU KV cache usage: 5-7%
  Avg generation throughput: ~1,900-2,100 tokens/s, every minute
  524 x "POST /v1/chat/completions" 200 OK, no aborts
  2,644 generated tokens PER REQUEST (second pod: 4,010)
  MM cache hit rate: 25-27%

Three things in those numbers, and only the third is a defect.

The GPU was never the bottleneck: 5-7% of the KV cache and `Waiting: 0` for
ten solid minutes. So a card to itself cannot be the fix, and the co-loaded
pyannote cost nothing measurable — the limit was never VRAM or batch slots.

Oversubscription converts throughput into latency, not into stalling. 64
pages into max_pods x scale_up_inflight_per_pod = 32 slots queues correctly,
but every page's latency rises by the queue, and a client counting finished
DOCUMENTS sees nothing until a document's LAST page lands: spread 64 pages
over sixteen documents and all sixteen advance while none completes. The
pool now says so at build time instead of leaving it to be inferred.

The defect: 2,644 generated tokens for a page whose text is ~400. surya
re-runs a page whose generation repeated itself (`_should_retry`) up to three
more times, each at a rising temperature and each able to reach its token
cap — four full generations to arrive at the same unreadable answer — and its
repeat detector cannot tell "the model broke" from "this page really does
repeat", which a tax table does. The 25-27% multimodal cache hit rate is the
direct evidence: a quarter of all requests re-sent an image the pod had
already read. None of it was visible: a page generated four times over and a
page that was merely slow look identical from outside.

`max_retries` is unreachable from surya's settings — the vLLM backend calls
chat_completions_batch without it, so the default 3 was the only value there
has ever been. So the decision is hooked, beside the token-trace hook already
in that module, and `ocr.surya_max_retries` is a CEILING (asking for 10 where
surya allows 3 gets 3): 1 by default, 0 to disable, -1 to hand it back.

Every retry is logged, and a page that needed more than one call carries
`model_attempts` in its result — always, not only under surya_token_trace. A
page read first time stays byte-identical to what it always returned. The
counter is per THREAD, like the trace, or one lane would be blamed for
another lane's retries.

Also pins SURYA_INFERENCE_PARALLEL=1 in the served modes: a lane is our unit
of concurrency, and left unset surya sizes its own thread pool from a
built-in GPU table and fans out inside the worker, so instances x lanes would
stop being the real number of pages in flight.

19 tests in tests/test_surya_retry_budget.py; settings carry the new knob.
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.44"
+__version__ = "0.3.45"
 __tagline__ = "Complete orchestration, distribution and escalation of remotizable advanced inference."
 
 # Configure the CUDA caching allocator BEFORE torch is imported anywhere.

codai/admin/routes.py

@@ -4139,6 +4139,7 @@ def build_settings_dict(c, gpu_cards):
             "surya_accept_license": c.ocr.surya_accept_license,
             "surya_instances": c.ocr.surya_instances,
             "surya_lanes": getattr(c.ocr, "surya_lanes", 4),
+            "surya_max_retries": getattr(c.ocr, "surya_max_retries", 1),
             "surya_langs": c.ocr.surya_langs,
             "surya_token_trace": c.ocr.surya_token_trace,
             "surya_repair_accents": c.ocr.surya_repair_accents,
@@ -4963,6 +4964,10 @@ async def api_save_settings(request: Request, username: str = Depends(require_ad
         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)
+        # -1 is meaningful here ("whatever surya decides"), so the floor is -1.
+        if "surya_max_retries" in d:
+            o.surya_max_retries = _i("surya_max_retries",
+                                     getattr(o, "surya_max_retries", 1), -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

@@ -1205,6 +1205,7 @@
       <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" 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>
+      <div class="form-group" style="flex:1;min-width:130px"><label class="form-label" title="How many times Surya may RE-run a page whose generation repeated itself. Surya's own number is 3 and each retry is another FULL generation at a higher temperature: a batch measured on 2026-10-13 spent 2,644 generated tokens per page on text worth ~400, with a quarter of all requests re-sending an image the pod had already read. 1 keeps the rescue a retry does buy and drops the long tail; 0 disables it; -1 means whatever Surya decides. A retried page is reported as model_attempts in the result.">Retries / page</label><input type="number" id="s-ocr-surya-max-retries" class="form-input" min="-1" placeholder="1"></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>
@@ -2204,6 +2205,7 @@ async function loadSettings(){
     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-max-retries').value = ocr.surya_max_retries ?? 1;
     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;
@@ -2490,6 +2492,9 @@ async function saveSettings(silent){
       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,
+      // 0 and -1 are both meaningful, so an explicit NaN check, not `|| 1`.
+      surya_max_retries: (function(v){ v = parseInt(v); return isNaN(v) ? 1 : v; })(
+        document.getElementById('s-ocr-surya-max-retries').value),
       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

@@ -1038,6 +1038,19 @@ class OcrConfig:
     #: 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
+    #: How many times surya may RE-run a page whose generation repeated itself.
+    #: surya's own number is 3, and it is unreachable from its settings (the vLLM
+    #: backend never passes `max_retries`), so this is a ceiling we apply by
+    #: hooking the decision. Each retry is another full generation at a higher
+    #: temperature: measured on the Digesta orchestrator on 2026-10-13, a batch
+    #: ran at 2,644 generated tokens per request against pages whose text is
+    #: ~400, with a 25-27% multimodal cache hit rate on the pod — a quarter of
+    #: all requests were re-sends of an image it had already read. The GPU was
+    #: never the limit (18 concurrent, `Waiting: 0`, 5-7% KV cache); the tokens
+    #: were being spent re-reading pages the model had already failed to read.
+    #: 1 keeps the rescue a retry does buy and drops the long tail; 0 disables
+    #: it; -1 means "whatever surya decides".
+    surya_max_retries: int = 1
     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
@@ -1788,6 +1801,7 @@ class ConfigManager:
                 "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_max_retries": getattr(self.config.ocr, "surya_max_retries", 1),
                 "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/manager.py

@@ -276,6 +276,66 @@ def _engine_instances(cfg, name: str) -> int:
     return want
 
 
+def _fleet_page_slots(name: str):
+    """How many pages the remote fleet can actually have in flight, or None.
+
+    ``max_pods x scale_up_inflight_per_pod``. Not a limit we impose — the pool
+    queues the excess perfectly correctly — but a number worth comparing with
+    ours, because pages beyond it do not go faster, they only wait, and a pod
+    takes ~12 minutes to appear so the fleet cannot grow into a burst.
+    """
+    try:
+        from codai.models.manager import _model_entry_for
+        from codai.api.runpod_worker import parse_model_runpod
+    except Exception:
+        return None
+    for served in _SERVED_NAMES.get(name, ()):
+        try:
+            entry = _model_entry_for(served)
+            block = (entry or {}).get("runpod")
+            if not isinstance(block, dict) or not block:
+                continue
+            mcfg = parse_model_runpod(block)
+            if (mcfg.mode or "").lower() != "pods":
+                continue
+            pods = max(1, int(getattr(mcfg, "max_pods", 1) or 1))
+            per = int(getattr(mcfg, "scale_up_inflight_per_pod", 0) or 0)
+            if per <= 0:
+                per = int(getattr(mcfg, "max_inflight_per_pod", 0) or 0)
+            if per > 0:
+                return pods * per
+        except Exception:
+            continue
+    return None
+
+
+#: The served model names an engine may be reading from, for the slot estimate.
+_SERVED_NAMES = {"surya": ("datalab-to/surya-ocr-2", "surya-ocr-2"),
+                 "chandra": ("datalab-to/chandra-ocr-2", "chandra-ocr-2"),
+                 "olmocr": ("allenai/olmOCR-2-7B-1025-FP8", "olmocr")}
+
+
+def _warn_if_oversubscribed(name: str, pages: int) -> None:
+    """Say so when we fan out more pages than the fleet can serve at once.
+
+    Oversubscription does not lose work and it does not go faster: the surplus
+    waits in the gateway, every page's latency rises by the queue, and because a
+    client usually counts finished DOCUMENTS, spreading N pages over many
+    documents at once means they all advance and none completes until near the
+    end. That reads from outside as "one opening burst, then silence" — which is
+    exactly how the 2026-10-13 batch was reported, while the pod sat at 18
+    concurrent requests, `Waiting: 0` and 5-7% KV cache the entire time.
+    """
+    slots = _fleet_page_slots(name)
+    if not slots or pages <= slots:
+        return
+    print(f"[ocr] engine {name!r}: {pages} pages at once, but the fleet serves "
+          f"{slots} (max_pods x scale_up_inflight_per_pod). The surplus waits in "
+          f"the gateway — it does not go faster, and a pod takes ~12 min to "
+          f"appear, so a burst cannot be scaled into. Lower instances/lanes, or "
+          f"raise the pod's inflight ceiling.", flush=True)
+
+
 def _engine_lanes(eng) -> int:
     """How many pages one instance of ``eng`` serves at once, asked politely.
 
@@ -458,6 +518,7 @@ class _Pool:
                       f"{lanes} lane(s) each — {self.size * lanes} pages at once")
             else:
                 print(f"[ocr] engine '{self.name}': {self.size} instance(s) ready")
+            _warn_if_oversubscribed(self.name, self.size * lanes)
 
     def release_sync(self) -> float:
         """Tear down all instances (free VRAM). SYNC — safe to call from the model

codai/ocr/surya.py

@@ -71,7 +71,11 @@ class SuryaEngine(SubprocessOcrEngine):
                 # patches surya's backend, which has to happen before the first page
                 # rather than per request.
                 "token_trace": bool(getattr(self.cfg, "surya_token_trace", False)),
-                "repair_accents": bool(getattr(self.cfg, "surya_repair_accents", True))}
+                "repair_accents": bool(getattr(self.cfg, "surya_repair_accents", True)),
+                # None hands the decision back to surya; anything else is a
+                # ceiling on how often it may re-run a repeating page.
+                "max_retries": (None if int(getattr(self.cfg, "surya_max_retries", 1)) < 0
+                                else int(getattr(self.cfg, "surya_max_retries", 1)))}
 
     #: Modes where the recognition VLM lives in a model server we only talk to.
     SERVED_MODES = ("model", "vllm", "llamacpp", "server")
@@ -132,6 +136,13 @@ class SuryaEngine(SubprocessOcrEngine):
             "SURYA_INFERENCE_BACKEND": "llamacpp" if mode == "llamacpp" else "vllm",
             "SURYA_INFERENCE_URL": url,
         }
+        # One page per call, because a LANE is our unit of concurrency. Left
+        # unset, surya sizes its own thread pool from a built-in GPU table
+        # (min(max_num_seqs, MAX_AUTO_PARALLEL)) and fans out inside the worker
+        # — so instances x lanes would stop being the real number of pages in
+        # flight the moment a call carried more than one page. Pinning it keeps
+        # the arithmetic in the admin UI true.
+        env["SURYA_INFERENCE_PARALLEL"] = "1"
         if self._served_name:
             env["SURYA_MODEL_CHECKPOINT"] = self._served_name
         if self._server_key:

codai/ocr/workers/ocr_worker.py

@@ -47,11 +47,17 @@ def _emit(obj):
 #: and `surya_repair_accents` would rewrite the wrong text.
 _TRACE = threading.local()
 
+#: How many RE-tries of a repeating page coderai allows, or None for surya's
+#: own number. One cell, not a per-worker attribute: the hook it feeds is
+#: module-level, so every lane in this process shares it.
+_RETRY_BUDGET = [None]
+
 
 def _trace_begin():
     """Start a page's trace. Returns its (calls, repairs) — this thread's own."""
     _TRACE.calls = []
     _TRACE.repairs = []
+    _TRACE.attempts = 0
     return _TRACE.calls, _TRACE.repairs
 
 
@@ -59,6 +65,26 @@ def _trace_calls():
     return getattr(_TRACE, "calls", None)
 
 
+def _attempts_bump():
+    """One more call to the model for THIS page. Returns the new count.
+
+    surya retries a page whose generation repeated itself, up to three times at
+    rising temperature, and each attempt may run to its token cap. Measured on
+    the Digesta orchestrator on 2026-10-13: 2,644 generated tokens per request
+    against a page whose text is ~400, and a 25-27% multimodal cache hit rate on
+    the pod — i.e. a quarter of all requests were re-sends of an image it had
+    already read. None of it was visible from outside: a retried page looks
+    exactly like a slow one.
+    """
+    n = getattr(_TRACE, "attempts", 0) + 1
+    _TRACE.attempts = n
+    return n
+
+
+def _attempts():
+    return getattr(_TRACE, "attempts", 0)
+
+
 def _trace_repairs():
     return getattr(_TRACE, "repairs", None)
 
@@ -680,6 +706,10 @@ class SuryaWorker:
         #: 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))
+        #: Our ceiling on surya's own page retries (None = surya's 3). Written to
+        #: the module cell because the hook that reads it is module-level.
+        budget = opts.get("max_retries", None)
+        _RETRY_BUDGET[0] = None if budget is None else max(0, int(budget))
         #: 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.
@@ -749,6 +779,7 @@ class SuryaWorker:
         def _wrapped(*args, **kwargs):
             result = inner(*args, **kwargs)
             try:
+                _attempts_bump()
                 sink = _trace_calls()
                 if sink is not None:
                     sink.append(getattr(result, "logprobs", None))
@@ -771,6 +802,33 @@ class SuryaWorker:
             return result
 
         _oc._generate_one = _wrapped
+
+        # surya decides on its own to re-run a page whose generation repeated
+        # itself: up to `max_retries` (3) more calls, each at a higher
+        # temperature, each able to run to its token cap. On a page the model
+        # genuinely cannot read that is 4 x 2048 tokens of GPU time to arrive at
+        # the same unreadable answer — and the repeat detector cannot tell
+        # "repeated because the model broke" from "repeated because the page
+        # really does say the same thing twenty times", which a tax table does.
+        #
+        # The retry stays (it does rescue pages), but the budget is ours and the
+        # spend is logged. `max_retries` is not reachable from surya's settings:
+        # the vLLM backend never passes it, so the default was the only value
+        # available. Hooking the decision is the only seam there is.
+        inner_retry = getattr(_oc, "_should_retry", None)
+        if inner_retry is not None:
+            def _wrapped_retry(result, retries, max_retries, *a, **kw):
+                budget = _RETRY_BUDGET[0]
+                if budget is not None:
+                    max_retries = min(int(max_retries or 0), int(budget))
+                again = inner_retry(result, retries, max_retries, *a, **kw)
+                if again:
+                    print(f"[surya] page repeated itself — retry "
+                          f"{retries + 1}/{max_retries} (temperature rises; a "
+                          f"retried page costs another full generation)",
+                          file=sys.stderr, flush=True)
+                return again
+            _oc._should_retry = _wrapped_retry
         _oc._coderai_traced = True
 
     def _load_legacy(self):
@@ -867,6 +925,19 @@ class SuryaWorker:
                 "how": ("rebuilt from the model's own per-token bytes, which "
                         "decoded cleanly where message.content did not"),
             }
+        # A page the model had to be asked for more than once. Reported always,
+        # not only under token_trace: this is the difference between a page that
+        # was slow and a page that was generated two, three or four times over,
+        # and without it the only visible symptom is a batch that produces far
+        # fewer completions than the GPU's throughput says it should.
+        tries = _attempts()
+        if tries > 1:
+            page["model_attempts"] = {
+                "calls": tries,
+                "why": ("surya re-ran this page because the generation repeated "
+                        "itself; each re-run is a full generation at a higher "
+                        "temperature"),
+            }
         if self._trace and UNREADABLE in page["text"]:
             try:
                 page["unreadable_trace"] = _trace_report(calls)

docs/ocr.md

@@ -1005,7 +1005,91 @@ 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.
 
-#### Lanes: concurrency without another process
+#### A batch can be GPU-idle and still look stalled
+
+Measured on the Digesta orchestrator on 2026-10-13, ten minutes of batch OCR at
+four client workers, read from the pod's own drained log
+(`pod-logs/<pod_id>.log`):
+
+```
+Running: 15-19 reqs, Waiting: 0 reqs, GPU KV cache usage: 5-7%
+Avg generation throughput: ~1,900-2,100 tokens/s, every minute, for ten minutes
+524 x "POST /v1/chat/completions HTTP/1.1" 200 OK, no aborts
+2,644 generated tokens PER REQUEST (second pod: 4,010)
+MM cache hit rate: 25-27%
+```
+
+The client reported a stall: one opening burst, then ~0 completions. Three
+separate things are in those numbers, and only the last is a defect.
+
+**The GPU was never the bottleneck.** 5-7% of the KV cache, `Waiting: 0` for ten
+minutes. The card could have taken several times the load. So giving Surya a
+card to itself does not help, and a co-loaded model on the same pod (pyannote,
+in that run) costs nothing measurable — the limit was never VRAM or batch slots.
+Check `Waiting:` and `KV cache usage:` in the pod log before moving anything.
+
+**Oversubscription converts throughput into latency, not into stalling.** The
+pool queues pages beyond `max_pods x scale_up_inflight_per_pod` and serves them
+perfectly correctly, but every page's latency rises by the queue — and a client
+that counts finished DOCUMENTS sees nothing at all until the first document's
+LAST page lands. Spread 64 pages over sixteen documents and all sixteen advance
+while none completes. That reads exactly like a stall. The pool logs the
+mismatch at build time:
+
+```
+[ocr] engine 'surya': 64 pages at once, but the fleet serves 32
+      (max_pods x scale_up_inflight_per_pod). The surplus waits in the gateway —
+      it does not go faster, and a pod takes ~12 min to appear, so a burst
+      cannot be scaled into.
+```
+
+**The defect: 2,644 generated tokens for a page whose text is ~400.** Surya
+re-runs a page whose generation repeated itself — `_should_retry` in
+`surya.inference.backends.openai_client` — up to `max_retries` (3) more times,
+each at a higher temperature and each able to run to its token cap. On a page
+the model genuinely cannot read that is four full generations to reach the same
+unreadable answer, and the repeat detector cannot distinguish "the model broke"
+from "this page really does repeat", which a tax table does. The 25-27%
+multimodal cache hit rate is the direct evidence: a quarter of all requests were
+re-sends of an image the pod had already read.
+
+`max_retries` is unreachable from Surya's settings — its vLLM backend calls
+`chat_completions_batch` without it, so the default was the only value
+available. CoderAI therefore hooks the decision, beside the token-trace hook it
+already installs in that module, and `ocr.surya_max_retries` is the ceiling:
+
+| value | meaning |
+|---|---|
+| `1` (default) | one re-run — keeps the rescue a retry does buy, drops the long tail |
+| `0` | no re-runs; a repeating page is returned as it came |
+| `-1` | whatever Surya decides (its own 3) |
+| `n` | at most `n`, and never more than Surya would have allowed |
+
+It is a ceiling, not a setting: asking for 10 where Surya allows 3 gets 3.
+
+Every retry is logged, and a page that needed more than one call carries it in
+the result — always, not only under `surya_token_trace`:
+
+```json
+"model_attempts": {"calls": 3, "why": "surya re-ran this page because the
+                   generation repeated itself; each re-run is a full generation
+                   at a higher temperature"}
+```
+
+A page read on the first attempt stays byte-identical to what it has always
+returned. Without this, a page generated four times over and a page that was
+merely slow look the same from outside — which is why ten minutes of this was
+invisible.
+
+CoderAI also pins `SURYA_INFERENCE_PARALLEL=1` in the served modes. A **lane**
+is coderai's unit of concurrency; left unset, Surya sizes its own thread pool
+from a built-in GPU table and fans out inside the worker, so `instances x lanes`
+would stop being the real number of pages in flight the moment one call carried
+more than one page.
+
+---
+
+## 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

tests/test_surya_retry_budget.py

+"""The tokens a batch spends re-reading pages it has already failed to read.
+
+Measured on the Digesta orchestrator, 2026-10-13 11:20-11:30, from the pod's own
+drained log. Ten minutes of batch OCR at `lavoratori=4`:
+
+    Running: 15-19 reqs, Waiting: 0 reqs, GPU KV cache usage: 5-7%
+    Avg generation throughput: ~1,900-2,100 tokens/s, sustained
+    524 x "POST /v1/chat/completions HTTP/1.1" 200 OK, no aborts
+    2,644 generated tokens PER REQUEST (pod 2: 4,010)
+    MM cache hit rate: 25-27%
+
+Three things follow, and only the third is a defect:
+
+1. The GPU was never the bottleneck. It held 5-7% of its KV cache and queued
+   nothing for ten minutes — it could have taken several times the load. So
+   "give Surya an A40 to itself" cannot be the fix: the card was ~95% idle on
+   the dimension that limits a VLM, and the co-loaded pyannote cost nothing.
+2. The pipeline never stalled. ~100,000 generated tokens per minute, every
+   minute, with ~18 pages in flight.
+3. It spent 2,644 generated tokens on pages whose text is ~400, and a quarter of
+   its requests were images the pod had already read. That is surya's own
+   `_should_retry`: a generation that repeats itself is re-run up to three more
+   times at a rising temperature, each run able to reach max_tokens. On a page
+   the model genuinely cannot read, that is four full generations to arrive at
+   the same unreadable answer — and nothing anywhere said so.
+
+`max_retries` is not reachable from surya's settings: its vLLM backend calls
+`chat_completions_batch` without it, so the default 3 was the only value
+available. Hooking the decision — in the same module, beside the token-trace
+hook that is already there — is the only seam.
+"""
+
+import pathlib
+import sys
+import threading
+import types
+
+ROOT = pathlib.Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT))
+
+import importlib.util                                                # noqa: E402
+
+SRC = ROOT / "codai/ocr/workers/ocr_worker.py"
+
+
+def _worker_module():
+    """The real worker module, loaded by path (it is not importable as a package)."""
+    spec = importlib.util.spec_from_file_location("ocr_worker_under_test", SRC)
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+    return mod
+
+
+class _Result:
+    def __init__(self, raw="ciao", logprobs=None):
+        self.raw = raw
+        self.token_count = 4
+        self.logprobs = logprobs
+        self.error = False
+
+
+def _fake_surya(monkeypatch, mod, retry_decision=True):
+    """A stand-in for surya.inference.backends.openai_client."""
+    oc = types.ModuleType("surya.inference.backends.openai_client")
+    oc.calls = []
+    oc.retry_asked = []
+
+    def _generate_one(*a, **k):
+        oc.calls.append(1)
+        return _Result()
+
+    def _should_retry(result, retries, max_retries, *a, **k):
+        oc.retry_asked.append((retries, max_retries))
+        return bool(retry_decision) and retries < max_retries
+
+    oc._generate_one = _generate_one
+    oc._should_retry = _should_retry
+    pkg = types.ModuleType("surya.inference.backends")
+    pkg.openai_client = oc
+    monkeypatch.setitem(sys.modules, "surya", types.ModuleType("surya"))
+    monkeypatch.setitem(sys.modules, "surya.inference", types.ModuleType("surya.inference"))
+    monkeypatch.setitem(sys.modules, "surya.inference.backends", pkg)
+    monkeypatch.setitem(sys.modules, "surya.inference.backends.openai_client", oc)
+    return oc
+
+
+def _worker(mod, **opts):
+    w = mod.SuryaWorker.__new__(mod.SuryaWorker)
+    o = {"langs": "it", "token_trace": False, "repair_accents": False}
+    o.update(opts)
+    mod.SuryaWorker.__init__(w, o)
+    return w
+
+
+# ------------------------------------------------------------- the ceiling #
+def test_our_budget_clamps_suryas_own_number(monkeypatch):
+    """surya asks for 3. We say 1, and 1 is what the decision sees."""
+    mod = _worker_module()
+    oc = _fake_surya(monkeypatch, mod)
+    w = _worker(mod, max_retries=1, token_trace=True)
+    w._install_trace_hook()
+    oc._should_retry(_Result(), 0, 3)
+    assert oc.retry_asked[-1] == (0, 1)
+
+
+def test_zero_disables_the_retry_entirely(monkeypatch):
+    mod = _worker_module()
+    oc = _fake_surya(monkeypatch, mod)
+    w = _worker(mod, max_retries=0, token_trace=True)
+    w._install_trace_hook()
+    assert oc._should_retry(_Result(), 0, 3) is False
+
+
+def test_none_hands_the_decision_back_to_surya(monkeypatch):
+    """-1 in the config means "whatever surya decides" — we must not silently
+    become the policy for an install that did not ask us to be."""
+    mod = _worker_module()
+    oc = _fake_surya(monkeypatch, mod)
+    w = _worker(mod, max_retries=None, token_trace=True)
+    w._install_trace_hook()
+    oc._should_retry(_Result(), 0, 3)
+    assert oc.retry_asked[-1] == (0, 3)
+
+
+def test_the_budget_never_raises_suryas_number(monkeypatch):
+    """A ceiling, not a setting: asking for 10 when surya allows 3 gets 3."""
+    mod = _worker_module()
+    oc = _fake_surya(monkeypatch, mod)
+    w = _worker(mod, max_retries=10, token_trace=True)
+    w._install_trace_hook()
+    oc._should_retry(_Result(), 0, 3)
+    assert oc.retry_asked[-1] == (0, 3)
+
+
+def test_a_retry_is_logged(capsys, monkeypatch):
+    """The whole reason this was invisible: a retried page looked like a slow
+    one, in a batch where slowness was expected."""
+    mod = _worker_module()
+    oc = _fake_surya(monkeypatch, mod)
+    w = _worker(mod, max_retries=2, token_trace=True)
+    w._install_trace_hook()
+    oc._should_retry(_Result(), 0, 3)
+    err = capsys.readouterr().err
+    assert "repeated itself" in err and "another full generation" in err
+
+
+# ------------------------------------------------------- counting attempts #
+def test_attempts_are_counted_per_page(monkeypatch):
+    mod = _worker_module()
+    oc = _fake_surya(monkeypatch, mod)
+    w = _worker(mod, token_trace=True)
+    w._install_trace_hook()
+    mod._trace_begin()
+    for _ in range(3):
+        oc._generate_one()
+    assert mod._attempts() == 3
+
+
+def test_each_page_starts_from_zero(monkeypatch):
+    """Page 2 must not inherit page 1's retries — that is what made the
+    per-worker trace wrong once lanes put several pages in one process."""
+    mod = _worker_module()
+    oc = _fake_surya(monkeypatch, mod)
+    w = _worker(mod, token_trace=True)
+    w._install_trace_hook()
+    mod._trace_begin()
+    oc._generate_one(); oc._generate_one()
+    assert mod._attempts() == 2
+    mod._trace_begin()
+    assert mod._attempts() == 0
+
+
+def test_attempts_are_per_thread(monkeypatch):
+    """Two lanes, two pages, two counts. A shared counter would blame one page
+    for another's retries."""
+    mod = _worker_module()
+    oc = _fake_surya(monkeypatch, mod)
+    w = _worker(mod, token_trace=True)
+    w._install_trace_hook()
+    seen = {}
+
+    def lane(name, n):
+        mod._trace_begin()
+        for _ in range(n):
+            oc._generate_one()
+        seen[name] = mod._attempts()
+
+    a = threading.Thread(target=lane, args=("a", 1))
+    b = threading.Thread(target=lane, args=("b", 4))
+    a.start(); b.start(); a.join(); b.join()
+    assert seen == {"a": 1, "b": 4}
+
+
+# --------------------------------------------------- it reaches the caller #
+def test_a_retried_page_says_so_in_its_result():
+    """Reported always, not only under token_trace: this is the difference
+    between a slow page and a page generated four times over."""
+    src = SRC.read_text()
+    assert '"model_attempts"' in src
+    blk = src.split("tries = _attempts()")[1][:400]
+    assert "if tries > 1" in blk          # a clean page stays byte-identical
+
+
+def test_a_clean_page_carries_no_attempt_block():
+    src = SRC.read_text()
+    i = src.index("tries = _attempts()")
+    assert "if tries > 1:" in src[i:i + 200]
+
+
+# -------------------------------------------- the engine passes it through #
+def test_the_engine_sends_the_budget_at_load():
+    from codai.ocr.surya import SuryaEngine
+    from codai.config import OcrConfig
+    cfg = OcrConfig()
+    cfg.surya_max_retries = 1
+    eng = SuryaEngine(cfg)
+    assert eng._worker_opts()["max_retries"] == 1
+    cfg.surya_max_retries = -1
+    assert SuryaEngine(cfg)._worker_opts()["max_retries"] is None
+
+
+def test_the_default_keeps_one_rescue_attempt():
+    """A retry does rescue pages; it is the long tail that was costing the
+    batch. 1 keeps the rescue and drops the tail."""
+    from codai.config import OcrConfig
+    assert OcrConfig().surya_max_retries == 1
+
+
+def test_suryas_own_fanout_is_pinned_to_one_page():
+    """A LANE is our unit of concurrency. Left unset, surya sizes its own thread
+    pool from a built-in GPU table and fans out inside the worker, so
+    instances x lanes would stop being the real number of pages in flight."""
+    from codai.ocr.surya import SuryaEngine
+    from codai.config import OcrConfig
+    cfg = OcrConfig()
+    cfg.surya_serve = "model"
+    eng = SuryaEngine(cfg)
+    eng._server_url = "https://pod:8000/v1"
+    eng._server_key = "k"
+    eng._served_name = "datalab-to/surya-ocr-2"
+    assert eng._worker_env()["SURYA_INFERENCE_PARALLEL"] == "1"
+
+
+def test_local_mode_gets_no_client_parallel():
+    """There is no HTTP client in local mode; the setting would be noise."""
+    from codai.ocr.surya import SuryaEngine
+    from codai.config import OcrConfig
+    cfg = OcrConfig()
+    cfg.surya_serve = "local"
+    assert "SURYA_INFERENCE_PARALLEL" not in SuryaEngine(cfg)._worker_env()
+
+
+# ------------------------------------------------- the oversubscription note #
+def test_more_pages_than_the_fleet_can_serve_is_reported(capsys, monkeypatch):
+    """64 pages into 2 pods x 16 slots does not go faster — it queues, every
+    page's latency rises, and a client counting DOCUMENTS sees one opening
+    burst then silence."""
+    from codai.ocr import manager as m
+    monkeypatch.setattr(m, "_fleet_page_slots", lambda name: 32)
+    m._warn_if_oversubscribed("surya", 64)
+    out = capsys.readouterr().out
+    assert "64 pages at once" in out and "fleet serves 32" in out
+
+
+def test_a_fleet_that_can_take_it_says_nothing(capsys, monkeypatch):
+    from codai.ocr import manager as m
+    monkeypatch.setattr(m, "_fleet_page_slots", lambda name: 64)
+    m._warn_if_oversubscribed("surya", 64)
+    assert capsys.readouterr().out == ""
+
+
+def test_an_unknown_fleet_says_nothing(capsys, monkeypatch):
+    """No RunPod config, a local engine, a lookup that failed — none of those
+    are oversubscription, and guessing would cry wolf on every build."""
+    from codai.ocr import manager as m
+    monkeypatch.setattr(m, "_fleet_page_slots", lambda name: None)
+    m._warn_if_oversubscribed("surya", 9999)
+    assert capsys.readouterr().out == ""
+
+
+def test_the_slot_estimate_reads_the_models_own_runpod_block(monkeypatch):
+    from codai.ocr import manager as m
+    import codai.models.manager as mm
+    monkeypatch.setattr(mm, "_model_entry_for",
+                        lambda name: {"runpod": {"mode": "pods", "max_pods": 4,
+                                                 "scale_up_inflight_per_pod": 16}},
+                        raising=False)
+    assert m._fleet_page_slots("surya") == 64
+
+
+def test_a_serverless_model_has_no_pod_slots(monkeypatch):
+    """The arithmetic is about pods; serverless scales on its own terms."""
+    from codai.ocr import manager as m
+    import codai.models.manager as mm
+    monkeypatch.setattr(mm, "_model_entry_for",
+                        lambda name: {"runpod": {"mode": "serverless",
+                                                 "max_pods": 4,
+                                                 "scale_up_inflight_per_pod": 16}},
+                        raising=False)
+    assert m._fleet_page_slots("surya") is None