absorb whatever a client offers: discovered pod width, and 429 (v0.4.38)

A client must be able to send too much. The server's job is to carry what
it can and say so about the rest -- not to depend on the client having
been told the right number, which is what the previous fix amounted to.

TWO THINGS WERE MISSING.

1. NOTHING BOUNDED WHAT ONE POD WAS SENT. `max_inflight_per_pod`
   defaults to 0, meaning unlimited, so out of the box every page handed
   to the pool is dispatched. A client offering more than a pod can serve
   therefore does not queue in the orchestrator, where the queue would be
   counted, reported as demand and bounded -- it queues in the POD'S
   LISTEN BACKLOG, and an overflowing backlog makes a healthy pod refuse
   connections. That is what cost eight A40s yesterday.

   The right width is a property of the pod, the model and the card, not
   of the install, so the pod is asked rather than configured. A refusal
   halves what we send that pod, FROM THE IN-FLIGHT COUNT AT THE REFUSAL
   -- its observed breaking point, not a guess -- and successes walk it
   back up one slot at a time (multiplicative decrease, additive
   increase: overshooting costs refused connections and a misdiagnosed
   fleet, undershooting costs a little latency). Floored at 4, never
   above the operator's own ceiling, per pod so one overloaded pod does
   not throttle the fleet, and inert on an install where no pod ever
   refuses. Live numbers: 64 configured, 58 in flight at the refusal,
   width becomes 29 -- the pods were measured serving 29-32.

2. A FULL QUEUE ANSWERED SLOWER INSTEAD OF SAYING NO. The queue itself is
   deliberate: when a pod finishes a page there must already be another
   ready to hand it, or the fleet idles between dispatches and is paid
   for the gap. So the server carries `ocr.max_concurrency` in flight
   PLUS `ocr.max_queued_pages` waiting (0 = follow max_concurrency) --
   and past that answers 429 with Retry-After and the numbers a client
   needs to size its window (X-OCR-Page-Width, X-OCR-Queue-Limit).

   Before, a client offering several times the capacity got no error at
   all: it got replies that took longer and longer, which reads as "the
   pages are getting slower" rather than "stop sending", so it sent more.
   Measured 2026-10-19: 522 pages offered against a fleet carrying ~250,
   mean page latency 238s against a 19s service time, nothing in any
   reply saying the server was over capacity.

   Refused BEFORE the upload is read. The client hears at once instead of
   after transferring a document the server will not look at, and the
   server stops buffering megabytes it is about to discard -- in-flight
   memory here is concurrent requests times upload size, the product that
   has OOM-killed this endpoint before.

Every page now takes its slot through one counted path, which is what
makes the queue depth knowable. Writing that exposed a real defect in
what it replaced: `self._sem or asyncio.Semaphore(4)` built a FRESH
semaphore per page, so an unconfigured manager bounded nothing at all --
forty pages of one document ran at once against what every line of code
called a limit of four. Caught by the existing pod-pipeline gate test.

38 tests. Full suite 3817 passed, 15 pre-existing failures unchanged.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KnfkxBpcZ1v6ypRFFyhrVM

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.4.37"
+__version__ = "0.4.38"
 __tagline__ = "Complete orchestration, distribution and escalation of remotizable advanced inference."
 
 # Configure the CUDA caching allocator BEFORE torch is imported anywhere.

codai/api/ocr.py

@@ -77,6 +77,54 @@ def _page_opts(max_tokens, max_continuations) -> dict:
     return out
 
 
