ocr+runpod: a cut-off reply is not an unreadable page; a fleet is not one pod per boot (v0.3.47)

Three defects, found from a live batch that reported ~98 pages of about 71
characters, 100 with NO text at all, and 13 reading perfectly.

1. `model_attempts` could not reach anyone. `_page_from_dict` copied the
   worker's diagnostics onto page.meta BY NAME and knew only
   `unreadable_trace` and `accents_repaired` — so the field added in v0.3.45
   was dropped at the subprocess pipe, and was being hunted for in a live
   batch that could not have contained it. The same reassembly had already
   dropped `accents_repaired` once, reporting null for a repair that worked.
   It now forwards by exclusion: every key that is not an OcrPage field
   becomes meta. A key-by-key copy means each new diagnostic is born broken
   and nobody finds out until they need it.

2. An empty page and a 71-character page are the same bug. surya-ocr-2 emits
   block markup — a category and a bbox per block, BEFORE the text — so a
   reply stopped at its token cap is incomplete markup, and incomplete markup
   parses to a few blocks or to none. The cap is surya's
   `max_tokens_default=2048`, which its vLLM backend never overrides, and the
   same batch averaged 2,644 generated tokens per request: above it.
   `ocr.surya_max_tokens` raises it (0 = surya's own), opt-in because a higher
   cap costs tokens on every page. What is not optional is telling the cases
   apart, so a page that hit the cap now carries tokens, cap and
   `truncated: true` with an explanation that it is a cut-off reply and NOT a
   page the model could not read. Counts are per THREAD, like the trace, so
   one lane never reports another lane's verdict.

3. The fleet could grow by at most one pod per BOOT. `_should_grow` opened on
   a boolean held for the whole provision, and a provision is ~12 minutes — so
   a burst needing four pods took forty minutes to be served by four, whatever
   max_pods said (measured: 64 pages at 16 per pod got two pods, ready ten
   minutes apart; reported as "no real scale-out"). The grower also could not
   see the SIZE of the demand: `min(inflight) >= threshold` answers "is the
   fleet full?", so the only request it could formulate was "one more", and
   the callers blocked in acquire() — the actual measure of unmet demand —
   were invisible. Both are counts now: `_booting` (booting pods ARE capacity,
   so one demand is never provisioned for twice) and `_waiting`. Growth runs
   while pods + booting < demand, bounded by max_pods and by the new
   `scale_up_max_parallel_boots` (default 4) — a real limit, because the pods
   the demand justifies are the GPUs that start billing together. The original
   single-flight anticipatory trigger is KEPT beside it: max_inflight_per_pod
   defaults to unlimited, so requests are often never blocked and a merely
   saturated fleet would otherwise never grow at all. A prediction gets one
   pod; measured demand may get several.

Found while testing (3): acquire() ignored its own timeout when a sibling
process was booting a pod for the same pool — it waited a flat 15s and looped
back into the same decision without re-checking the deadline, for as long as
the sibling's registry entry survived (up to 15 minutes). A test hung on it.

42 tests across tests/test_runpod_scale_out.py (new) and
tests/test_surya_retry_budget.py. Two of my own test harnesses were wrong and
are fixed, not worked around: a sys.modules ledger stub that `from codai.api
import runpod_ledger` ignores once anything has imported the real module, and
a static count of increments against decrements, which cannot express "one
increment, four mutually exclusive exits" — replaced by runtime balance tests.
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.46"
+__version__ = "0.3.47"
 __tagline__ = "Complete orchestration, distribution and escalation of remotizable advanced inference."
 
 # Configure the CUDA caching allocator BEFORE torch is imported anywhere.

codai/admin/routes.py

@@ -4140,6 +4140,7 @@ def build_settings_dict(c, gpu_cards):
             "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_max_tokens": getattr(c.ocr, "surya_max_tokens", 0),
             "surya_langs": c.ocr.surya_langs,
             "surya_token_trace": c.ocr.surya_token_trace,
             "surya_repair_accents": c.ocr.surya_repair_accents,
@@ -4965,6 +4966,9 @@ async def api_save_settings(request: Request, username: str = Depends(require_ad
         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_tokens" in d:
+            o.surya_max_tokens = _i("surya_max_tokens",
+                                    getattr(o, "surya_max_tokens", 0), 0)
         if "surya_max_retries" in d:
             o.surya_max_retries = _i("surya_max_retries",
                                      getattr(o, "surya_max_retries", 1), -1)

codai/admin/templates/settings.html

@@ -1206,6 +1206,7 @@
       <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>
+      <div class="form-group" style="flex:1;min-width:140px"><label class="form-label" title="Cap on the tokens Surya may generate for ONE page. 0 uses Surya's own 2048, which its vLLM backend never overrides. Surya-ocr-2 emits block markup (a category and a bbox per block) BEFORE the text, so a reply stopped at the cap is incomplete markup and parses to a few blocks or to none — the page then arrives as a short fragment or as no text at all, which looks exactly like a page the model could not read. A dense page can legitimately need several thousand tokens. model_attempts.truncated in a page's result tells you when this is what went wrong.">Max tokens / page</label><input type="number" id="s-ocr-surya-max-tokens" class="form-input" min="0" placeholder="0 = 2048"></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>
@@ -2206,6 +2207,7 @@ async function loadSettings(){
     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-max-tokens').value = ocr.surya_max_tokens ?? 0;
     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;
@@ -2495,6 +2497,7 @@ async function saveSettings(silent){
       // 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_max_tokens: parseInt(document.getElementById('s-ocr-surya-max-tokens').value) || 0,
       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/api/runpod_worker.py

@@ -158,6 +158,15 @@ class RunpodModelConfig:
     # pool only ever grows when it has NO healthy pod, so fifty concurrent
     # requests would all pile onto pod #1 while max_pods sat unused.
     scale_up_inflight_per_pod: int = 4
+    # How many pods may be BOOTING at once when the load justifies them. 1 was
+    # the old behaviour, and it was not a choice: the grower held a single
+    # boolean for the whole ~12-minute boot, so a burst needing four pods took
+    # forty minutes to be served by four. It is still a real limit and not just
+    # a formality, because the pods the demand justifies are also the GPUs that
+    # start billing together — so this is the knob that says how fast an
+    # operator is willing to spend to absorb a burst. Growth never exceeds
+    # max_pods, and the cost caps still refuse each pod individually.
+    scale_up_max_parallel_boots: int = 4
     # Scaling DOWN again when the pressure that justified the pods is gone.
     # Idle teardown alone is not enough: it only fires on a pod with NO work and
     # no request for idle_timeout_s, so a fleet that grew to absorb a burst and
@@ -1745,6 +1754,8 @@ def parse_model_runpod(block: Optional[dict]) -> RunpodModelConfig:
                                              cfg.client_scale_ttl_s) or 1800)
     cfg.scale_up_inflight_per_pod = max(1, _as_int(b.get("scale_up_inflight_per_pod"),
                                                    cfg.scale_up_inflight_per_pod))
+    cfg.scale_up_max_parallel_boots = max(1, _as_int(
+        b.get("scale_up_max_parallel_boots"), cfg.scale_up_max_parallel_boots))
     # 0 disables shedding entirely (pods then go only when idle_timeout_s says so).
     cfg.scale_down_after_s = max(0, _as_int(b.get("scale_down_after_s"),
                                             cfg.scale_down_after_s))
@@ -2190,6 +2201,14 @@ class RunpodPodPool:
     busy pod for a dead one.
     """
 
+    #: Class-level so an instance built without __init__ (several tests, and
+    #: any future path that skips it) reads 0 rather than raising. `_booting` is
+    #: pods THIS process is provisioning right now; `_waiting` is acquire()
+    #: callers blocked for a slot. Both are counts, because the grower has to
+    #: know how MUCH demand there is, not merely whether the fleet is full.
+    _booting = 0
+    _waiting = 0
+
     def __init__(self, model_key, account_cfg, mcfg: "RunpodModelConfig", served_name,
                  entry: dict = None):
         #: The model's models.json entry, when this pool serves one specific
@@ -2222,7 +2241,16 @@ class RunpodPodPool:
         self._client_floor = 0
         self._client_floor_until = 0.0
         self._client_floor_logged = -1
-        self._provisioning = False
+        #: Pods THIS process is provisioning right now. A count, not a flag: it
+        #: used to be a boolean, and because a pod takes minutes to boot, the
+        #: pool could grow by at most one pod per boot however much work was
+        #: queued. Measured on 2026-10-13: 64 pages wanting 4 pods got 2, ready
+        #: ten minutes apart, and the client reported "no real scale-out".
+        self._booting = 0
+        #: acquire() callers blocked waiting for a slot. Without this the grower
+        #: can see THAT the fleet is full (every pod at its ceiling) but not BY
+        #: HOW MUCH, so it could only ever ask for one more pod.
+        self._waiting = 0
         self._closed = False
         #: When the fleet first looked larger than the load needs. Scale-down
         #: waits out scale_down_after_s from here, so a momentary dip in a burst
@@ -2833,9 +2861,9 @@ class RunpodPodPool:
             # self.pods until it does. Without this flag the 15 s scaler sees
             # "0 healthy, floor 1" and rents a second one — two A40s where the
             # configuration asked for one.
-            if self._provisioning:
+            if self._booting:
                 return
-            self._provisioning = True
+            self._booting += 1
         try:
             # A pod for this pool may already be up — a sibling engine's, or our
             # own from before the last restart. Renting beside it is how a warm
@@ -2860,7 +2888,7 @@ class RunpodPodPool:
                     raise
         finally:
             with self._cv:
-                self._provisioning = False
+                self._booting = max(0, self._booting - 1)
 
     def _pick(self, healthy: list, affinity: str):
         """Which pod serves this request: the one that already holds the
@@ -2885,13 +2913,42 @@ class RunpodPodPool:
         while len(self._affinity) > 4096:
             self._affinity.popitem(last=False)
 
+    def _pods_for_demand(self, serving: list) -> int:
+        """How many pods the work we can SEE justifies, capped at max_pods.
+
+        The work we can see is what the pods are carrying plus what is blocked
+        in acquire() waiting for a slot. The waiters are the part that was
+        missing: without them the grower knew the fleet was full but not by how
+        much, so the only question it could answer was "one more?".
+        """
+        threshold = max(1, int(self.mcfg.scale_up_inflight_per_pod or 1))
+        load = sum(p.inflight for p in serving) + max(0, self._waiting)
+        want = (load + threshold - 1) // threshold
+        return max(1, min(int(self.mcfg.max_pods), want))
+
     def _should_grow(self, healthy: list) -> bool:
         """True when real concurrency justifies another pod.
 
         The pool used to grow only when it had NO healthy pod, so every request
         beyond the first piled onto pod #1 and max_pods never came into play.
+
+        Then it grew one pod at a time — and since `_provisioning` stayed true
+        for the whole boot, "at a time" meant one per BOOT. A pod takes ~12
+        minutes to become ready (image pull plus warm-up), so a burst needing
+        four pods took forty minutes to be served by four, whatever max_pods
+        said. Measured on 2026-10-13: 64 pages in flight at 16 per pod, two pods
+        rented, ready ten minutes apart.
+
+        So the bound is the DEMAND, not the number one — while never exceeding
+        max_pods, and never more than `scale_up_max_parallel_boots` on the way
+        at once, because the demand-derived number is also the number of GPUs
+        that start billing together.
         """
-        if len(self.pods) >= self.mcfg.max_pods or self._provisioning:
+        in_hand = len(self.pods) + self._booting
+        if in_hand >= self.mcfg.max_pods:
+            return False
+        parallel = max(1, int(getattr(self.mcfg, "scale_up_max_parallel_boots", 1) or 1))
+        if self._booting >= parallel:
             return False
         # A draining pod is on its way out and takes no new work, so counting it
         # as capacity would answer "no need to grow" with the one pod that cannot
@@ -2899,9 +2956,25 @@ class RunpodPodPool:
         # opposite. It is simply not part of the fleet any more.
         serving = [p for p in healthy if not p.draining]
         if not serving:
+            # Nothing can serve a request. One on the way is enough to wait for;
+            # a second here would be two cards for the same first request, which
+            # is how a warm floor of one became two.
+            return self._booting == 0
+        # Demand we can measure: what the pods carry plus what is blocked. This
+        # is the part that may grow SEVERAL pods at once, because the number is
+        # real rather than a guess about the future.
+        if in_hand < self._pods_for_demand(serving):
             return True
+        # And the original, anticipatory trigger: every pod is at its in-flight
+        # threshold, so the next request has nowhere good to go. Kept because
+        # `max_inflight_per_pod` defaults to unlimited, so requests are often
+        # never blocked and `_waiting` stays 0 — without this a fleet whose pods
+        # are merely saturated would never grow at all. Single-flight, because
+        # unlike the demand above this is a prediction, and a prediction should
+        # not rent four cards.
         threshold = max(1, int(self.mcfg.scale_up_inflight_per_pod or 1))
-        return min(p.inflight for p in serving) >= threshold
+        return (self._booting == 0
+                and min(p.inflight for p in serving) >= threshold)
 
     def _grow_in_background(self):
         """Provision another pod without making the triggering request wait for it.
@@ -2918,7 +2991,10 @@ class RunpodPodPool:
                 print(f"[runpod] scale-up for {self.model_key!r} failed: {exc}", flush=True)
             finally:
                 with self._cv:
-                    self._provisioning = False
+                    # Decrement, not reset: several boots may be on the way, and
+                    # zeroing here would let the grower start the whole fleet
+                    # again for demand that is already being provisioned for.
+                    self._booting = max(0, self._booting - 1)
                     self._cv.notify_all()
         threading.Thread(target=_run, daemon=True,
                          name=f"runpod-grow-{self.model_key}").start()
@@ -2939,8 +3015,9 @@ class RunpodPodPool:
                 healthy = [p for p in self.pods if p.healthy]
                 chosen = self._pick(healthy, affinity) if healthy else None
                 if self._should_grow(healthy):
-                    self._provisioning = True
+                    self._booting += 1
                     grow_now = True
+                    want = self._pods_for_demand([p for p in healthy if not p.draining])
                 if chosen is not None:
                     chosen.inflight += 1
                     chosen.last_used = time.time()
@@ -2948,6 +3025,9 @@ class RunpodPodPool:
                     url = chosen.url
             if chosen is not None:
                 if grow_now:
+                    print(f"[runpod] {self.model_key!r}: growing to meet "
+                          f"{want} pod(s) of demand — {len(self.pods)} up, "
+                          f"{self._booting} booting", flush=True)
                     self._grow_in_background()      # absorbs the NEXT request
                 return chosen, url
 
@@ -2961,13 +3041,24 @@ class RunpodPodPool:
                     print(f"[runpod] another engine is booting a pod for "
                           f"{self.model_key!r} — waiting for it", flush=True)
                     with self._cv:
-                        self._provisioning = False
+                        self._booting = max(0, self._booting - 1)
                         self._cv.notify_all()
-                        self._cv.wait(15.0)
+                        # Never past the caller's deadline. This waited a flat
+                        # 15s and looped straight back into _should_grow, which
+                        # answers the same way while the sibling is still
+                        # booting — so the deadline was never reached and the
+                        # caller's timeout was ignored for as long as a sibling
+                        # entry survived (up to REAP_GRACE_SECONDS).
+                        remaining = deadline - time.time()
+                        if remaining <= 0:
+                            raise RunpodError(
+                                f"RunPod: another process is still booting a pod "
+                                f"for {self.model_key!r} after {timeout}s.")
+                        self._cv.wait(min(remaining, 15.0))
                     continue
                 if self._adopt_shared_pod():
                     with self._cv:
-                        self._provisioning = False
+                        self._booting = max(0, self._booting - 1)
                         self._cv.notify_all()
                     continue
                 # Nothing to serve this request yet — provision inline and retry.
@@ -2975,12 +3066,15 @@ class RunpodPodPool:
                     self._provision_one()
                 finally:
                     with self._cv:
-                        self._provisioning = False
+                        self._booting = max(0, self._booting - 1)
                         self._cv.notify_all()
                 continue
 
             # At max_pods with every pod at its in-flight ceiling, or someone else
-            # is provisioning: wait for a slot rather than piling on.
+            # is provisioning: wait for a slot rather than piling on. Counted
+            # while waiting, so the next pass through _should_grow can see how
+            # much work is actually queued rather than only that the fleet is
+            # full.
             with self._cv:
                 remaining = deadline - time.time()
                 if remaining <= 0:
@@ -2989,7 +3083,11 @@ class RunpodPodPool:
                         f"(max_pods={self.mcfg.max_pods}, "
                         f"max_inflight_per_pod={self.mcfg.max_inflight_per_pod}) "
                         f"within {timeout}s.")
+                self._waiting += 1
+                try:
                     self._cv.wait(min(remaining, 10.0))
+                finally:
+                    self._waiting = max(0, self._waiting - 1)
 
     def release(self, pod: PodHandle):
         with self._cv:
@@ -3345,16 +3443,16 @@ class RunpodPodPool:
             healthy_n = len([p for p in self.pods if p.healthy])
             if self._release_pending and healthy_n <= keep_min:
                 self._release_pending = False          # satisfied
-        if healthy_n < keep_min and not self._closed and not self._provisioning:
+        if healthy_n < keep_min and not self._closed and not self._booting:
             # One already on the way (from here, from ensure_ready, or from a
             # sibling engine sharing this pool) is one pod, not none: topping up
             # against a booting pod is how a warm floor of 1 became two.
             if sibling_is_provisioning(str(self.model_key)):
                 return
             with self._cv:
-                if self._provisioning:
+                if self._booting:
                     return
-                self._provisioning = True
+                self._booting += 1
             # Say so before spending. A warm floor silently failing to be
             # re-established is indistinguishable, from outside, from a warm
             # floor that is fine — which is how a pod terminated out of band
@@ -3368,7 +3466,7 @@ class RunpodPodPool:
                 print(f"[runpod] maintain: warm provision failed: {exc}", flush=True)
             finally:
                 with self._cv:
-                    self._provisioning = False
+                    self._booting = max(0, self._booting - 1)
 
     def close(self):
         self._closed = True

codai/config.py

@@ -1051,6 +1051,20 @@ class OcrConfig:
     #: 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
+    #: Cap on the tokens surya may generate for one page, or 0 for surya's own
+    #: 2048 (`chat_completions_batch(max_tokens_default=2048)`, which its vLLM
+    #: backend never overrides, so 2048 was the only value available).
+    #:
+    #: It matters because surya-ocr-2 emits BLOCK MARKUP -- a category and a
+    #: bbox per block, before the text -- so a reply stopped at the cap is
+    #: incomplete markup, and incomplete markup parses to a few blocks or to
+    #: none. The page then arrives as a short fragment or as no text at all,
+    #: which is indistinguishable in the finished text from a page the model
+    #: could not read. A dense page can legitimately need several thousand
+    #: tokens. 0 by default because raising it costs tokens on every page and
+    #: the right value depends on the documents; `model_attempts.truncated` in a
+    #: page's result says when this is what went wrong.
+    surya_max_tokens: int = 0
     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
@@ -1802,6 +1816,7 @@ class ConfigManager:
                 "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_max_tokens": getattr(self.config.ocr, "surya_max_tokens", 0),
                 "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/subprocess_engine.py

@@ -55,6 +55,11 @@ def _resolve_venv_dir(configured: str, name: str, baked_aliases: tuple = ()) ->
     return os.path.expanduser(f"~/.coderai/{name}")
 
 
+#: Keys of the worker's page dict that are OcrPage FIELDS. Everything else it
+#: sends is a diagnostic and is forwarded onto `page.meta` untouched.
+_PAGE_FIELDS = ("index", "text", "width", "height", "lines", "regions", "tables")
+
+
 class SubprocessOcrEngine(OcrEngine):
     """Base for engines that run in an isolated venv subprocess."""
 
@@ -299,6 +304,7 @@ class SubprocessOcrEngine(OcrEngine):
 
     @staticmethod
     def _page_from_dict(d: dict) -> OcrPage:
+        """Rebuild an OcrPage from the worker's JSON."""
         page = OcrPage(index=int(d.get("index", 0)), text=d.get("text", ""),
                        width=int(d.get("width", 0)), height=int(d.get("height", 0)))
         page.lines = [OcrLine(text=l.get("text", ""), bbox=list(l.get("bbox", [0, 0, 0, 0])),
@@ -320,4 +326,14 @@ class SubprocessOcrEngine(OcrEngine):
         repaired = d.get("accents_repaired")
         if repaired:
             page.meta["accents_repaired"] = repaired
+        # Everything else the worker chose to say. Forwarded by exclusion, not by
+        # name: this reassembly has now dropped a diagnostic TWICE for the same
+        # reason -- `accents_repaired` once, so a working repair reported null,
+        # and `model_attempts` again, which was being hunted for in a live batch
+        # at the moment it was noticed. A key-by-key copy means every future
+        # diagnostic is born broken and nobody finds out until they need it.
+        for k, v in d.items():
+            if k in _PAGE_FIELDS or k in page.meta or v in (None, "", [], {}):
+                continue
+            page.meta[k] = v
         return page

codai/ocr/surya.py

@@ -75,7 +75,9 @@ class SuryaEngine(SubprocessOcrEngine):
                 # 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)))}
+                                else int(getattr(self.cfg, "surya_max_retries", 1))),
+                # 0 leaves surya's own 2048 alone.
+                "max_tokens": max(0, int(getattr(self.cfg, "surya_max_tokens", 0) or 0))}
 
     #: Modes where the recognition VLM lives in a model server we only talk to.
     SERVED_MODES = ("model", "vllm", "llamacpp", "server")

codai/ocr/workers/ocr_worker.py

@@ -52,12 +52,21 @@ _TRACE = threading.local()
 #: module-level, so every lane in this process shares it.
 _RETRY_BUDGET = [None]
 
+#: Our override for surya's per-reply token cap, or 0 for surya's own 2048.
+#: Module-level for the same reason the retry budget is: the hook is.
+_MAX_TOKENS = [0]
+
+#: What surya uses when nobody tells it otherwise
+#: (`chat_completions_batch(max_tokens_default=2048)`).
+_SURYA_DEFAULT_MAX_TOKENS = 2048
+
 
 def _trace_begin():
     """Start a page's trace. Returns its (calls, repairs) — this thread's own."""
     _TRACE.calls = []
     _TRACE.repairs = []
     _TRACE.attempts = 0
+    _TRACE.tokens = []
     return _TRACE.calls, _TRACE.repairs
 
 
@@ -81,10 +90,33 @@ def _attempts_bump():
     return n
 
 
+def _attempt_tokens(n: int) -> None:
+    """How many tokens THIS attempt generated.
+
+    The number that separates a page the model could not read from a page whose
+    reply was cut off. surya-ocr-2 emits block markup -- a bbox and a tag per
+    block before any of the text -- so a generation stopped at its cap yields
+    markup that parses to a few blocks, or to none at all. "A few" is a page of
+    ~71 characters; "none" is a page with no text whatsoever. Both look like a
+    model that cannot read, and neither is.
+    """
+    try:
+        lst = getattr(_TRACE, "tokens", None)
+        if lst is None:
+            lst = _TRACE.tokens = []
+        lst.append(int(n or 0))
+    except Exception:
+        pass
+
+
 def _attempts():
     return getattr(_TRACE, "attempts", 0)
 
 
+def _attempt_token_counts():
+    return list(getattr(_TRACE, "tokens", None) or ())
+
+
 def _trace_repairs():
     return getattr(_TRACE, "repairs", None)
 
@@ -710,6 +742,8 @@ class SuryaWorker:
         #: 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))
+        #: 0 leaves surya's own 2048 in place.
+        _MAX_TOKENS[0] = max(0, int(opts.get("max_tokens", 0) or 0))
         #: 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.
@@ -777,9 +811,17 @@ class SuryaWorker:
             return
 
         def _wrapped(*args, **kwargs):
+            # surya's vLLM backend never passes `max_tokens_default`, so its
+            # 2048 was the only value available -- the same reason `max_retries`
+            # needed hooking. It arrives here as a keyword, so it can be raised
+            # without touching surya.
+            cap = _MAX_TOKENS[0]
+            if cap and "max_tokens_default" in kwargs:
+                kwargs["max_tokens_default"] = int(cap)
             result = inner(*args, **kwargs)
             try:
                 _attempts_bump()
+                _attempt_tokens(getattr(result, "token_count", 0))
                 sink = _trace_calls()
                 if sink is not None:
                     sink.append(getattr(result, "logprobs", None))
@@ -931,13 +973,29 @@ class SuryaWorker:
         # 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"),
-            }
+        toks = _attempt_token_counts()
+        cap = int(_MAX_TOKENS[0] or _SURYA_DEFAULT_MAX_TOKENS)
+        # Reported whenever there is anything to report -- a retry, or a reply
+        # that stopped at its cap. A page cut off mid-markup parses to a few
+        # blocks or to none, which is indistinguishable in the finished text
+        # from a page the model genuinely could not read. This is the only place
+        # the difference is visible.
+        capped = [t for t in toks if t >= cap * 0.98]
+        if tries > 1 or capped:
+            info = {"calls": tries or 1, "tokens": toks, "cap": cap}
+            if capped:
+                info["truncated"] = True
+                info["why"] = (
+                    f"the reply stopped at the {cap}-token cap, so the block "
+                    f"markup it was in the middle of is incomplete and parses "
+                    f"to less text than the page holds (or to none). This is a "
+                    f"cut-off reply, NOT a page the model could not read -- "
+                    f"raise ocr.surya_max_tokens.")
+            else:
+                info["why"] = ("surya re-ran this page because the generation "
+                               "repeated itself; each re-run is a full "
+                               "generation at a higher temperature")
+            page["model_attempts"] = info
         if self._trace and UNREADABLE in page["text"]:
             try:
                 page["unreadable_trace"] = _trace_report(calls)

docs/ocr.md

@@ -1005,7 +1005,63 @@ 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.
 
-#### A batch can be GPU-idle and still look stalled
+#### An empty page and a 71-character page are the same bug
+
+Two failure modes from one live batch: ~98 pages of about **71 characters** and
+**100 with no text at all**, against 13 that read perfectly. Neither is a model
+that cannot read.
+
+`surya-ocr-2` emits **block markup** — a category and a bounding box per block,
+*before* that block's text. A reply stopped at its token cap is therefore
+incomplete markup, and incomplete markup parses to a handful of blocks, or to
+none:
+
+| what the reply had room for | what the page looks like |
+|---|---|
+| all of it | the page, correctly |
+| some blocks, then the cap | a short fragment — the uniform low chars-per-page |
+| only markup, then the cap | **no text at all** |
+
+The cap is surya's `chat_completions_batch(max_tokens_default=2048)`, which its
+vLLM backend never overrides — so 2048 was the only value there has ever been.
+A dense page can legitimately need several thousand tokens, and a batch measured
+on 2026-10-13 averaged **2,644 generated tokens per request**, i.e. above the
+cap.
+
+`ocr.surya_max_tokens` raises it (0 = leave surya's 2048). It is opt-in because
+a higher cap costs tokens on every page and the right value depends on the
+documents. What is **not** optional is being able to tell the two cases apart,
+so every page that hit the cap says so in its own result:
+
+```json
+"model_attempts": {
+  "calls": 1, "tokens": [2048], "cap": 2048, "truncated": true,
+  "why": "the reply stopped at the 2048-token cap, so the block markup it was in
+          the middle of is incomplete and parses to less text than the page holds
+          (or to none). This is a cut-off reply, NOT a page the model could not
+          read — raise ocr.surya_max_tokens."
+}
+```
+
+A page whose reply fitted carries nothing, and is byte-identical to what it
+always returned. The per-attempt token counts are kept **per thread**, like the
+token trace, so one lane never reports another lane's verdict.
+
+### A diagnostic that does not arrive is not a diagnostic
+
+`model_attempts` was added, shipped, and then hunted for in a live batch that
+could not have contained it: `SubprocessOcrEngine._page_from_dict` copied the
+worker's diagnostics onto `page.meta` **by name**, and it knew only
+`unreadable_trace` and `accents_repaired`. The same reassembly had already
+dropped `accents_repaired` once, so a working repair reported `null`.
+
+It now forwards by exclusion — every key that is not an `OcrPage` field becomes
+`page.meta`. A key-by-key copy means each new diagnostic is born broken and
+nobody finds out until the moment they need it.
+
+---
+
+## 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

docs/runpod.md

@@ -334,6 +334,58 @@ Consequences worth knowing:
 
 `atexit` also tears down every pool on a clean shutdown.
 
+### Growing the fleet: demand, not one pod per boot
+
+Scale-up is triggered in `acquire()`. It used to be gated on a boolean,
+`_provisioning`, held for the whole provision — and a provision is a pod boot,
+~12 minutes. So however much work was queued, **the fleet grew by at most one
+pod per boot**: a burst needing four pods took forty minutes to be served by
+four, whatever `max_pods` said. Measured on 2026-10-13 with 64 pages in flight
+at `scale_up_inflight_per_pod: 16` (four pods of demand) and `max_pods: 64`: two
+pods, ready ten minutes apart.
+
+The grower also could not see the *size* of the demand. It asked
+`min(p.inflight) >= threshold`, which answers "is the fleet full?" — so the only
+request it could formulate was "one more". The requests blocked in `acquire()`
+waiting for a slot, which are the actual measure of unmet demand, were invisible.
+
+Both are now counts rather than flags:
+
+- **`_booting`** — pods this process is provisioning right now. Booting pods
+  count as capacity, so the same demand is never provisioned for twice.
+- **`_waiting`** — callers blocked for a slot, added to what the pods carry to
+  give the demand in pods: `ceil((inflight + waiting) / scale_up_inflight_per_pod)`.
+
+Growth continues while `pods + booting < demand`, bounded by `max_pods` and by
+**`scale_up_max_parallel_boots`** (default 4). That last one is a real limit and
+not a formality: the pods the demand justifies are also the GPUs that start
+billing together, so it is the knob that says how fast an operator is willing to
+spend to absorb a burst. The per-pod cost caps still refuse each pod
+individually.
+
+The original single-flight trigger is kept beside it — every pod at its
+in-flight threshold grows **one** more, single-flight, because
+`max_inflight_per_pod` defaults to unlimited and requests are then never
+blocked, so `_waiting` stays 0 and a merely-saturated fleet would never grow at
+all. A prediction gets one pod; measured demand may get several.
+
+Growth says so before it spends:
+
+```
+[runpod] 'datalab-to/surya-ocr-2': growing to meet 4 pod(s) of demand — 1 up, 2 booting
+```
+
+Two things found while fixing this, both worth knowing:
+
+- **A pod's boot time bounds the first burst, not `max_pods`.** Twelve minutes
+  pass before the first new pod answers anything, so a burst is served by
+  whatever is already warm. Raising `max_pods` does not help it.
+- **`acquire()` ignored its own timeout** when a *sibling process* was booting a
+  pod for the same pool: it waited a flat 15 s and looped straight back into the
+  same decision, never re-checking the deadline, for as long as the sibling's
+  registry entry survived (up to `REAP_GRACE_SECONDS`, 15 minutes). It now
+  raises when the caller's deadline passes.
+
 ### The same sweep reconciles the pools (a pod that dies behind our back)
 
 The listing is not only used to find pods that should not exist. It is also the

tests/test_runpod_client_scale.py

@@ -331,15 +331,15 @@ def test_ensure_ready_marks_the_pool_as_provisioning():
     boot: two A40s where the configuration asked for one."""
     src = __import__("pathlib").Path(rw.__file__).read_text()
     blk = src.split("def ensure_ready")[1].split("def _pick")[0]
-    assert "self._provisioning = True" in blk
-    assert "finally:" in blk and "self._provisioning = False" in blk
-    assert blk.index("if self._provisioning:") < blk.index("self._provisioning = True")
+    assert "self._booting += 1" in blk
+    assert "finally:" in blk and "self._booting = max(0, self._booting - 1)" in blk
+    assert blk.index("if self._booting:") < blk.index("self._booting += 1")
 
 
 def test_the_warm_top_up_does_not_stack_on_a_booting_pod():
     src = __import__("pathlib").Path(rw.__file__).read_text()
     blk = src.split("def maintain")[1].split("def close")[0]
-    assert "not self._provisioning" in blk
+    assert "not self._booting" in blk
     assert "sibling_is_provisioning" in blk
 
 
@@ -348,7 +348,11 @@ def test_the_demand_path_already_had_the_guard_and_keeps_it():
     the two paths cannot drift apart again."""
     src = __import__("pathlib").Path(rw.__file__).read_text()
     blk = src.split("def _should_grow")[1].split("def _grow_in_background")[0]
-    assert "or self._provisioning" in blk
+    # The flag became a COUNT: a boolean held for the whole ~12-minute boot
+    # meant the fleet could grow by one pod per boot however much was queued.
+    # The guard is still here, it just reads how many rather than whether.
+    assert "self._booting >= parallel" in blk
+    assert "in_hand >= self.mcfg.max_pods" in blk
 
 
 def test_the_answer_names_the_model_the_client_asked_about():

tests/test_runpod_scale_out.py

+"""A pool that could only ever grow by one pod per pod-boot.
+
+`_should_grow` opened with:
+
+    if len(self.pods) >= self.mcfg.max_pods or self._provisioning:
+        return False
+
+and `_provisioning` was a boolean held for the WHOLE provision — which is a pod
+boot, measured at ~12 minutes on the Digesta orchestrator (~7 min image pull plus
+~5 min warm-up). So however much work was queued, the fleet grew one pod per
+boot: a burst needing four pods took forty minutes to be served by four,
+whatever `max_pods` said.
+
+Measured 2026-10-13, 64 pages in flight at `scale_up_inflight_per_pod: 16`
+(= four pods of demand) against `max_pods: 64`: two pods were rented, ready at
+11:20 and 11:30 — ten minutes apart, one boot apart. The client reported it as
+"a per-pod concurrency issue with no real scale-out", which is exactly what it
+was.
+
+The other half was that the grower could not see the size of the demand. It
+looked at `min(p.inflight) >= threshold`, which answers "is the fleet full?" —
+so the only request it could ever formulate was "one more pod". The requests
+blocked in `acquire()` waiting for a slot, which are the actual measure of unmet
+demand, were invisible.
+
+What these tests pin is the caution as much as the fix: the demand-derived
+number is also the number of GPUs that start billing together, and an earlier
+incident in this same file's history was two pods rented for ONE request 45
+seconds apart.
+"""
+
+import pathlib
+import sys
+import threading
+
+ROOT = pathlib.Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT))
+
+from codai.api import runpod_worker as rw                          # noqa: E402
+
+
+def _pool(max_pods=64, threshold=16, parallel=4, pods=(), booting=0, waiting=0):
+    pool = rw.RunpodPodPool.__new__(rw.RunpodPodPool)
+    pool._cv = threading.Condition()
+    pool.model_key = "datalab-to/surya-ocr-2"
+    pool._closed = False
+    pool._booting = booting
+    pool._waiting = waiting
+    pool._surplus_since = 0.0
+    pool.pods = list(pods)
+    cfg = rw.RunpodModelConfig()
+    cfg.max_pods = max_pods
+    cfg.scale_up_inflight_per_pod = threshold
+    cfg.scale_up_max_parallel_boots = parallel
+    pool.mcfg = cfg
+    return pool
+
+
+def _pod(pid, inflight=0, draining=False):
+    return rw.PodHandle(pod_id=pid, url=f"http://{pid}", hourly_usd=0.59,
+                        started_at=0.0, inflight=inflight, draining=draining)
+
+
+def _healthy(pool):
+    return [p for p in pool.pods if p.healthy]
+
+
+# ------------------------------------------------- how much demand is there #
+def test_demand_counts_what_is_queued_not_just_what_is_running():
+    """The missing input. 16 on the one pod plus 48 waiting is 64 pages, which
+    at 16 per pod is four pods — not "one more"."""
+    pool = _pool(pods=[_pod("a", inflight=16)], waiting=48)
+    assert pool._pods_for_demand(_healthy(pool)) == 4
+
+
+def test_demand_rounds_up_a_partial_pod():
+    pool = _pool(pods=[_pod("a", inflight=16)], waiting=1)
+    assert pool._pods_for_demand(_healthy(pool)) == 2
+
+
+def test_demand_never_exceeds_max_pods():
+    """max_pods is the budget, and it is still the budget."""
+    pool = _pool(max_pods=2, pods=[_pod("a", inflight=16)], waiting=1000)
+    assert pool._pods_for_demand(_healthy(pool)) == 2
+
+
+def test_an_idle_fleet_wants_one_pod_not_zero():
+    """Zero would make the arithmetic argue for shedding the pod serving the
+    next request; shedding is _scale_down's job, with its own hysteresis."""
+    pool = _pool(pods=[_pod("a", inflight=0)])
+    assert pool._pods_for_demand(_healthy(pool)) == 1
+
+
+# ------------------------------------------------------- growing in parallel #
+def test_a_burst_grows_past_one_pod_at_a_time():
+    """The fix. One pod up, 48 pages queued, nothing booting: grow."""
+    pool = _pool(pods=[_pod("a", inflight=16)], waiting=48)
+    assert pool._should_grow(_healthy(pool)) is True
+
+
+def test_it_keeps_growing_while_earlier_pods_are_still_booting():
+    """What the boolean made impossible. Four pods of demand, one up and one
+    booting is two in hand — still short, so still grow."""
+    pool = _pool(pods=[_pod("a", inflight=16)], waiting=48, booting=1)
+    assert pool._should_grow(_healthy(pool)) is True
+
+
+def test_it_stops_once_the_pods_in_hand_cover_the_demand():
+    """Booting pods ARE capacity. Counting only the ready ones is how the same
+    demand gets provisioned for over and over."""
+    pool = _pool(pods=[_pod("a", inflight=16)], waiting=48, booting=3)
+    assert pool._should_grow(_healthy(pool)) is False
+
+
+def test_parallel_boots_are_capped():
+    """The demand-derived number is also the number of GPUs that start billing
+    together, so an operator gets to say how fast that may happen."""
+    pool = _pool(pods=[_pod("a", inflight=16)], waiting=1000, parallel=2, booting=2)
+    assert pool._should_grow(_healthy(pool)) is False
+    assert _pool(pods=[_pod("a", inflight=16)], waiting=1000, parallel=8,
+                 booting=2)._should_grow([_pod("a", inflight=16)]) is True
+
+
+def test_the_cap_defaults_to_more_than_one_but_stays_finite():
+    assert rw.RunpodModelConfig().scale_up_max_parallel_boots > 1
+    assert rw.RunpodModelConfig().scale_up_max_parallel_boots < 64
+
+
+def test_max_pods_still_stops_growth_with_boots_counted():
+    """Two up and two booting against max_pods 4 is the fleet, even though the
+    queue would justify forty."""
+    pool = _pool(max_pods=4, pods=[_pod("a", inflight=16), _pod("b", inflight=16)],
+                 waiting=600, booting=2)
+    assert pool._should_grow(_healthy(pool)) is False
+
+
+# ------------------------------- the incident the single-flight guard existed for #
+def test_an_empty_pool_rents_exactly_one_for_the_first_request():
+    """Two pods for ONE request, 45 seconds apart, is in this file's history.
+    With nothing able to serve, one on the way is enough to wait for."""
+    pool = _pool(pods=[], waiting=1)
+    assert pool._should_grow([]) is True
+    assert _pool(pods=[], waiting=1, booting=1)._should_grow([]) is False
+
+
+def test_a_draining_pod_is_not_capacity():
+    """It takes no new work, so counting it would answer "no need to grow" with
+    the one pod that cannot help."""
+    pool = _pool(pods=[_pod("a", inflight=16, draining=True)], waiting=48)
+    assert pool._should_grow([p for p in pool.pods if not p.draining]) is True
+
+
+def test_a_closed_pool_is_not_grown():
+    pool = _pool(max_pods=0, pods=[], waiting=10)
+    assert pool._should_grow([]) is False
+
+
+# --------------------------------------------- the counter, and the old flag #
+def test_the_counters_exist_without___init__():
+    """Several fixtures (and any path that skips __init__) build a pool with
+    __new__. Class-level defaults mean the grower reads 0 rather than raising."""
+    pool = rw.RunpodPodPool.__new__(rw.RunpodPodPool)
+    assert pool._booting == 0 and pool._waiting == 0
+
+
+def test_the_single_flight_paths_ask_how_many_not_whether():
+    """ensure_ready (the FIRST pod) and the warm-floor top-up must still be
+    single-flight — they are not allowed to grow a fleet — so they check the
+    count directly rather than a boolean that scale-up also writes."""
+    src = (ROOT / "codai/api/runpod_worker.py").read_text()
+    ready = src.split("def ensure_ready")[1].split("\n    def ")[0]
+    assert "if self._booting:" in ready and "self._booting += 1" in ready
+    maint = src.split("def maintain")[1].split("\n    def ")[0]
+    assert "not self._booting" in maint
+
+
+def test_the_counter_returns_to_zero_after_a_scale_up():
+    """A leaked increment is a fleet that never grows again — the original bug,
+    reintroduced and harder to see. Counting `+= 1` against `-= 1` in the source
+    cannot show this: acquire() increments ONCE and has four mutually exclusive
+    exits, each of which must decrement."""
+    pool = _pool(pods=[_pod("a", inflight=16)], waiting=48)
+    provisioned = []
+    pool._provision_one = lambda: provisioned.append(1)
+    pool._pick = lambda healthy, affinity: healthy[0]
+    pool._remember_affinity = lambda a, p: None
+    got, url = pool.acquire(timeout=5)
+    assert got.pod_id == "a"
+    for _ in range(200):                      # the grow thread is a daemon
+        if pool._booting == 0 and provisioned:
+            break
+        __import__("time").sleep(0.01)
+    assert provisioned == [1], "scale-up did not run"
+    assert pool._booting == 0, "the boot counter leaked"
+
+
+def test_the_counter_returns_to_zero_when_a_sibling_is_already_booting():
+    """The bail-out paths are where a leak hides: nothing was provisioned, so
+    nothing obviously went wrong, and the pool simply never grows again."""
+    import codai.api.runpod_worker as mod
+    pool = _pool(pods=[], waiting=1)
+    saved = mod.sibling_is_provisioning
+    mod.sibling_is_provisioning = lambda key: True
+    try:
+        try:
+            pool.acquire(timeout=0.3)
+        except Exception:
+            pass                               # the timeout is the point
+    finally:
+        mod.sibling_is_provisioning = saved
+    assert pool._booting == 0, "the boot counter leaked on the sibling path"
+
+
+def test_the_callers_timeout_is_honoured_while_a_sibling_boots():
+    """This path waited a flat 15s and looped back into _should_grow, which
+    answers the same way while the sibling is still booting — so the deadline
+    was never reached and acquire() ignored its timeout for as long as the
+    sibling's registry entry survived (up to REAP_GRACE_SECONDS, 15 minutes).
+    Found by a test that hung."""
+    import time as _t
+    import codai.api.runpod_worker as mod
+    from codai.api.runpod_client import RunpodError
+    pool = _pool(pods=[], waiting=1)
+    saved = mod.sibling_is_provisioning
+    mod.sibling_is_provisioning = lambda key: True
+    t0 = _t.time()
+    try:
+        raised = None
+        try:
+            pool.acquire(timeout=0.5)
+        except RunpodError as exc:
+            raised = exc
+    finally:
+        mod.sibling_is_provisioning = saved
+    assert raised is not None, "acquire() never gave up"
+    assert _t.time() - t0 < 10, "it waited past its deadline"
+    assert pool._booting == 0
+
+
+def test_waiters_are_counted_only_while_waiting():
+    """A counter that leaked would make an idle pool look permanently busy and
+    rent pods for work that finished long ago."""
+    src = (ROOT / "codai/api/runpod_worker.py").read_text()
+    blk = src.split("def acquire")[1].split("\n    def ")[0]
+    assert "self._waiting += 1" in blk
+    assert "self._waiting = max(0, self._waiting - 1)" in blk
+    assert "finally:" in blk.split("self._waiting += 1")[1][:200]
+
+
+def test_growth_is_announced_with_the_numbers():
+    """A fleet that silently fails to grow is indistinguishable from a fleet
+    that does not need to — which is how ten minutes of this went unnoticed."""
+    src = (ROOT / "codai/api/runpod_worker.py").read_text()
+    assert "growing to meet" in src and "booting" in src

tests/test_runpod_warm_floor_rerent.py

@@ -70,13 +70,18 @@ def _install(monkeypatch, pool, ledger=None, unregistered=None):
                         lambda pid: (unregistered if unregistered is not None
                                      else []).append(pid))
 
