a dead socket is not a pod refusing the pipeline (v0.4.6)

A pooled keep-alive connection that a pod closes while idle fails on its
next use:

  pod OCR at https://.../v1/ocr is unreachable: ('Connection aborted.',
  RemoteDisconnected('Remote end closed connection without response'))

_post_page_to_pod turned every transport failure into OcrError(503), 503 is
in _POD_CANNOT_RUN_PIPELINE, so one dead TCP connection was read as "these
pods cannot run the OCR pipeline" and set the whole pool aside for five
minutes. It happened twelve times in four hours on a production run: the
GPU-less orchestrator rasterised every page of every document for those
windows and climbed to 40 GB at ~0.8 GB/min, while sixteen rented A40s sat
idle and willing.

An ANSWER may condemn an image; a socket may not. So:

* PodUnreachable, a distinct error for "nothing came back", which never sets
  the pool aside. A pod that ANSWERS 404/501/503 still does.
* One page is retried on a fresh connection (nothing was sent, so there is no
  side effect) and then on another pod — but a pod is dropped from the pool
  only after it fails to answer /healthz. Writing a pod off because an idle
  connection was reused has cost seven healthy cards before.
* A timeout is NOT retried: the pod may be ninety seconds into a 130-second
  read, and sending the page again doubles the work on that card.
* An active cooldown is no longer extended. It used to be, silently: every
  further refusal pushed the expiry out while only the first one printed, so
  a five-minute window could last twenty with nothing in the log.

Two more ways to end up local, both of which were SILENT:

* The gate compared the request's engine against the three VLM engine names.
  /v1/ocr has always also accepted the MODEL a caller knows in place of the
  engine driving it — resolve_engine maps the Surya-2 checkpoint to 'surya'
  and the catalogue advertises those ids as aliases — so a caller naming the
  model got no pod-side pipeline and no message. The switch has to be read
  the way a READ reads it, for the same reason it has to be read the way a
  LAUNCH reads it.
* Serve mode not 'model', an empty model id, an engine with no pod pipeline,
  a standing refusal: each returned "" or None without a word. The log of the
  run that prompted this holds 8,278 successful pod reads and nothing at all
  explaining the thousands of documents in between. Every one of them now
  says which way the decision went, rate limited rather than said once,
  because a cooldown is minutes and a run is hours.

And, since reading a page here means rasterising here, ocr.pod_pipeline:

  prefer  (default) wait up to ocr.pod_pipeline_wait_s for a pod, then read
          it here — a pool with nothing free is usually one still booting.
          A timed-out wait is not paid again for five minutes, so the ceiling
          cannot become a per-page tax.
  auto    read it here at once (0.4.1-0.4.5 behaviour).
  require never rasterise on this box: 503, saying why.

Tests: 32 new (tests/test_ocr_pod_pipeline_policy.py) covering the alias, all
four silent routes, the three policies and the transport/answer line.
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.5"
+__version__ = "0.4.6"
 __tagline__ = "Complete orchestration, distribution and escalation of remotizable advanced inference."
 
 # Configure the CUDA caching allocator BEFORE torch is imported anywhere.

codai/admin/routes.py

@@ -4136,6 +4136,8 @@ def build_settings_dict(c, gpu_cards):
             "surya_dpi": c.ocr.surya_dpi,
             "max_concurrency": c.ocr.max_concurrency,
             "max_pages": c.ocr.max_pages,
+            "pod_pipeline": c.ocr.pod_pipeline,
+            "pod_pipeline_wait_s": c.ocr.pod_pipeline_wait_s,
             "max_resident_mb": c.ocr.max_resident_mb,
             "page_window": c.ocr.page_window,
             "raster_workers": c.ocr.raster_workers,