+def _refuse_if_full(cfg) -> None:
+    """429 when the queue that keeps the pods fed is itself full.
+
+    THE QUEUE IS DELIBERATE. When a pod finishes a page there has to be another
+    page already waiting, or the fleet idles between every dispatch and is paid
+    for the gap. So the server accepts `ocr.max_concurrency` pages in flight
+    PLUS `ocr.max_queued_pages` waiting, and only past that does it refuse.
+
+    What it must not do is accept without limit. A client offering several times
+    the capacity does not get an error -- it gets replies that take longer and
+    longer, which reads as "the pages are getting slower", not as "stop
+    sending". It then does exactly the wrong thing and sends more. Measured
+    2026-10-19: a client offered 522 pages at once against a fleet that could
+    carry ~250, mean page latency went to 238 s against a 19 s service time,
+    and nothing in any reply said the server was over its capacity.
+
+    Refused BEFORE the upload is read, which matters twice: the client learns
+    immediately instead of after transferring a document the server was never
+    going to look at, and the server does not buffer megabytes it is about to
+    discard. The in-flight memory of this endpoint is concurrent requests times
+    upload size, and that product is the one that has OOM-killed it before.
+
+    `Retry-After` is the actionable half. A client that honours it finishes
+    sooner than one that retries immediately, because its retries stop
+    competing with the work already accepted.
+    """
+    try:
+        state = ocr_manager.admission()
+    except Exception:
+        return                      # never let admission accounting refuse work
+    if not state.get("full"):
+        return
+    raise HTTPException(
+        status_code=429,
+        detail=(f"{state['queued']} page(s) are already queued for a slot "
+                f"(ocr.max_queued_pages = {state['queue_limit']}, with "
+                f"{state['running']} of {state['width']} in flight). The queue "
+                f"is kept full on purpose so no pod waits for work; this "
+                f"request is past it. Retry after the Retry-After interval — "
+                f"pushing harder makes every accepted page slower, including "
+                f"this one when it is accepted."),
+        headers={"Retry-After": "2",
+                 "X-OCR-Pages-Running": str(state["running"]),
+                 "X-OCR-Pages-Queued": str(state["queued"]),
+                 "X-OCR-Page-Width": str(state["width"]),
+                 "X-OCR-Queue-Limit": str(state["queue_limit"])})
+
+
 async def _bounded(cfg, coro):
     """Run an OCR read under the install's wall-clock ceiling, if it set one.
 
@@ -151,6 +199,7 @@ async def create_ocr(
     if cfg is None:
         raise HTTPException(status_code=503, detail="OCR subsystem is not available")
     ocr_manager.configure(cfg)
+    _refuse_if_full(cfg)
 
     data = await file.read()
     if not data:
@@ -202,6 +251,7 @@ async def create_ocr_batch(
     if cfg is None:
         raise HTTPException(status_code=503, detail="OCR subsystem is not available")
     ocr_manager.configure(cfg)
+    _refuse_if_full(cfg)
 
     payloads = []
     for f in files:

codai/api/runpod_worker.py

@@ -2647,6 +2647,27 @@ class PodHandle:
     #: keeps serving what it already has: killing a pod with requests on it to
     #: save money would be paying for the work twice.
     draining: bool = False
+    #: What THIS pod has shown it will accept at once, 0 until it has shown
+    #: anything. An upper bound discovered from the pod's own behaviour, applied
+    #: on top of whatever the operator configured.
+    #:
+    #: `max_inflight_per_pod` defaults to 0, which means unlimited -- so out of
+    #: the box the pool dispatches every page it is given, and a client that
+    #: offers more than a pod can serve does not queue in the orchestrator where
+    #: it would be visible and bounded. It queues in the POD'S LISTEN BACKLOG,
+    #: and an overflowing backlog refuses connections from a perfectly healthy
+    #: pod. Measured 2026-10-19: a client offering 64 per pod against pods that
+    #: serve ~32 (`Running: 29-32 reqs, Waiting: 0`) produced refusals that cost
+    #: eight A40s, and no configuration anywhere said 64 was too many.
+    #:
+    #: Halved on a refusal, from the in-flight count AT the refusal -- which is
+    #: the pod's observed breaking point, not a guess -- and walked back up one
+    #: slot at a time as pages succeed. So an install needs no knowledge of its
+    #: pods' capacity, and a client may offer whatever it likes: the surplus
+    #: waits on this side, where it is counted.
+    soft_limit: int = 0
+    #: Successful releases since the limit last moved. Drives the increase.
+    since_limit_change: int = 0
     #: Set aside after a connection-level failure, until this moment, and NOT
     #: terminated. See `discard`.
     #:
@@ -3954,7 +3975,10 @@ class RunpodPodPool:
             if p.draining:
                 draining += 1
                 continue
-            cap = self.pod_capacity(p)          # scaled to THIS pod's card
+            # Scaled to THIS pod's card, and narrowed by what this pod has
+            # shown it will accept: a configured ceiling the pod cannot honour
+            # is not a ceiling, it is an overload waiting to be misdiagnosed.
+            cap = self.pod_capacity_now(p)
             if cap <= 0 or p.inflight < cap:
                 free.append(p)
             else:
@@ -4093,6 +4117,44 @@ class RunpodPodPool:
         cache[key] = facts
         return facts
 
+    def pod_capacity_now(self, pod: "PodHandle") -> int:
+        """How many pages this pod may hold RIGHT NOW: 0 for no bound.
+
+        The operator's configured capacity, narrowed by what the pod has shown
+        it will actually accept. The narrowing is what lets a client offer more
+        than the fleet can take without any of it reaching a socket backlog --
+        the surplus waits in `acquire`, which counts it, reports it as demand
+        and is where a queue belongs.
+        """
+        cap = int(self.pod_capacity(pod) or 0)
+        soft = int(getattr(pod, "soft_limit", 0) or 0)
+        if soft <= 0:
+            return cap
+        return soft if cap <= 0 else min(cap, soft)
+
+    def note_refusal(self, pod: "PodHandle") -> int:
+        """This pod refused a connection: halve what we send it. Returns the new
+        width.
+
+        From the in-flight count at the moment of refusal, because that is the
+        pod's OBSERVED breaking point. Halving a configured number instead would
+        be halving a number that was already wrong.
+        """
+        with self._cv:
+            observed = max(int(pod.inflight), int(getattr(pod, "soft_limit", 0) or 0))
+            if observed <= 0:
+                observed = SOFT_LIMIT_FLOOR * 2
+            new = max(SOFT_LIMIT_FLOOR, observed // 2)
+            prev = int(getattr(pod, "soft_limit", 0) or 0)
+            pod.soft_limit = new
+            pod.since_limit_change = 0
+        if new != prev:
+            print(f"[runpod] pod {pod.pod_id} refused a connection with "
+                  f"{observed} page(s) on it — sending it at most {new} at a "
+                  f"time from now (was {prev or 'unbounded'}). The surplus waits "
+                  f"here rather than in its socket backlog.", flush=True)
+        return new
+
     def pod_weight(self, pod: "PodHandle") -> float:
         """This pod's share of the load — memoised per card. See `_card_facts`."""
         return self._card_facts(
@@ -4600,6 +4662,18 @@ class RunpodPodPool:
         with self._cv:
             pod.inflight = max(0, pod.inflight - 1)
             pod.last_ok = time.time()
+            # Walk a narrowed pod back up. Only a pod that WAS narrowed: an
+            # unbounded pool stays unbounded unless a pod has told us otherwise,
+            # so nothing changes for an install this never fired on.
+            soft = int(getattr(pod, "soft_limit", 0) or 0)
+            if soft > 0:
+                ceiling = int(self.pod_capacity(pod) or 0)
+                pod.since_limit_change = int(
+                    getattr(pod, "since_limit_change", 0) or 0) + 1
+                if (pod.since_limit_change >= SOFT_LIMIT_RECOVER_EVERY
+                        and (ceiling <= 0 or soft < ceiling)):
+                    pod.soft_limit = soft + 1
+                    pod.since_limit_change = 0
             # Oldest first: the slots are interchangeable (nothing identifies
             # which page holds which), and dropping the oldest keeps the list a
             # true picture of how long the REMAINING ones have been held, which
@@ -4647,18 +4721,27 @@ class RunpodPodPool:
         finally treated as dead -- see `_recheck_quarantined`.
         """
         now = time.time()
+        with self._cv:
+            if pod not in self.pods:
+                return
+            held = int(pod.inflight)
+        # BEFORE the quarantine, so the width is learned from the load that
+        # actually broke it. This is the part that makes the fleet work at
+        # whatever concurrency a client offers: the pod tells us its width by
+        # refusing, and we stop exceeding it.
+        width = self.note_refusal(pod)
         with self._cv:
             if pod not in self.pods:
                 return
             pod.healthy = False
             pod.quarantined_until = now + QUARANTINE_S
             self._bump_cards()
-            held = int(pod.inflight)
         print(f"[runpod] pod {pod.pod_id} set aside for {QUARANTINE_S:.0f}s: "
               f"{reason}. NOT terminated — a refused connection is as likely to "
               f"mean the pod is overloaded as gone, and it still has {held} "
               f"page(s) charged to it. It is re-probed when the quarantine "
-              f"expires.", flush=True)
+              f"expires, and will then be fed at most {width} at a time.",
+              flush=True)
 
     def forget(self, pod: PodHandle, reason: str = "gone") -> bool:
         """Drop a pod that no longer exists. Like ``discard``, minus the kill.
@@ -5498,6 +5581,19 @@ QUARANTINE_S = 45.0
 #: for nothing. Fewer would start guessing again.
 QUARANTINE_STRIKES = 3
 
+#: Never let a pod's discovered width fall below this. A pod that refused once
+#: is not a pod worth one page at a time -- collapsing it that far would turn a
+#: brief overload into a fleet that cannot drain a queue.
+SOFT_LIMIT_FLOOR = 4
+
+#: Successful pages between one-slot increases. The decrease is multiplicative
+#: and the increase additive, the standard shape, because the two errors are not
+#: symmetric: overshooting costs refused connections, undershooting costs a
+#: little latency. 64 means a pod that was halved from 32 to 16 is back at 32
+#: after ~1000 successful pages -- fast enough to recover from a transient,
+#: slow enough not to re-provoke a real limit.
+SOFT_LIMIT_RECOVER_EVERY = 64
+
 
 def _slots_of(pod) -> list:
     """A pod's in-flight timestamp list, created on demand.
@@ -6320,6 +6416,13 @@ def pods_status() -> list:
                         max(0, int(p.quarantined_until - now))
                         if getattr(p, "quarantined_until", 0) else 0),
                     "probe_failures": int(getattr(p, "probe_failures", 0) or 0),
+                    # What this pod is actually being fed, and what it was
+                    # configured for. A fleet running at half its configured
+                    # width is the single most useful thing to know when
+                    # throughput is below expectation.
+                    "capacity": pool.pod_capacity_now(p) if hasattr(
+                        pool, "pod_capacity_now") else 0,
+                    "soft_limit": int(getattr(p, "soft_limit", 0) or 0),
                     "data_center": getattr(p, "data_center", "") or "",
                     "cloud_type": getattr(p, "cloud_type", "") or "",
                     "pool": (pool.mcfg.pool or "") if getattr(pool, "mcfg", None) else "",

codai/config.py

@@ -1023,6 +1023,18 @@ class OcrConfig:
     # `max_resident_mb` below; this stays what its first line always said it was,
     # a ceiling on concurrent page WORK.
     max_concurrency: int = 64            # global ceiling on in-flight OCR pages
+    # How many pages may be ACCEPTED AND WAITING for a slot before the server
+    # starts refusing new requests with 429. 0 = follow `max_concurrency`.
+    #
+    # The queue is deliberate, not an overflow: when a pod finishes a page there
+    # must already be another page ready to hand it, or the fleet idles between
+    # every dispatch and the pods are paid for the gap. What the queue must not
+    # do is grow without limit -- a client that offers ten times the capacity
+    # then waits longer and longer for each reply, which it reads as "the pages
+    # are getting slower" rather than "stop sending". Past this depth the
+    # honest answer is 429 with a Retry-After, and a client that obeys it gets
+    # its work done sooner than one that keeps pushing.
+    max_queued_pages: int = 0
     # Ceiling on the bytes that in-flight documents may hold in RASTERISED pages,
     # charged before a page exists (codai/ocr/budget.py prices a document from its
     # page tree). 0 = derive it from the cgroup limit or host RAM, which is the
@@ -2441,6 +2453,7 @@ class ConfigManager:
                     ("CODERAI_OCR_OLMOCR_MAX_CONTINUATIONS",
                      "olmocr_max_continuations", int),
                     ("CODERAI_OCR_MAX_CONCURRENCY", "max_concurrency", int),
+                    ("CODERAI_OCR_MAX_QUEUED_PAGES", "max_queued_pages", int),
                     ("CODERAI_OCR_MAX_RESIDENT_MB", "max_resident_mb", int),
                     ("CODERAI_OCR_PAGE_WINDOW", "page_window", int),
                     ("CODERAI_OCR_RASTER_WORKERS", "raster_workers", int),

codai/ocr/manager.py

@@ -28,6 +28,7 @@ bounds the whole subsystem.
 """
 
 import asyncio
+import contextlib
 import os
 import threading
 import time
@@ -1410,6 +1411,15 @@ class OcrManager:
         self._sem: Optional[asyncio.Semaphore] = None
         self._lock = asyncio.Lock()
         self._releaser_registered = False
+        #: Pages holding a slot, and pages waiting for one. Two numbers, because
+        #: they answer different questions: the first is how much work is
+        #: happening, the second is how much has been accepted and is not
+        #: happening yet. Only the second can say "stop sending".
+        self._pages_running = 0
+        self._pages_queued = 0
+        #: Used until `configure` supplies the real one. Cached, because a
+        #: semaphore created per page is not a limit.
+        self._fallback_sem: Optional[asyncio.Semaphore] = None
 
     def configure(self, cfg) -> None:
         """Point the manager at the current OcrConfig. Rebuilds pools if params changed."""
@@ -1808,13 +1818,75 @@ class OcrManager:
         return [n for n in names
                 if n != failed and n in _ENGINE_FACTORIES and _engine_enabled(cfg, n)]
 
+    def _slot_sem(self) -> "asyncio.Semaphore":
+        """The one semaphore every page waits on.
+
+        ONE, cached. The per-page expression this replaced -- `self._sem or
+        asyncio.Semaphore(4)` -- built a FRESH semaphore for every page when the
+        manager had not been configured, so the fallback bounded nothing at all:
+        forty pages of a document all ran at once against a limit of four. It
+        read as a limit in every line of code that mentioned it.
+        """
+        if self._sem is None:
+            if self._fallback_sem is None:
+                self._fallback_sem = asyncio.Semaphore(4)
+            return self._fallback_sem
+        return self._sem
+
+    @contextlib.asynccontextmanager
+    async def _page_slot(self):
+        """Hold one of the install's page slots, counted while waiting for it.
+
+        Every page goes through here so that `admission()` can answer honestly.
+        The waiting count is the whole point: without it a server under more
+        load than it can carry is indistinguishable, from outside, from one
+        working normally -- it simply answers later and later, which a client
+        reads as the pages getting slower rather than as "stop sending".
+        """
+        sem = self._slot_sem()
+        self._pages_queued += 1
+        try:
+            await sem.acquire()
+        except BaseException:
+            self._pages_queued -= 1
+            raise
+        self._pages_queued -= 1
+        self._pages_running += 1
+        try:
+            yield
+        finally:
+            self._pages_running -= 1
+            sem.release()
+
+    def admission(self) -> dict:
+        """What the install is carrying, and whether it can take more.
+
+        ``queue_limit`` is how many pages may be ACCEPTED AND WAITING before a
+        new request is refused. Derived from `max_concurrency` when the operator
+        has not set `max_queued_pages`, because the two are the same kind of
+        number and an install that tunes one should not have to discover the
+        other exists.
+        """
+        cfg = getattr(self._cfg, "max_concurrency", 0) if self._cfg else 0
+        try:
+            width = max(1, int(cfg or 0))
+        except (TypeError, ValueError):
+            width = 1
+        try:
+            qmax = int(getattr(self._cfg, "max_queued_pages", 0) or 0)
+        except (TypeError, ValueError):
+            qmax = 0
+        if qmax <= 0:
+            qmax = width
+        return {"running": self._pages_running, "queued": self._pages_queued,
+                "width": width, "queue_limit": qmax,
+                "full": self._pages_queued >= qmax}
+
     async def _pages_with(self, name: str, images: List, lang: str = "",
                           base: int = 0) -> List[OcrPage]:
         pool = await self._get_pool(name, lang)
-        sem = self._sem or asyncio.Semaphore(4)
-
         async def _one(idx, img):
-            async with sem:
+            async with self._page_slot():
                 page = await pool.recognize(img)
                 # `base` is the document-wide index of this window's first page, so
                 # a windowed read numbers pages the same way a whole-document read
@@ -2339,7 +2411,6 @@ class OcrManager:
                  if is_pdf else [data])
         if not parts:
             return None