-    class _Ledger:
-        @staticmethod
-        def record_spend(key, cost, meta):
+    # Patch the FUNCTION on the real module, not the module in sys.modules:
+    # `from codai.api import runpod_ledger` reads the attribute already set on
+    # the `codai.api` package at first import, so a sys.modules stub is ignored
+    # the moment anything else has imported the real one — which made this pass
+    # alone and fail in a full run.
+    from codai.api import runpod_ledger as _real
+
+    def _record(key, cost, meta):
         if ledger is not None:
             ledger.append((key, round(cost, 4), meta.get("pod_id")))
 
-    monkeypatch.setitem(sys.modules, "codai.api.runpod_ledger", _Ledger)
+    monkeypatch.setattr(_real, "record_spend", _record)
 
 
 # -------------------------------------------- what the pool knows on its own #

tests/test_surya_retry_budget.py

@@ -197,14 +197,86 @@ def test_a_retried_page_says_so_in_its_result():
     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
+    blk = src.split("tries = _attempts()")[1][:1600]
+    assert "if tries > 1 or capped:" in blk      # a clean page stays untouched
+
+
+def test_a_truncated_reply_is_distinguished_from_an_unreadable_page():
+    """The two failure modes a live batch produced: ~98 pages of ~71 characters
+    and 100 with no text at all. Both are what a reply cut off mid-markup parses
+    to, and both look exactly like a page the model could not read."""
+    src = SRC.read_text()
+    blk = src.split("tries = _attempts()")[1][:1600]
+    assert '"truncated"' in blk
+    assert "cap" in blk and "NOT a page the model could not read" in blk
 
 
 def test_a_clean_page_carries_no_attempt_block():