@@ -4987,6 +4989,14 @@ async def api_save_settings(request: Request, username: str = Depends(require_ad
         if "max_concurrency" in d: o.max_concurrency = _i("max_concurrency", o.max_concurrency, 1)
         # 0 = derive from the cgroup limit or host RAM; see codai/ocr/budget.py
         if "max_pages" in d: o.max_pages = _i("max_pages", o.max_pages, 0)
+        if "pod_pipeline" in d:
+            _pp = str(d.get("pod_pipeline") or "").strip().lower()
+            o.pod_pipeline = _pp if _pp in ("auto", "prefer", "require") else o.pod_pipeline
+        if "pod_pipeline_wait_s" in d:
+            try:
+                o.pod_pipeline_wait_s = max(0.0, float(d.get("pod_pipeline_wait_s") or 0))
+            except (TypeError, ValueError):
+                pass
         if "max_resident_mb" in d: o.max_resident_mb = _i("max_resident_mb", o.max_resident_mb, 0)
         if "page_window" in d: o.page_window = _i("page_window", o.page_window, 0)
         if "raster_workers" in d: o.raster_workers = _i("raster_workers", o.raster_workers, 0)

codai/admin/templates/settings.html

@@ -1131,6 +1131,25 @@
         over-long document is refused in milliseconds with a 413 naming its page
         count &mdash; never a silent or empty read.</span>
       </div>
+      <div class="form-group" style="flex:1;min-width:170px">
+        <label class="form-label">When no pod can take a page</label>
+        <select id="s-ocr-pod-pipeline" class="form-input">
+          <option value="prefer">Wait for a pod, then read it here</option>
+          <option value="auto">Read it here straight away</option>
+          <option value="require">Refuse the page (never read it here)</option>
+        </select>
+        <span class="form-hint">Only applies where the pods&rsquo; image carries the OCR
+        pipeline. Reading a page here means <b>rasterising</b> here, which is what every
+        OCR memory incident has begun with &mdash; so the default waits for a pod first,
+        and says in the log whenever it gives up and reads one locally anyway.</span>
+      </div>
+      <div class="form-group" style="flex:1;min-width:130px">
+        <label class="form-label">Wait for a pod (s)</label>
+        <input type="number" id="s-ocr-pod-pipeline-wait" class="form-input" min="0" placeholder="60">
+        <span class="form-hint">How long to wait before giving up on a pod. Covers a pool
+        that is momentarily full, not a cold one: after a timeout the next pages go
+        straight through for five minutes rather than paying this per page.</span>
+      </div>
       <div class="form-group" style="flex:1;min-width:140px">
         <label class="form-label">Max resident pages (MB)</label>
         <input type="number" id="s-ocr-max-resident-mb" class="form-input" min="0" placeholder="0">
@@ -2255,6 +2274,8 @@ async function loadSettings(){
     document.getElementById('s-ocr-surya-dpi').value = ocr.surya_dpi ?? 250;
     document.getElementById('s-ocr-max-concurrency').value = ocr.max_concurrency ?? 64;
     document.getElementById('s-ocr-max-pages').value = ocr.max_pages ?? 0;
+    document.getElementById('s-ocr-pod-pipeline').value = ocr.pod_pipeline ?? 'prefer';
+    document.getElementById('s-ocr-pod-pipeline-wait').value = ocr.pod_pipeline_wait_s ?? 60;
     document.getElementById('s-ocr-max-resident-mb').value = ocr.max_resident_mb ?? 0;
     document.getElementById('s-ocr-page-window').value = ocr.page_window ?? 0;
     document.getElementById('s-ocr-raster-workers').value = ocr.raster_workers ?? 0;
@@ -2560,6 +2581,8 @@ async function saveSettings(silent){
       surya_dpi: parseInt(document.getElementById('s-ocr-surya-dpi').value) || 0,
       max_concurrency: parseInt(document.getElementById('s-ocr-max-concurrency').value) || 64,
       max_pages: parseInt(document.getElementById('s-ocr-max-pages').value) || 0,
+      pod_pipeline: document.getElementById('s-ocr-pod-pipeline').value || 'prefer',
+      pod_pipeline_wait_s: parseFloat(document.getElementById('s-ocr-pod-pipeline-wait').value) || 0,
       max_resident_mb: parseInt(document.getElementById('s-ocr-max-resident-mb').value) || 0,
       page_window: parseInt(document.getElementById('s-ocr-page-window').value) || 0,
       raster_workers: parseInt(document.getElementById('s-ocr-raster-workers').value) || 0,

codai/config.py

@@ -1048,6 +1048,29 @@ class OcrConfig:
     # It must never look like an empty or failed read -- see the "a degraded
     # page still answers 200" trap this subsystem already carries.
     max_pages: int = 0
+    # What to do when the model's pods are the ones that run the OCR pipeline
+    # (their image carries the rasteriser) but no pod can take a page.
+    #
+    #   auto    -- read it here at once. What 0.4.1-0.4.5 did, kept because an
+    #              install with a GPU of its own may genuinely prefer it.
+    #   prefer  -- wait up to `pod_pipeline_wait_s` for a pod, then read it
+    #              here. The default, because a pool with nothing free this
+    #              second is usually a pool whose pods are still booting.
+    #   require -- never rasterise on this box: answer 503 and say why.
+    #
+    # The default is not a performance setting, it is a memory one. A GPU-less
+    # orchestrator that quietly takes the pipeline back holds every page it
+    # renders: measured at ~0.8 GB/min and 36 GB on a 125 GB box, with sixteen
+    # rented cards idle, because a caller named the model instead of the engine
+    # and the gate said nothing. Waiting a minute is cheaper than that in every
+    # install where the pods exist at all.
+    pod_pipeline: str = "prefer"
+    # How long `prefer` (and `require`) wait for a pod before giving up on one.
+    # A pod takes minutes to boot, so this is deliberately NOT long enough to
+    # cover a cold pool: it covers a pool that is momentarily full. After a
+    # wait times out the next pages go straight through for a cooldown rather
+    # than paying the ceiling one page at a time.
+    pod_pipeline_wait_s: float = 60.0
     # How many pages of one document are rasterised at a time. The document is read
     # in windows, so residency is O(window) instead of O(document): a 400-page
     # filing costs the same as an 8-page one. 0 = follow `max_concurrency` (capped
@@ -2053,6 +2076,8 @@ class ConfigManager:
                 "max_concurrency": self.config.ocr.max_concurrency,
                 "max_resident_mb": self.config.ocr.max_resident_mb,
                 "max_pages": self.config.ocr.max_pages,
+                "pod_pipeline": self.config.ocr.pod_pipeline,
+                "pod_pipeline_wait_s": self.config.ocr.pod_pipeline_wait_s,
                 "page_window": self.config.ocr.page_window,
                 "raster_workers": self.config.ocr.raster_workers,
                 "lang": self.config.ocr.lang,

codai/ocr/manager.py

@@ -348,6 +348,73 @@ def _warn_if_oversubscribed(name: str, pages: int) -> None:
               f"pods ({per} in flight each); {tail}.", flush=True)
 
 
+class PodUnreachable(OcrError):
+    """Nothing came back over the socket — which is NOT an answer.
+
+    The distinction decides what may be concluded. A pod that ANSWERS 404 or
+    503 has told us about the image it runs: it cannot run the pipeline, it
+    will say the same for every page, and the pool belongs on the shelf. A
+    socket that dies has told us about one TCP connection.
+
+    Conflating the two cost a production run its afternoon. A pooled keep-alive
+    connection that a pod closed while idle fails on its next use with
+    RemoteDisconnected; that became `OcrError(503)`, 503 meant "these pods
+    cannot run the pipeline", and one dead socket moved every document onto the
+    orchestrator for five minutes. It happened twelve times in four hours, the
+    log said it once, and a GPU-less box climbed to 38 GB rasterising work that
+    sixteen rented A40s were idle and willing to do.
+    """
+
+
+#: Attempts per page, counting the first. Covers a dead pooled socket (nothing
+#: was sent, so a retry has no side effect) and a pod that has genuinely gone,
+#: which the next attempt's placement routes around.
+POD_PAGE_ATTEMPTS = 3
+
+#: Text that means "the connection failed", as opposed to a reply we dislike.
+#: Matched on the message because requests wraps urllib3 which wraps the
+#: socket, and the useful name is rarely the outermost one.
+_DEAD_SOCKET_HINTS = ("remotedisconnected", "connection aborted",
+                      "connection reset", "connection refused",
+                      "protocolerror", "badstatusline", "broken pipe",
+                      "closed connection", "connectionerror",
+                      "connection error", "ssleoferror", "eof occurred")
+
+
+def _looks_like_a_dead_socket(exc: BaseException) -> bool:
+    """A failed CONNECTION, not a timeout and not a reply.
+
+    Timeouts are deliberately excluded: a pod may be ninety seconds into a
+    130-second read, and sending the page again would double the work on a card
+    that is already doing it.
+    """
+    text = f"{type(exc).__name__}: {exc}".lower()
+    if "timeout" in text or "timed out" in text:
+        return False
+    return any(hint in text for hint in _DEAD_SOCKET_HINTS)
+
+
+def _pod_still_alive(base: str, api_key: str) -> bool:
+    """Does this pod answer /healthz at all? Asked BEFORE calling it dead.
+
+    A pod reclaimed out of band and a pod whose idle socket we reused fail
+    identically on one request. Treating the second as the first terminated
+    seven healthy A40s in an earlier incident, so the address gets a question of
+    its own before anyone writes it off.
+    """
+    from codai.api import pod_http
+    url = str(base or "").rstrip("/")
+    if url.endswith("/v1"):
+        url = url[:-3]
+    url = url.rstrip("/") + "/healthz"
+    headers = {"Authorization": f"Bearer {api_key}"} if api_key else {}
+    try:
+        r = pod_http.get(url, headers=headers, timeout=15)
+    except Exception:
+        return False
+    return int(getattr(r, "status_code", 0) or 0) < 500
+
+
 def _post_page_to_pod(base: str, api_key: str, blob: bytes, name: str,
                       dpi: int, opts: dict, detect, engine: str = "surya") -> dict:
     """POST one page to a pod's own /v1/ocr and return its response dict.
@@ -379,11 +446,20 @@ def _post_page_to_pod(base: str, api_key: str, blob: bytes, name: str,
     headers = {}
     if api_key:
         headers["Authorization"] = f"Bearer {api_key}"
-    try:
-        r = pod_http.post(url, files={"file": (name, blob, "application/pdf")},
-                          data=fields, headers=headers, timeout=600)
-    except Exception as e:
-        raise OcrError(f"pod OCR at {url} is unreachable: {e}", status=503)
+    r = None
+    for attempt in range(1, POD_PAGE_ATTEMPTS + 1):
+        try:
+            r = pod_http.post(url, files={"file": (name, blob, "application/pdf")},
+                              data=fields, headers=headers, timeout=600)
+            break
+        except Exception as e:
+            # A pooled connection the pod closed while idle fails on its next
+            # use with nothing having been sent, so one more attempt costs a
+            # connect and gets a new socket.
+            if attempt < POD_PAGE_ATTEMPTS and _looks_like_a_dead_socket(e):
+                continue
+            raise PodUnreachable(f"pod OCR at {url} is unreachable: {e}",
+                                 status=503)
     if r.status_code >= 400:
         body = ""
         try:
@@ -454,7 +530,14 @@ def _set_aside_pod_pipeline(mid: str, reason: str) -> bool:
     with _POD_PIPELINE_LOCK:
         hit = _POD_PIPELINE_UNUSABLE.get(mid)
         fresh = not hit or hit[0] <= now
-        _POD_PIPELINE_UNUSABLE[mid] = (now + POD_PIPELINE_COOLDOWN_S, reason)
+        if fresh:
+            _POD_PIPELINE_UNUSABLE[mid] = (now + POD_PIPELINE_COOLDOWN_S, reason)
+        # An ACTIVE cooldown is not extended. It used to be, silently: every
+        # further refusal pushed the expiry out while only the first one
+        # printed, so a pool set aside for five minutes could stay aside for
+        # twenty with nothing in the log between. The window is the window;
+        # when it ends the pods are asked again, and _read_on_pods says, at
+        # most once per window, that this box is rasterising until then.
     return fresh
 
 
@@ -495,6 +578,166 @@ def _say_pipeline_gate(name: str, mid: str, image: str, runs: bool) -> None:
               f"model's runpod block has an `engine` or an `image`", flush=True)
 
 
+#: (reason key) -> when it was last said. A reason that recurs per document
+#: would bury the log at a hundred documents a minute, and one said once per
+#: process goes stale in a run that lasts hours -- so every "why this went
+#: local" line is rate limited rather than deduplicated outright.
+_SAID_AT: dict = {}
+_SAID_LOCK = threading.Lock()
+
+
+def _say_rate_limited(key, message: str, every: float = 60.0) -> None:
+    """Print ``message`` at most once per ``every`` seconds for this ``key``."""
+    now = time.time()
+    with _SAID_LOCK:
+        last = _SAID_AT.get(key, 0.0)
+        if now - last < max(0.0, every):
+            return
+        _SAID_AT[key] = now
+    print(message, flush=True)
+
+
+def _vlm_engine_for(cfg, name: str) -> str:
+    """Which VLM engine a request means, resolving a MODEL ID like a read does.
+
+    `/v1/ocr` accepts the model a caller knows about in place of the engine
+    name driving it -- `resolve_engine` has mapped the Surya-2 checkpoint to
+    'surya' since the endpoint existed, and the catalogue advertises those ids
+    as aliases. The pod-side gate did NOT: it compared the raw string against
+    the three engine names, so a caller naming the model got "" -- no pod-side
+    pipeline, no message, and this box rasterising every page of every
+    document. The switch has to be read the way a READ reads it, for the same
+    reason it has to be read the way a LAUNCH reads it.
+    """
+    want = (name or "").strip().lower()
+    if want in ("surya", "chandra", "olmocr"):
+        return want
+    if not want:
+        return ""
+    for eng in ("surya", "chandra", "olmocr"):
+        for field in (f"{eng}_model_id", f"{eng}_model"):
+            got = str(getattr(cfg, field, "") or "").strip().lower()
+            if got and got == want:
+                return eng
+    return ""
+
+
+def _pod_pipeline_policy(cfg) -> tuple:
+    """(policy, wait_seconds) for what to do when no pod can take a page.
+
+    * ``auto``    -- read it on this box at once (what 0.4.1-0.4.5 did).
+    * ``prefer``  -- wait up to the ceiling for a pod, then read it here. The
+      default: a pool that cannot serve this second is usually a pool whose
+      pods are booting, and the alternative is this box rasterising, which is
+      what every OCR memory incident on record has been.
+    * ``require`` -- never rasterise here; answer 503 saying why.
+    """
+    raw = str(getattr(cfg, "pod_pipeline", "") or "prefer").strip().lower()
+    if raw in ("off", "never", "local"):
+        raw = "auto"
+    if raw not in ("auto", "prefer", "require"):
+        raw = "prefer"
+    try:
+        wait = float(getattr(cfg, "pod_pipeline_wait_s", 60.0) or 0.0)
+    except Exception:
+        wait = 60.0
+    return raw, max(0.0, wait)
+
+
+#: model id -> when waiting for a pod may be tried again. A pool that is out of
+#: budget, or at its ceiling, stays that way for minutes: without this every
+#: page of every document would pay the full wait before going local, turning a
+#: 60s ceiling into a 60s tax per page.
+_POD_WAIT_COOLDOWN: dict = {}
+POD_WAIT_COOLDOWN_S = 300.0
+
+
+def _pod_wait_cooling(mid: str) -> bool:
+    """True when a wait for this model's pods just timed out."""
+    with _POD_PIPELINE_LOCK:
+        until = _POD_WAIT_COOLDOWN.get(mid, 0.0)
+        if until <= time.time():
+            _POD_WAIT_COOLDOWN.pop(mid, None)
+            return False
+        return True
+
+
+def _set_pod_wait_cooldown(mid: str) -> None:
+    with _POD_PIPELINE_LOCK:
+        _POD_WAIT_COOLDOWN[mid] = time.time() + POD_WAIT_COOLDOWN_S
+
+
+async def _lease_a_pod(model_id: str, policy: str, wait_s: float):
+    """A lease on a pod that runs the pipeline, or None to read on this box.
+
+    Raises OcrError(503) under ``require``, which is the whole point of that
+    setting: an install that has decided this box must never rasterise would
+    rather have a page refused, and retried, than have the orchestrator quietly
+    take on the work it has no memory for.
+    """
+    from codai.models import placement
+    waiting = policy in ("prefer", "require") and not (
+        policy == "prefer" and _pod_wait_cooling(model_id))
+    deadline = time.time() + wait_s if waiting else 0.0
+    waited = 0.0
+    while True:
+        try:
+            lease = placement.choose(
+                model_id, spill_after=placement.spill_threshold(model_id))
+        except OcrError:
+            raise
+        except Exception as exc:
+            raise OcrError(f"no upstream for {model_id}: {exc}", status=503)
+        if lease.kind != "local":
+            if waited:
+                print(f"[ocr] a pod for {model_id!r} became free after "
+                      f"{waited:.0f}s — the page goes there and this box does "
+                      f"not rasterise it", flush=True)
+            return lease
+        # Placement would have this box serve the page, which means rasterising
+        # here: the condition that every OCR memory incident on record began
+        # with. Say which it is, then act on the policy.
+        lease.release()
+        why = ""
+        try:
+            why = placement.why_no_candidates(model_id)
+        except Exception:
+            why = ""
+        why = why or "the pool had no pod to hand out"
+        now = time.time()
+        if now < deadline:
+            await asyncio.sleep(min(2.0, max(0.1, deadline - now)))
+            waited = time.time() + wait_s - deadline
+            continue
+        if policy == "require":
+            raise OcrError(
+                f"the pods for {model_id!r} run the OCR pipeline and this "
+                f"install sets ocr.pod_pipeline=require, so this box will not "
+                f"rasterise the page: no pod could take it"
+                + (f" after {wait_s:.0f}s" if wait_s else "")
+                + f" — {why}. Retry, raise the pool's ceiling or its budget, "
+                f"or set ocr.pod_pipeline=prefer to let this box read it.",
+                status=503)
+        if waiting:
+            _set_pod_wait_cooldown(model_id)
+            _say_rate_limited(
+                ("wait-timeout", model_id),
+                f"[ocr] waited {wait_s:.0f}s for a pod that runs the pipeline "
+                f"for {model_id!r} and none came ({why}) — reading on THIS BOX, "
+                f"which means rasterising here, and not waiting again for "
+                f"{POD_WAIT_COOLDOWN_S:.0f}s. Watch this process's memory.",
+                every=POD_WAIT_COOLDOWN_S)
+        else:
+            _say_rate_limited(
+                ("placement-local", model_id),
+                f"[ocr] placement answered 'local' for {model_id!r}, so no "
+                f"pod is available to run the pipeline ({why}) — THIS BOX "
+                f"rasterises and reads the "
+                f"page. Set ocr.pod_pipeline=prefer to wait for a pod first, "
+                f"or require to refuse the page instead.")
+        return None
+
+
 def pod_pipeline_model(cfg, engine: str = "") -> str:
     """The model id whose pods run the OCR pipeline themselves, or "".
 
@@ -514,20 +757,56 @@ def pod_pipeline_model(cfg, engine: str = "") -> str:
     Returns "" for every other case -- no model mode, no entry, a different
     image -- so the caller simply does what it has always done.
     """
-    name = (engine or "").strip().lower() or str(
+    raw = (engine or "").strip().lower() or str(
         getattr(cfg, "default_engine", "") or "").strip().lower()
+    # A caller may name the MODEL instead of the engine -- documented, accepted
+    # everywhere else, and silently fatal here until 0.4.6. See _vlm_engine_for.
+    name = _vlm_engine_for(cfg, raw)
+    if name and name != raw:
+        _say_rate_limited(
+            ("alias", raw, name),
+            f"[ocr] {raw!r} names the {name!r} engine on this install, so the "
+            f"pod-side pipeline applies to it", every=600.0)
     # All three VLM engines can run on the pod, not just surya. The image is
     # what differs: surya needs its client venv baked in, while chandra and
     # olmOCR in served mode speak only HTTP (PIL + requests, already present),
     # so for them the pod side costs nothing beyond the rasteriser. Leaving
     # them on the orchestrator while surya moved was an asymmetry, not a
     # design.
-    if name not in ("surya", "chandra", "olmocr"):
+    if not name:
+        # SAY SO. Returning "" in silence here is what let a whole production
+        # run rasterise on the orchestrator with 8,278 successful pod reads in
+        # the same log: nothing distinguishes "the pods are doing it" from "the
+        # gate did not recognise the engine" when neither prints anything.
+        if raw in _ENGINE_FACTORIES:
+            _say_rate_limited(
+                ("no-pod-pipeline", raw),
+                f"[ocr] {raw!r} has no pod-side pipeline (only surya, chandra "
+                f"and olmOCR run on the pods), so THIS BOX rasterises and "
+                f"reads every page it is given", every=600.0)
+        else:
+            _say_rate_limited(
+                ("unknown-engine", raw),
+                f"[ocr] {raw!r} is not an OCR engine or a model id this "
+                f"install serves, so the pod-side pipeline cannot be chosen "
+                f"for it and THIS BOX rasterises", every=600.0)
         return ""
-    if str(getattr(cfg, f"{name}_serve", "") or "").strip().lower() != "model":
+    serve = str(getattr(cfg, f"{name}_serve", "") or "").strip().lower()
+    if serve != "model":
+        _say_rate_limited(
+            ("serve-mode", name, serve),
+            f"[ocr] {name!r}: serve mode is {serve or 'unset'!r}, not 'model', "
+            f"so there is no served model for a pod to run the pipeline "
+            f"against — THIS BOX rasterises. Set {name}_serve to 'model' to "
+            f"move it to the pods", every=600.0)
         return ""
     mid = str(getattr(cfg, f"{name}_model_id", "") or "").strip()
     if not mid:
+        _say_rate_limited(
+            ("no-model-id", name),
+            f"[ocr] {name!r}: serve mode is 'model' but {name}_model_id is "
+            f"empty, so there is nothing for the pods to serve — THIS BOX "
+            f"rasterises", every=600.0)
         return ""
     try:
         from codai.api.runpod_worker import (_model_entry,
@@ -1719,6 +1998,15 @@ class OcrManager:
         # aside; repeating it per document would bury it.
         standing = _pod_pipeline_refusal(model_id)
         if standing:
+            # Said when it was set aside, but a cooldown is minutes long and a
+            # run is hours: without a word here the documents in between look
+            # exactly like documents the pods read.
+            _say_rate_limited(
+                ("standing-refusal", model_id),
+                f"[ocr] still not asking the pods for {model_id!r} to run the "
+                f"pipeline ({standing}) — THIS BOX rasterises until the "
+                f"{POD_PIPELINE_COOLDOWN_S:.0f}s cooldown expires",
+                every=POD_PIPELINE_COOLDOWN_S)
             return None
         parts = (list(iter_pdf_page_pdfs(data)) if is_pdf else [data])
         if not parts:
@@ -1726,6 +2014,8 @@ class OcrManager:
         sem = self._sem or asyncio.Semaphore(4)
         opts = dict(page_opts or {})
         refused: list = []
+        unreachable: list = []
+        policy, wait_s = _pod_pipeline_policy(cfg)
 
         async def _one(idx: int, blob: bytes):
             async with sem:
@@ -1733,34 +2023,61 @@ class OcrManager:
                     # Another page already found the pods cannot run it. Do not
                     # send the rest of the document one failure at a time.
                     return None
-                try:
-                    lease = placement.choose(
-                        model_id, spill_after=placement.spill_threshold(model_id))
-                except Exception as exc:
-                    raise OcrError(f"no upstream for {model_id}: {exc}", status=503)
-                try:
-                    if lease.kind == "local":
-                        # Nothing rented: the local pipeline is the right answer
-                        # and the caller already has it.
+                # TRY NOT TO GO LOCAL. A pool with no free pod right now is
+                # usually a pool whose pods are booting; this box rasterising
+                # instead is how a GPU-less orchestrator reached 38 GB and
+                # climbing while sixteen A40s it was paying for sat idle.
+                for attempt in range(1, POD_PAGE_ATTEMPTS + 1):
+                    if refused:
                         return None
-                    return await asyncio.to_thread(
-                        _post_page_to_pod, lease.base, lease.api_key, blob,
-                        f"{idx}.pdf" if is_pdf else (filename or "page"),
-                        dpi, opts, detect, engine_name)
-                except OcrError as exc:
-                    # A POD THAT CANNOT RUN THE PIPELINE IS NOT A FAILED PAGE.
-                    # It answers the same way for every page of every document,
-                    # so raising turned "this optimisation is unavailable" into
-                    # "this document cannot be read" -- while the local pipeline
-                    # sat there able to read it. The comment on this method
-                    # already said an optimisation that cannot run must cost
-                    # latency and never an error; this is that, implemented.
-                    if int(getattr(exc, "status", 0) or 0) not in _POD_CANNOT_RUN_PIPELINE:
-                        raise
-                    refused.append(str(exc))
-                    return None
-                finally:
-                    lease.release()
+                    lease = await _lease_a_pod(model_id, policy, wait_s)
+                    if lease is None:
+                        return None
+                    try:
+                        return await asyncio.to_thread(
+                            _post_page_to_pod, lease.base, lease.api_key, blob,
+                            f"{idx}.pdf" if is_pdf else (filename or "page"),
+                            dpi, opts, detect, engine_name)
+                    except PodUnreachable as exc:
+                        # A DEAD SOCKET IS NOT A VERDICT ON THE IMAGE. Ask the
+                        # address whether anything is still there: a pod that
+                        # answers /healthz keeps its place in the pool (killing
+                        # it on a reused idle connection has cost seven healthy
+                        # cards before), and one that does not is reported so
+                        # the next attempt is routed elsewhere.
+                        alive = await asyncio.to_thread(
+                            _pod_still_alive, lease.base, lease.api_key)
+                        if not alive:
+                            lease.dead()
+                        _say_rate_limited(
+                            ("unreachable", model_id),
+                            f"[ocr] a pod for {model_id!r} did not answer this "
+                            f"page ({exc}); it is "
+                            + ("GONE and has been dropped from the pool"
+                               if not alive else
+                               "still answering /healthz, so this was the "
+                               "connection and not the pod")
+                            + f" — attempt {attempt} of {POD_PAGE_ATTEMPTS}. "
+                            f"This does NOT set the pool aside: only a pod that "
+                            f"ANSWERS can say it cannot run the pipeline.",
+                            every=30.0)
+                        if attempt >= POD_PAGE_ATTEMPTS:
+                            unreachable.append(str(exc))
+                            return None
+                        continue
+                    except OcrError as exc:
+                        # A POD THAT CANNOT RUN THE PIPELINE IS NOT A FAILED
+                        # PAGE. It answers the same way for every page of every
+                        # document, so raising turned "this optimisation is
+                        # unavailable" into "this document cannot be read" --
+                        # while the local pipeline sat there able to read it.
+                        if int(getattr(exc, "status", 0) or 0) not in _POD_CANNOT_RUN_PIPELINE:
+                            raise
+                        refused.append(str(exc))
+                        return None
+                    finally:
+                        lease.release()
+                return None
 
         got = await asyncio.gather(*[_one(i, b) for i, b in enumerate(parts)],
                                    return_exceptions=True)
@@ -1776,7 +2093,16 @@ class OcrManager:
             # had never run. One line is the difference between a test that
             # proves something and a test that proves nothing.
             _n = sum(1 for r in got if r is None)
-            if refused:
+            if unreachable and not refused:
+                _say_rate_limited(
+                    ("unreachable-fallback", model_id),
+                    f"[ocr] {_n} of {len(parts)} page(s) could not be sent to "
+                    f"any pod for {model_id!r} after {POD_PAGE_ATTEMPTS} "
+                    f"attempts each ({unreachable[0]}) — reading the WHOLE "
+                    f"document on THIS BOX, which rasterises here. The pool is "
+                    f"NOT set aside: the next document asks the pods again.",
+                    every=30.0)
+            elif refused:
                 # The pods answered, and what they said was "not me". That is a
                 # different fault from "nothing is rented", needs a different
                 # fix, and must carry the pod's OWN words: this is the only

tests/test_ocr_dual_mode.py

@@ -305,16 +305,24 @@ def _fake_placement(monkeypatch, kind="pod", on_choose=None):
 def test_no_pod_means_the_local_pipeline(monkeypatch):
     """`placement.choose` answering "local" means nothing is rented. The whole
     document is then read locally -- not half one way and half the other, which
-    is the hardest kind of result to explain."""
+    is the hardest kind of result to explain.
+
+    With `ocr.pod_pipeline = auto`, which is this immediate fallback: the
+    shipped default waits for a pod first (see
+    tests/test_ocr_pod_pipeline_policy.py), because reading here means
+    rasterising here."""
+    class _Auto(_Cfg):
+        pod_pipeline = "auto"
+
     mgr = om.OcrManager()
-    mgr._require_cfg = lambda: _Cfg()
+    mgr._require_cfg = lambda: _Auto()
     mgr._sem = asyncio.Semaphore(4)
 
     _fake_placement(monkeypatch, kind="local")
 
     out = asyncio.run(mgr._read_on_pods(
         _pdf(3), "d.pdf", "application/pdf", "m",
-        dpi=250, page_opts=None, detect=None, cfg=_Cfg()))
+        dpi=250, page_opts=None, detect=None, cfg=_Auto()))
     assert out is None
 
 

tests/test_ocr_pod_pipeline_policy.py

+# CoderAI - OpenAI-compatible API server
+# Copyright (C) 2026 Stefy Lanza <stefy@nexlab.net>
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+
+"""Going local must be a DECISION, and it must be audible.
+
+Two defects, found together on a production run that had already been fixed
+three times:
+
+1. The gate compared the request's engine against the three VLM engine names.
+   `/v1/ocr` has always also accepted the MODEL a caller knows about in place
+   of the engine driving it -- `resolve_engine` maps the Surya-2 checkpoint to
+   'surya', and the catalogue advertises those ids as aliases -- so a caller
+   naming the model got no pod-side pipeline at all.
+
+2. Every way of ending up on the local pipeline that was not "a pod said no"
+   returned "" or None in SILENCE. The log of the run that prompted this held
+   8,278 successful pod reads and not one line explaining why the next four
+   thousand documents were rasterised on a GPU-less orchestrator that climbed
+   0.8 GB/min to 36 GB while sixteen rented A40s idled.
+
+So: resolve the engine the way a read resolves it, say which way the decision
+went every time it is not the pod, and prefer waiting for a pod over taking
+the work on.
+"""
+
+import asyncio
+import types
+
+import pytest
+
+import codai.ocr.manager as om
+
+
+@pytest.fixture
+def anyio_backend():
+    return "asyncio"
+
+
+@pytest.fixture(autouse=True)
+def _forget_everything_said():
+    om._PIPELINE_GATE_SAID.clear()
+    om._SAID_AT.clear()
+    om._POD_PIPELINE_UNUSABLE.clear()
+    om._POD_WAIT_COOLDOWN.clear()
+    yield
+    om._PIPELINE_GATE_SAID.clear()
+    om._SAID_AT.clear()
+    om._POD_PIPELINE_UNUSABLE.clear()
+    om._POD_WAIT_COOLDOWN.clear()
+
+
+OCR_IMAGE = "ghcr.io/nextime/coderai-vllm-ocr:latest"
+
+
+def _cfg(**kw):
+    base = {"default_engine": "surya", "surya_serve": "model",
+            "surya_model_id": "datalab-to/surya-ocr-2",
+            "chandra_serve": "model",
+            "chandra_model_id": "datalab-to/chandra-ocr-2",
+            "olmocr_serve": "model",
+            "olmocr_model_id": "allenai/olmOCR-2-7B-1025-FP8",
+            "max_concurrency": 8, "page_window": 4,
+            "pod_pipeline": "prefer", "pod_pipeline_wait_s": 60.0}
+    base.update(kw)
+    return types.SimpleNamespace(**base)
+
+
+def _ocr_image_entry(monkeypatch):
+    from codai.api import runpod_worker as rw
+    monkeypatch.setattr(rw, "_model_entry", lambda _mid: {
+        "id": "m", "backend": "runpod",
+        "runpod": {"mode": "pods", "image": OCR_IMAGE}})
+
+
+# ---------------------------------------------------------------------------
+# 1. The engine name resolves the way a READ resolves it
+# ---------------------------------------------------------------------------
+
+def test_the_model_id_names_the_engine_just_like_everywhere_else(monkeypatch):
+    """THE defect: documented, accepted by /v1/ocr, and fatal only here."""
+    _ocr_image_entry(monkeypatch)
+    got = om.pod_pipeline_model(_cfg(), "datalab-to/surya-ocr-2")
+    assert got == "datalab-to/surya-ocr-2"
+
+
+def test_the_model_id_is_matched_case_insensitively(monkeypatch):
+    _ocr_image_entry(monkeypatch)
+    assert om.pod_pipeline_model(_cfg(), "DataLab-To/Surya-OCR-2")
+
+
+def test_every_vlm_engine_can_be_named_by_its_model(monkeypatch):
+    _ocr_image_entry(monkeypatch)
+    cfg = _cfg()
+    assert om.pod_pipeline_model(cfg, "datalab-to/chandra-ocr-2") == \
+        "datalab-to/chandra-ocr-2"
+    assert om.pod_pipeline_model(cfg, "allenai/olmOCR-2-7B-1025-FP8") == \
+        "allenai/olmOCR-2-7B-1025-FP8"
+
+
+def test_resolving_an_alias_is_said_out_loud(monkeypatch, capsys):
+    """Proving the fix in a live log needs the alias to be visible."""
+    _ocr_image_entry(monkeypatch)
+    om.pod_pipeline_model(_cfg(), "datalab-to/surya-ocr-2")
+    out = capsys.readouterr().out
+    assert "names the 'surya' engine" in out
+
+
+def test_the_engine_name_itself_still_works(monkeypatch):
+    _ocr_image_entry(monkeypatch)
+    assert om.pod_pipeline_model(_cfg(), "surya") == "datalab-to/surya-ocr-2"
+
+
+# ---------------------------------------------------------------------------
+# 2. Every silent route now says which it was
+# ---------------------------------------------------------------------------
+
+def test_an_engine_with_no_pod_pipeline_says_so(monkeypatch, capsys):
+    _ocr_image_entry(monkeypatch)
+    assert om.pod_pipeline_model(_cfg(), "doctr") == ""
+    out = capsys.readouterr().out
+    assert "no pod-side pipeline" in out
+    assert "THIS BOX rasterises" in out
+
+
+def test_an_unknown_engine_name_says_so(monkeypatch, capsys):
+    """The shape a model id used to take: unrecognised, silent, local."""
+    _ocr_image_entry(monkeypatch)
+    assert om.pod_pipeline_model(_cfg(), "some-model-nobody-serves") == ""
+    out = capsys.readouterr().out
+    assert "not an OCR engine or a model id" in out
+    assert "some-model-nobody-serves" in out
+
+
+def test_a_serve_mode_that_is_not_model_says_so_and_says_the_mode(
+        monkeypatch, capsys):
+    _ocr_image_entry(monkeypatch)
+    assert om.pod_pipeline_model(_cfg(surya_serve="local"), "surya") == ""
+    out = capsys.readouterr().out
+    assert "serve mode is 'local'" in out
+    assert "surya_serve" in out
+
+
+def test_an_empty_model_id_says_so(monkeypatch, capsys):
+    _ocr_image_entry(monkeypatch)
+    assert om.pod_pipeline_model(_cfg(surya_model_id=""), "surya") == ""
+    out = capsys.readouterr().out
+    assert "surya_model_id is " in out
+
+
+def test_the_reasons_are_rate_limited_not_repeated_per_document(
+        monkeypatch, capsys):
+    _ocr_image_entry(monkeypatch)
+    for _ in range(50):
+        om.pod_pipeline_model(_cfg(surya_serve="local"), "surya")
+    assert capsys.readouterr().out.count("serve mode is") == 1
+
+
+def test_a_standing_refusal_keeps_saying_this_box_rasterises(monkeypatch, capsys):
+    """A cooldown is minutes; a run is hours. Silence in between looked like
+    success -- the documents read here are indistinguishable from the ones the
+    pods read unless something says so."""
+    om._POD_PIPELINE_UNUSABLE["m"] = (9e18, "the pod answered 503: no model")
+    mgr = om.OcrManager()
+    out = asyncio.run(mgr._read_on_pods(
+        b"%PDF-1.4 x", "d.pdf", "application/pdf", "m", dpi=200,
+        page_opts=None, detect=None, cfg=_cfg(), engine_name="surya"))
+    assert out is None
+    said = capsys.readouterr().out
+    assert "still not asking the pods" in said
+    assert "no model" in said                      # the pod's own words
+
+
+# ---------------------------------------------------------------------------
+# 3. Try not to go local
+# ---------------------------------------------------------------------------
+
+class _LocalLease:
+    kind = "local"
+    base = "http://127.0.0.1:8000/v1"
+    api_key = ""
+
+    def release(self):
+        pass
+
+
+class _PodLease:
+    kind = "pod"
+    base = "https://10.0.0.1:1234/v1"
+    api_key = "k"
+
+    def __init__(self):
+        self.released = False
+
+    def release(self):
+        self.released = True
+
+
+def _placement(monkeypatch, leases):
+    """`choose` hands back each lease in turn, repeating the last for ever."""
+    from codai.models import placement
+    seen = []
+
+    def _choose(*_a, **_k):
+        got = leases[min(len(seen), len(leases) - 1)]
+        seen.append(got)
+        return got() if callable(got) else got
+
+    monkeypatch.setattr(placement, "choose", _choose)
+    monkeypatch.setattr(placement, "spill_threshold", lambda *_a, **_k: 0)
+    monkeypatch.setattr(placement, "why_no_candidates",
+                        lambda *_a, **_k: "the pool is at its ceiling")
+    return seen
+
+
+def _one_page(monkeypatch):
+    monkeypatch.setattr(om, "iter_pdf_page_pdfs", lambda _d: [b"%PDF-page0"])
+
+
+def _page_read(monkeypatch):
+    monkeypatch.setattr(om, "_post_page_to_pod",
+                        lambda *_a, **_k: {"engine": "surya",
+                                           "pages": [{"text": "ok"}]})
+
+
+async def _read(cfg):
+    mgr = om.OcrManager()
+    return await mgr._read_on_pods(b"%PDF-1.4 x", "d.pdf", "application/pdf",
+                                   "m", dpi=200, page_opts=None, detect=None,
+                                   cfg=cfg, engine_name="surya")
+
+
+@pytest.mark.anyio
+async def test_prefer_waits_and_takes_the_pod_that_turns_up(monkeypatch, capsys):
+    """A pool with nothing free is usually a pool whose pods are booting."""
+    _one_page(monkeypatch)
+    _page_read(monkeypatch)
+    _placement(monkeypatch, [_LocalLease(), _PodLease()])
+    got = await _read(_cfg(pod_pipeline="prefer", pod_pipeline_wait_s=5.0))
+    assert got is not None, "waited, a pod appeared, and it still went local"
+    assert "became free after" in capsys.readouterr().out
+
+
+@pytest.mark.anyio
+async def test_prefer_falls_back_when_the_wait_runs_out_and_says_so(
+        monkeypatch, capsys):
+    """Bounded: an optimisation that cannot run must cost latency, not pages."""
+    _one_page(monkeypatch)
+    _page_read(monkeypatch)
+    _placement(monkeypatch, [_LocalLease()])
+    got = await _read(_cfg(pod_pipeline="prefer", pod_pipeline_wait_s=0.2))
+    assert got is None
+    out = capsys.readouterr().out
+    assert "waited" in out and "THIS BOX" in out
+    assert "memory" in out
+
+
+@pytest.mark.anyio
+async def test_a_timed_out_wait_is_not_paid_again_by_the_next_page(monkeypatch):
+    """60s per page is a tax, not a ceiling: once a wait fails, pages go
+    straight through for a cooldown."""
+    _one_page(monkeypatch)
+    _page_read(monkeypatch)
+    _placement(monkeypatch, [_LocalLease()])
+    cfg = _cfg(pod_pipeline="prefer", pod_pipeline_wait_s=0.3)
+    await _read(cfg)
+    loop = asyncio.get_running_loop()
+    started = loop.time()
+    await _read(cfg)
+    assert loop.time() - started < 0.2, "the second document waited again"
+
+
+@pytest.mark.anyio
+async def test_require_refuses_the_page_rather_than_rasterising_here(
+        monkeypatch):
+    _one_page(monkeypatch)
+    _page_read(monkeypatch)
+    _placement(monkeypatch, [_LocalLease()])
+    with pytest.raises(om.OcrError) as exc:
+        await _read(_cfg(pod_pipeline="require", pod_pipeline_wait_s=0.1))
+    assert exc.value.status == 503
+    assert "will not" in str(exc.value) and "rasterise" in str(exc.value)
+    assert "ceiling" in str(exc.value)             # WHY, not just "no"
+
+
+@pytest.mark.anyio
+async def test_auto_keeps_the_old_immediate_fallback(monkeypatch):
+    _one_page(monkeypatch)
+    _page_read(monkeypatch)
+    _placement(monkeypatch, [_LocalLease()])
+    loop = asyncio.get_running_loop()
+    started = loop.time()
+    assert await _read(_cfg(pod_pipeline="auto", pod_pipeline_wait_s=30)) is None
+    assert loop.time() - started < 0.2
+
+
+@pytest.mark.anyio
+async def test_auto_still_says_which_way_it_went(monkeypatch, capsys):
+    _one_page(monkeypatch)
+    _page_read(monkeypatch)
+    _placement(monkeypatch, [_LocalLease()])
+    await _read(_cfg(pod_pipeline="auto"))
+    out = capsys.readouterr().out
+    assert "placement answered 'local'" in out
+    assert "at its ceiling" in out                 # the REASON, from placement
+
+
+@pytest.mark.anyio
+async def test_a_pod_on_the_first_ask_never_waits(monkeypatch):
+    _one_page(monkeypatch)
+    _page_read(monkeypatch)
+    _placement(monkeypatch, [_PodLease()])
+    loop = asyncio.get_running_loop()
+    started = loop.time()
+    assert await _read(_cfg()) is not None
+    assert loop.time() - started < 0.2
+
+
+@pytest.mark.anyio
+async def test_the_lease_is_always_returned(monkeypatch):
+    """A lease leaked on the local path shrinks the pool for good."""
+    _one_page(monkeypatch)
+    _page_read(monkeypatch)
+    lease = _PodLease()
+    _placement(monkeypatch, [lease])
+    await _read(_cfg())
+    assert lease.released
+
+
+def test_an_unknown_policy_falls_back_to_the_safe_one():
+    """A typo in a config file must not turn the memory guard off."""
+    assert om._pod_pipeline_policy(_cfg(pod_pipeline="wait-forever"))[0] == "prefer"
+    assert om._pod_pipeline_policy(types.SimpleNamespace())[0] == "prefer"
+    # The words an operator might reasonably write for "don't wait".
+    assert om._pod_pipeline_policy(_cfg(pod_pipeline="off"))[0] == "auto"
+
+
+def test_the_shipped_default_prefers_a_pod():
+    from codai.config import OcrConfig
+    assert OcrConfig().pod_pipeline == "prefer"
+    assert OcrConfig().pod_pipeline_wait_s > 0
+
+
+# ---------------------------------------------------------------------------
+# 4. A dead socket is not a verdict on the image
+#
+# THE defect this file was written for. Twelve times in four hours a pooled
+# keep-alive connection a pod had closed while idle failed on its next use:
+#
+#   pod OCR at https://194.68.245.122:22005/v1/ocr is unreachable:
+#   ('Connection aborted.', RemoteDisconnected('Remote end closed connection
+#   without response'))
+#
+# requests raised, `_post_page_to_pod` turned it into OcrError(503), 503 is in
+# `_POD_CANNOT_RUN_PIPELINE`, and so one dead TCP connection said "these pods
+# cannot run the OCR pipeline" and moved every document onto the orchestrator
+# for five minutes at a time.
+# ---------------------------------------------------------------------------
+
+class _Flaky:
+    """A pod transport that dies the first n times, then answers."""
+
+    def __init__(self, deaths, exc=None):
+        self.deaths = deaths
+        self.calls = 0
+        self.exc = exc or Exception(
+            "('Connection aborted.', RemoteDisconnected('Remote end closed "
+            "connection without response'))")
+
+    def __call__(self, *_a, **_k):
+        self.calls += 1
+        if self.calls <= self.deaths:
+            raise self.exc
+        return types.SimpleNamespace(
+            status_code=200, text="",
+            json=lambda: {"engine": "surya", "pages": [{"text": "ok"}]})
+
+
+def _transport(monkeypatch, fn):
+    from codai.api import pod_http
+    monkeypatch.setattr(pod_http, "post", fn)
+    monkeypatch.setattr(pod_http, "get", lambda *_a, **_k:
+                        types.SimpleNamespace(status_code=200))
+
+
+def test_a_dead_pooled_socket_is_retried_on_a_fresh_one(monkeypatch):
+    """Nothing was sent, so a second attempt costs a connect and no work."""
+    flaky = _Flaky(deaths=1)
+    _transport(monkeypatch, flaky)
+    got = om._post_page_to_pod("https://10.0.0.1:1234/v1", "k", b"%PDF", "0.pdf",
+                               200, {}, None, "surya")
+    assert got["engine"] == "surya"
+    assert flaky.calls == 2
+
+
+def test_a_timeout_is_never_retried(monkeypatch):
+    """The pod may be ninety seconds into a 130-second read; sending the page
+    again would double the work on a card already doing it."""
+    flaky = _Flaky(deaths=99, exc=Exception("HTTPSConnectionPool: Read timed out"))
+    _transport(monkeypatch, flaky)
+    with pytest.raises(om.PodUnreachable):
+        om._post_page_to_pod("https://10.0.0.1:1234/v1", "k", b"%PDF", "0.pdf",
+                             200, {}, None, "surya")
+    assert flaky.calls == 1
+
+
+def test_an_unreachable_pod_raises_the_transport_error_not_an_answer(monkeypatch):
+    _transport(monkeypatch, _Flaky(deaths=99))
+    with pytest.raises(om.PodUnreachable) as exc:
+        om._post_page_to_pod("https://10.0.0.1:1234/v1", "k", b"%PDF", "0.pdf",
+                             200, {}, None, "surya")
+    # Still a 503 to anything that only reads the status -- but a DIFFERENT
+    # type, which is what stops it being read as "this image cannot do OCR".
+    assert exc.value.status == 503
+    assert isinstance(exc.value, om.OcrError)
+
+
+def test_a_pod_that_answers_503_is_still_a_refusal(monkeypatch):
+    """The other side of the line: an ANSWER may condemn the image."""
+    _transport(monkeypatch, lambda *_a, **_k: types.SimpleNamespace(
+        status_code=503, text="model is not served over HTTP by this install",
+        json=lambda: {}))
+    with pytest.raises(om.OcrError) as exc:
+        om._post_page_to_pod("https://10.0.0.1:1234/v1", "k", b"%PDF", "0.pdf",
+                             200, {}, None, "surya")
+    assert not isinstance(exc.value, om.PodUnreachable)
+    assert exc.value.status == 503
+
+
+@pytest.mark.anyio
+async def test_an_unreachable_pod_does_not_set_the_whole_pool_aside(monkeypatch):
+    """THE five-minute outage, twelve times over: one dead socket must not take
+    the pod-side pipeline away from every document that follows."""
+    _one_page(monkeypatch)
+    _placement(monkeypatch, [_PodLease()])
+    monkeypatch.setattr(om, "_post_page_to_pod", lambda *_a, **_k: (
+        _ for _ in ()).throw(om.PodUnreachable("unreachable: aborted", status=503)))
+    monkeypatch.setattr(om, "_pod_still_alive", lambda *_a, **_k: True)
+    assert await _read(_cfg(pod_pipeline="auto")) is None
+    assert not om._POD_PIPELINE_UNUSABLE, \
+        "a dead socket set the pool aside, so the next document goes local too"
+
+
+@pytest.mark.anyio
+async def test_a_page_whose_socket_died_is_tried_on_another_pod(monkeypatch):
+    _one_page(monkeypatch)
+    _placement(monkeypatch, [_PodLease()])
+    seen = []
+
+    def _post(base, *_a, **_k):
+        seen.append(base)
+        if len(seen) < 2:
+            raise om.PodUnreachable("unreachable: aborted", status=503)
+        return {"engine": "surya", "pages": [{"text": "ok"}]}
+
+    monkeypatch.setattr(om, "_post_page_to_pod", _post)
+    monkeypatch.setattr(om, "_pod_still_alive", lambda *_a, **_k: True)
+    got = await _read(_cfg(pod_pipeline="auto"))
+    assert got is not None, "gave the document to the local pipeline instead"
+    assert len(seen) == 2
+
+
+@pytest.mark.anyio
+async def test_a_pod_answering_healthz_is_not_dropped_from_the_pool(monkeypatch):
+    """Killing a pod on a reused idle connection cost seven healthy A40s once."""
+    _one_page(monkeypatch)
+    lease = _PodLease()
+    lease.declared_dead = False
+
+    def _dead():
+        lease.declared_dead = True
+    lease.dead = _dead
+    _placement(monkeypatch, [lease])
+    monkeypatch.setattr(om, "_post_page_to_pod", lambda *_a, **_k: (
+        _ for _ in ()).throw(om.PodUnreachable("unreachable: aborted", status=503)))
+    monkeypatch.setattr(om, "_pod_still_alive", lambda *_a, **_k: True)
+    await _read(_cfg(pod_pipeline="auto"))
+    assert not lease.declared_dead
+
+
+@pytest.mark.anyio
+async def test_a_pod_that_answers_nothing_at_all_is_dropped(monkeypatch):
+    """The opposite case, and the reason the probe exists rather than a guess."""
+    _one_page(monkeypatch)
+    lease = _PodLease()
+    lease.declared_dead = False
+
+    def _dead():
+        lease.declared_dead = True
+    lease.dead = _dead
+    _placement(monkeypatch, [lease])
+    monkeypatch.setattr(om, "_post_page_to_pod", lambda *_a, **_k: (
+        _ for _ in ()).throw(om.PodUnreachable("unreachable: aborted", status=503)))
+    monkeypatch.setattr(om, "_pod_still_alive", lambda *_a, **_k: False)
+    await _read(_cfg(pod_pipeline="auto"))
+    assert lease.declared_dead
+
+
+@pytest.mark.anyio
+async def test_falling_back_on_unreachable_pods_says_the_pool_is_not_set_aside(
+        monkeypatch, capsys):
+    _one_page(monkeypatch)
+    _placement(monkeypatch, [_PodLease()])
+    monkeypatch.setattr(om, "_post_page_to_pod", lambda *_a, **_k: (
+        _ for _ in ()).throw(om.PodUnreachable(
+            "pod OCR at https://10.0.0.1:1234/v1/ocr is unreachable: "
+            "('Connection aborted.', RemoteDisconnected(...))", status=503)))
+    monkeypatch.setattr(om, "_pod_still_alive", lambda *_a, **_k: True)
+    await _read(_cfg(pod_pipeline="auto"))
+    out = capsys.readouterr().out
+    assert "could not be sent to any pod" in out
+    assert "NOT set aside" in out
+    assert "Connection aborted" in out             # the transport's own words
+
+
+def test_an_active_cooldown_is_never_extended_silently():
+    """09:10 to 09:26 with nothing in the log: refusals during a cooldown kept
+    pushing the expiry out while only the first one printed."""
+    assert om._set_aside_pod_pipeline("m", "first") is True
+    first = om._POD_PIPELINE_UNUSABLE["m"][0]
+    assert om._set_aside_pod_pipeline("m", "second") is False
+    assert om._POD_PIPELINE_UNUSABLE["m"][0] == first
+    assert om._POD_PIPELINE_UNUSABLE["m"][1] == "first"
+
+
+def test_the_hint_list_recognises_what_production_actually_logged():
+    msg = ("('Connection aborted.', RemoteDisconnected('Remote end closed "
+           "connection without response'))")
+    assert om._looks_like_a_dead_socket(Exception(msg))
+    assert om._looks_like_a_dead_socket(ConnectionResetError("Connection reset by peer"))
+    assert not om._looks_like_a_dead_socket(Exception("Read timed out. (read timeout=600)"))
+    assert not om._looks_like_a_dead_socket(Exception("that is not a PDF"))