-        sem = self._sem or asyncio.Semaphore(4)
         opts = dict(page_opts or {})
         refused: list = []
         unreachable: list = []
@@ -2357,7 +2428,7 @@ class OcrManager:
             every=300.0)
 
         async def _one(idx: int, blob: bytes):
-            async with sem:
+            async with self._page_slot():
                 if refused:
                     # Another page already found the pods cannot run it. Do not
                     # send the rest of the document one failure at a time.
@@ -2598,10 +2669,8 @@ class OcrManager:
         if mode == "off":
             return [], []
 
-        sem = self._sem or asyncio.Semaphore(4)
-
         async def _one(idx, img, page):
-            async with sem:
+            async with self._page_slot():
                 res = await asyncio.to_thread(detect_page, img, page, cfg, mode)
                 for d in res["stamps"]:
                     d["page"] = base + idx

tests/test_ocr_queue_full_is_429.py

+"""A full queue must say so, with 429, not answer slower and slower.
+
+The queue in front of the pods is deliberate: when a pod finishes a page there
+has to be another ready to hand it, or the fleet idles between every dispatch
+and is paid for the gap. So the server carries `ocr.max_concurrency` pages in
+flight PLUS `ocr.max_queued_pages` waiting.
+
+What it did past that point was accept anyway. A client offering several times
+the capacity got no error -- it got replies that took longer and longer, which
+reads as "the pages are getting slower" rather than "stop sending", so it sent
+more. Measured 2026-10-19: 522 pages offered at once against a fleet that could
+carry ~250; mean page latency 238 s against a 19 s service time; nothing in any
+reply said the server was over capacity.
+
+  1. under the limit nothing changes -- the queue is supposed to be full
+  2. past the limit the answer is 429 with an actionable Retry-After
+  3. the refusal happens BEFORE the upload is read
+  4. the accounting can never itself refuse work
+"""
+import asyncio
+
+import pytest
+
+from codai.ocr import manager as ocr_manager
+
+
+class _Cfg:
+    def __init__(self, width=8, queued=0):
+        self.max_concurrency = width
+        self.max_queued_pages = queued
+
+
+def _mgr(width=8, queued=0, running=0, waiting=0):
+    m = ocr_manager.OcrManager.__new__(ocr_manager.OcrManager)
+    m._cfg = _Cfg(width, queued)
+    m._pages_running = running
+    m._pages_queued = waiting
+    m._sem = None
+    return m
+
+
+# --- 1. the queue is meant to be full ------------------------------------
+
+def test_an_idle_server_is_not_full():
+    assert _mgr().admission()["full"] is False
+
+
+def test_a_fleet_at_full_width_with_an_empty_queue_is_not_full():
+    """Every slot busy is the server working, not the server overloaded."""
+    st = _mgr(width=8, running=8, waiting=0).admission()
+    assert st["full"] is False
+
+
+def test_a_partly_filled_queue_is_not_full():
+    """This is the state the queue exists to be in: pods never starved."""
+    st = _mgr(width=8, running=8, waiting=4).admission()
+    assert st["full"] is False
+    assert st["queue_limit"] == 8
+
+
+def test_the_queue_limit_follows_max_concurrency_when_unset():
+    assert _mgr(width=64).admission()["queue_limit"] == 64
+
+
+def test_an_explicit_queue_limit_is_obeyed():
+    assert _mgr(width=64, queued=16).admission()["queue_limit"] == 16
+
+
+# --- 2. past the limit, 429 ----------------------------------------------
+
+def test_a_full_queue_is_full():
+    st = _mgr(width=8, running=8, waiting=8).admission()
+    assert st["full"] is True
+
+
+def test_the_endpoint_raises_429(monkeypatch):
+    from fastapi import HTTPException
+    from codai.api import ocr as api_ocr
+    monkeypatch.setattr(api_ocr.ocr_manager, "admission",
+                        lambda: {"running": 8, "queued": 8, "width": 8,
+                                 "queue_limit": 8, "full": True})
+    with pytest.raises(HTTPException) as got:
+        api_ocr._refuse_if_full(None)
+    assert got.value.status_code == 429
+
+
+def test_the_429_carries_a_retry_after(monkeypatch):
+    """Without it a client can only guess, and guessing means retrying at once
+    -- which makes every accepted page slower, including its own."""
+    from fastapi import HTTPException
+    from codai.api import ocr as api_ocr
+    monkeypatch.setattr(api_ocr.ocr_manager, "admission",
+                        lambda: {"running": 8, "queued": 8, "width": 8,
+                                 "queue_limit": 8, "full": True})
+    with pytest.raises(HTTPException) as got:
+        api_ocr._refuse_if_full(None)
+    hdrs = got.value.headers or {}
+    assert "Retry-After" in hdrs
+    assert int(hdrs["Retry-After"]) >= 1
+
+
+def test_the_429_says_what_the_numbers_are(monkeypatch):
+    """A client tuning its window needs the width and the depth, not a
+    sentence. These are the two numbers it would otherwise have to discover by
+    bisection against a production fleet."""
+    from fastapi import HTTPException
+    from codai.api import ocr as api_ocr
+    monkeypatch.setattr(api_ocr.ocr_manager, "admission",
+                        lambda: {"running": 5, "queued": 9, "width": 8,
+                                 "queue_limit": 9, "full": True})
+    with pytest.raises(HTTPException) as got:
+        api_ocr._refuse_if_full(None)
+    hdrs = got.value.headers or {}
+    assert hdrs["X-OCR-Page-Width"] == "8"
+    assert hdrs["X-OCR-Queue-Limit"] == "9"
+    assert hdrs["X-OCR-Pages-Queued"] == "9"
+
+
+def test_a_server_under_the_limit_is_not_refused(monkeypatch):
+    from codai.api import ocr as api_ocr
+    monkeypatch.setattr(api_ocr.ocr_manager, "admission",
+                        lambda: {"running": 8, "queued": 1, "width": 8,
+                                 "queue_limit": 8, "full": False})
+    assert api_ocr._refuse_if_full(None) is None
+
+
+# --- 3. refused before the body is read ----------------------------------
+
+@pytest.mark.parametrize("fn", ["create_ocr", "create_ocr_batch"])
+def test_both_endpoints_refuse_before_reading_the_upload(fn):
+    """Twice as important as it looks: the client hears at once instead of
+    after transferring a document the server will not look at, and the server
+    does not buffer megabytes it is about to discard. In-flight memory here is
+    concurrent requests times upload size, and that product has OOM-killed it."""
+    import inspect
+    from codai.api import ocr as api_ocr
+    src = inspect.getsource(getattr(api_ocr, fn))
+    assert "_refuse_if_full(cfg)" in src, fn
+    assert src.index("_refuse_if_full(cfg)") < src.index("await f"), fn
+
+
+# --- 4. the accounting must never be the thing that refuses --------------
+
+def test_broken_accounting_does_not_refuse_work(monkeypatch):
+    from codai.api import ocr as api_ocr
+
+    def _boom():
+        raise RuntimeError("counters are confused")
+    monkeypatch.setattr(api_ocr.ocr_manager, "admission", _boom)
+    assert api_ocr._refuse_if_full(None) is None
+
+
+def test_admission_survives_a_config_that_says_nonsense():
+    m = _mgr()
+    m._cfg = type("C", (), {"max_concurrency": "eight",
+                            "max_queued_pages": None})()
+    st = m.admission()
+    assert st["width"] >= 1 and st["queue_limit"] >= 1
+
+
+# --- the counters are real -------------------------------------------------
+
+def test_the_slot_counts_waiting_and_running():
+    async def _run():
+        m = _mgr(width=1)
+        m._sem = asyncio.Semaphore(1)
+        seen = {}
+
+        async def _hold():
+            async with m._page_slot():
+                await asyncio.sleep(0.08)
+
+        async def _wait():
+            async with m._page_slot():
+                pass
+
+        h = asyncio.ensure_future(_hold())
+        w = asyncio.ensure_future(_wait())
+        # Observed from outside both, while the second is blocked on the slot
+        # the first is holding: that is the state the 429 is derived from.
+        await asyncio.sleep(0.03)
+        seen["running"] = m._pages_running
+        seen["queued"] = m._pages_queued
+        await asyncio.gather(h, w)
+        return seen, m
+
+    seen, m = asyncio.run(_run())
+    assert seen["running"] == 1
+    assert seen["queued"] == 1, "a page waiting for a slot was not counted"
+    assert m._pages_running == 0 and m._pages_queued == 0, "counters leaked"
+
+
+def test_a_cancelled_page_does_not_leak_a_queue_slot():
+    """A client that times out in bulk cancels every page of a batch at once.
+    If a cancelled waiter stayed counted, the queue would fill permanently and
+    the server would 429 everything for the rest of its life."""
+    async def _run():
+        m = _mgr(width=1)
+        m._sem = asyncio.Semaphore(1)
+
+        async def _hold():
+            async with m._page_slot():
+                await asyncio.sleep(0.2)
+
+        async def _wait():
+            async with m._page_slot():
+                pass
+
+        h = asyncio.ensure_future(_hold())
+        w = asyncio.ensure_future(_wait())
+        await asyncio.sleep(0.03)
+        assert m._pages_queued == 1
+        w.cancel()
+        try:
+            await w
+        except asyncio.CancelledError:
+            pass
+        await asyncio.sleep(0.01)
+        queued = m._pages_queued
+        h.cancel()
+        try:
+            await h
+        except asyncio.CancelledError:
+            pass
+        return queued
+
+    assert asyncio.run(_run()) == 0
+
+
+def test_the_fallback_semaphore_is_one_semaphore():
+    """Caught by tests/test_ocr_pod_pipeline_gate.py when this was first
+    written: `self._sem or asyncio.Semaphore(4)` built a FRESH semaphore per
+    page, so an unconfigured manager bounded nothing -- forty pages of one
+    document ran at once against what every line of code called a limit of
+    four."""
+    m = _mgr()
+    m._sem = None
+    m._fallback_sem = None
+    assert m._slot_sem() is m._slot_sem()
+    src = open("codai/ocr/manager.py").read()
+    assert "self._sem or asyncio.Semaphore(4)" not in src
+
+
+def test_a_configured_semaphore_wins_over_the_fallback():
+    m = _mgr()
+    real = asyncio.Semaphore(3)
+    m._sem = real
+    m._fallback_sem = asyncio.Semaphore(99)
+    assert m._slot_sem() is real
+
+
+def test_every_page_path_goes_through_the_counted_slot():
+    """An uncounted path is a queue the server cannot see, so 429 would never
+    fire for the traffic that matters."""
+    src = open("codai/ocr/manager.py").read()
+    assert "async with sem:" not in src
+    assert src.count("async with self._page_slot():") >= 3