+    """One attempt, well under the cap: byte-identical to what it always
+    returned."""
     src = SRC.read_text()
     i = src.index("tries = _attempts()")
-    assert "if tries > 1:" in src[i:i + 200]
+    assert "if tries > 1 or capped:" in src[i:i + 1600]
+
+
+def test_the_token_counts_are_per_thread():
+    """Like the trace and the attempt count: one lane must not report another
+    lane's token counts, or the truncation verdict lands on the wrong page."""
+    mod = _worker_module()
+    import threading as _th
+    seen = {}
+
+    def lane(name, counts):
+        mod._trace_begin()
+        for c in counts:
+            mod._attempt_tokens(c)
+        seen[name] = mod._attempt_token_counts()
+
+    a = _th.Thread(target=lane, args=("a", [2048]))
+    b = _th.Thread(target=lane, args=("b", [11, 12]))
+    a.start(); b.start(); a.join(); b.join()
+    assert seen == {"a": [2048], "b": [11, 12]}
+
+
+def test_the_cap_can_be_raised_without_touching_surya(monkeypatch):
+    """surya's vLLM backend never passes max_tokens_default, so 2048 was the
+    only value available. It arrives at _generate_one as a keyword, so the hook
+    can raise it."""
+    mod = _worker_module()
+    oc = _fake_surya(monkeypatch, mod)
+    seen = {}
+
+    def _gen(*a, **k):
+        seen.update(k)
+        return _Result()
+
+    oc._generate_one = _gen
+    w = _worker(mod, token_trace=True, max_tokens=8192)
+    w._install_trace_hook()
+    oc._generate_one(max_tokens_default=2048)
+    assert seen["max_tokens_default"] == 8192
+
+
+def test_zero_leaves_suryas_own_cap_alone(monkeypatch):
+    """Raising it costs tokens on every page, so it is opt-in."""
+    mod = _worker_module()
+    oc = _fake_surya(monkeypatch, mod)
+    seen = {}
+    oc._generate_one = lambda *a, **k: (seen.update(k), _Result())[1]
+    w = _worker(mod, token_trace=True, max_tokens=0)
+    w._install_trace_hook()
+    oc._generate_one(max_tokens_default=2048)
+    assert seen["max_tokens_default"] == 2048
+    from codai.config import OcrConfig
+    assert OcrConfig().surya_max_tokens == 0
+
+
+def test_the_engine_sends_the_token_cap_at_load():
+    from codai.ocr.surya import SuryaEngine
+    from codai.config import OcrConfig
+    cfg = OcrConfig()
+    cfg.surya_max_tokens = 6000
+    assert SuryaEngine(cfg)._worker_opts()["max_tokens"] == 6000
 
 
 # -------------------------------------------- the engine passes it through #