tests/test_runpod_adaptive_pod_width.py

+"""A client may offer any concurrency; the pool must only ever send a pod what
+that pod will take.
+
+`max_inflight_per_pod` defaults to 0 -- unlimited -- so out of the box the pool
+dispatches every page it is handed. A client that offers more than a pod can
+serve therefore does NOT queue in the orchestrator, where the queue would be
+counted, reported as demand and bounded. It queues in the POD'S LISTEN BACKLOG,
+and an overflowing backlog makes a perfectly healthy pod refuse connections.
+
+Measured 2026-10-19: a client offering 64 pages per pod, against pods serving
+~32 (`Running: 29-32 reqs, Waiting: 0`), produced refusals that were read as
+dead pods and cost eight A40s. Nothing in any configuration said 64 was too
+many, and nothing could have: the right number is a property of the pod, the
+model and the card, not of the install.
+
+So the pod is asked. A refusal halves what we send it, FROM THE IN-FLIGHT COUNT
+AT THE REFUSAL -- the observed breaking point rather than a guess -- and
+successes walk it back up one slot at a time. Multiplicative decrease, additive
+increase, because the errors are not symmetric: overshooting costs refused
+connections and a misdiagnosed fleet, undershooting costs a little latency.
+
+The properties pinned here:
+
+  1. a refusal narrows that pod, and narrows it from what it was carrying
+  2. the narrowing is what `_pick` enforces, or it is decoration
+  3. it never collapses to uselessness, and never exceeds the operator's own
+     ceiling
+  4. an install this never fires on behaves exactly as before
+  5. a narrowed pod recovers by itself
+"""
+import threading
+import time
+
+import pytest
+
+from codai.api import runpod_worker as rw
+
+
+def _pool(max_inflight=0, max_pods=8):
+    class _Acct:
+        enabled = True
+
+    p = rw.RunpodPodPool.__new__(rw.RunpodPodPool)
+    p._cv = threading.Condition()
+    p.model_key = "m"
+    p._closed = False
+    p._booting = 0
+    p._waiting = 0
+    p._surplus_since = 0.0
+    p._denied = []
+    p.pods = []
+    p._affinity = {}
+    cfg = rw.RunpodModelConfig()
+    cfg.max_pods = max_pods
+    cfg.max_inflight_per_pod = max_inflight
+    p.mcfg = cfg
+    p._bump_cards = lambda: None
+    p._say_no_pod = lambda *a, **k: None
+    return p
+
+
+def _pod(pod_id="p1", inflight=0):
+    return rw.PodHandle(pod_id=pod_id, url="https://10.0.0.1:2222",
+                        hourly_usd=0.59, started_at=time.time(), gpu="A40",
+                        inflight=inflight)
+
+
+# --- 1. the pod's own behaviour sets its width ---------------------------
+
+def test_a_refusal_narrows_this_pod():
+    p = _pool(max_inflight=0)
+    pod = _pod(inflight=58)
+    p.pods = [pod]
+    new = p.note_refusal(pod)
+    assert new == 29, "halve the OBSERVED load, not a configured number"
+    assert pod.soft_limit == 29
+
+
+def test_the_live_numbers():
+    """64 configured, 58 in flight when it refused, pods serving ~32."""
+    p = _pool(max_inflight=64)
+    pod = _pod(inflight=58)
+    p.pods = [pod]
+    p.note_refusal(pod)
+    assert p.pod_capacity_now(pod) == 29
+    assert 24 <= p.pod_capacity_now(pod) <= 34, "lands near the measured width"
+
+
+def test_a_second_refusal_narrows_further():
+    p = _pool()
+    pod = _pod(inflight=32)
+    p.pods = [pod]
+    assert p.note_refusal(pod) == 16
+    pod.inflight = 16
+    assert p.note_refusal(pod) == 8
+
+
+def test_a_refusal_with_nothing_in_flight_still_narrows():
+    """The counter can be stale; the refusal is still real."""
+    p = _pool()
+    pod = _pod(inflight=0)
+    p.pods = [pod]
+    assert p.note_refusal(pod) >= rw.SOFT_LIMIT_FLOOR
+
+
+def test_narrowing_is_per_pod():
+    """One overloaded pod must not throttle the whole fleet."""
+    p = _pool()
+    a, b = _pod("a", inflight=40), _pod("b")
+    p.pods = [a, b]
+    p.note_refusal(a)
+    assert p.pod_capacity_now(a) == 20
+    assert p.pod_capacity_now(b) == 0, "untouched pod keeps the pool's setting"
+
+
+# --- 2. _pick enforces it -----------------------------------------------
+
+def test_pick_refuses_a_pod_that_is_at_its_discovered_width():
+    p = _pool(max_inflight=0)
+    pod = _pod(inflight=40)
+    p.pods = [pod]
+    p.note_refusal(pod)                 # width now 20, inflight 40
+    assert p._pick([pod], "") is None, "dispatched past the discovered width"
+
+
+def test_pick_uses_a_pod_below_its_discovered_width():
+    p = _pool(max_inflight=0)
+    pod = _pod(inflight=40)
+    p.pods = [pod]
+    p.note_refusal(pod)
+    pod.inflight = 5
+    assert p._pick([pod], "") is pod
+
+
+def test_pick_reads_the_narrowed_capacity_not_the_raw_one():
+    import inspect
+    src = inspect.getsource(rw.RunpodPodPool._pick)
+    assert "self.pod_capacity_now(p)" in src
+    assert "cap = self.pod_capacity(p)" not in src
+
+
+# --- 3. bounded both ends ------------------------------------------------
+
+def test_it_never_collapses_to_uselessness():
+    """A pod worth one page at a time cannot drain a queue."""
+    p = _pool()
+    pod = _pod(inflight=2)
+    p.pods = [pod]
+    for _ in range(20):
+        p.note_refusal(pod)
+        pod.inflight = pod.soft_limit
+    assert pod.soft_limit == rw.SOFT_LIMIT_FLOOR
+    assert rw.SOFT_LIMIT_FLOOR >= 2
+
+
+def test_it_never_exceeds_the_operator_ceiling():
+    """The discovered width narrows the configured one; it cannot widen it."""
+    p = _pool(max_inflight=16)
+    pod = _pod(inflight=8)
+    p.pods = [pod]
+    pod.soft_limit = 999
+    assert p.pod_capacity_now(pod) <= 16
+
+
+def test_recovery_stops_at_the_operator_ceiling():
+    p = _pool(max_inflight=16)
+    pod = _pod()
+    p.pods = [pod]
+    pod.soft_limit = 16
+    for _ in range(rw.SOFT_LIMIT_RECOVER_EVERY * 3):
+        pod.inflight = 1
+        p.release(pod)
+    assert pod.soft_limit == 16
+
+
+# --- 4. nothing changes for an install this never fires on --------------
+
+def test_an_untouched_pod_is_unbounded_when_the_operator_said_so():
+    p = _pool(max_inflight=0)
+    pod = _pod(inflight=10000)
+    p.pods = [pod]
+    assert p.pod_capacity_now(pod) == 0
+    assert p._pick([pod], "") is pod
+
+
+def test_an_untouched_pod_keeps_its_configured_capacity():
+    p = _pool(max_inflight=32)
+    pod = _pod()
+    p.pods = [pod]
+    assert p.pod_capacity_now(pod) == p.pod_capacity(pod)
+
+
+def test_release_does_not_invent_a_limit():
+    """A pod that was never narrowed must not acquire a width from succeeding."""
+    p = _pool(max_inflight=0)
+    pod = _pod(inflight=1)
+    p.pods = [pod]
+    for _ in range(rw.SOFT_LIMIT_RECOVER_EVERY * 2):
+        pod.inflight = 1
+        p.release(pod)
+    assert pod.soft_limit == 0
+
+
+# --- 5. a narrowed pod recovers -----------------------------------------
+
+def test_successes_walk_the_width_back_up():
+    p = _pool(max_inflight=64)
+    pod = _pod(inflight=32)
+    p.pods = [pod]
+    p.note_refusal(pod)
+    assert pod.soft_limit == 16
+    for _ in range(rw.SOFT_LIMIT_RECOVER_EVERY):
+        pod.inflight = 1
+        p.release(pod)
+    assert pod.soft_limit == 17
+
+
+def test_recovery_is_additive_not_immediate():
+    """Jumping straight back to the configured number would re-provoke the
+    limit that was just discovered."""
+    p = _pool(max_inflight=64)
+    pod = _pod(inflight=32)
+    p.pods = [pod]
+    p.note_refusal(pod)
+    for _ in range(rw.SOFT_LIMIT_RECOVER_EVERY - 1):
+        pod.inflight = 1
+        p.release(pod)
+    assert pod.soft_limit == 16, "moved before enough pages had succeeded"
+
+
+def test_a_refusal_resets_the_recovery_counter():
+    p = _pool(max_inflight=64)
+    pod = _pod(inflight=32)
+    p.pods = [pod]
+    p.note_refusal(pod)
+    for _ in range(rw.SOFT_LIMIT_RECOVER_EVERY - 1):
+        pod.inflight = 1
+        p.release(pod)
+    pod.inflight = 16
+    p.note_refusal(pod)
+    assert pod.since_limit_change == 0
+
+
+# --- the operator can see a narrowed fleet ------------------------------
+
+def test_the_snapshot_reports_the_width_in_force():
+    """A fleet running at half its configured width is the first thing to know
+    when throughput is below expectation."""
+    src = open("codai/api/runpod_worker.py").read()
+    assert '"soft_limit": int(getattr(p, "soft_limit", 0) or 0)' in src
+    assert '"capacity": pool.pod_capacity_now(p)' in src
+
+
+def test_discard_learns_the_width_before_setting_the_pod_aside():
+    """Afterwards it would read an in-flight count the quarantine had already
+    begun to unwind, and learn the wrong number."""
+    import inspect
+    src = inspect.getsource(rw.RunpodPodPool.discard)
+    assert src.index("note_refusal") < src.index("quarantined_until = now")