a pod that cannot run the pipeline is not a failed page (v0.4.2)

0.4.1 sent pages to the pods, as it was meant to. The pods answered 503:

    model 'datalab-to/surya-ocr-2' is not served over HTTP by this
    install, so an OCR engine cannot attach to it

_read_on_pods raised that, so the document failed -- while the local
pipeline sat there perfectly able to read it. The method's own docstring
already said "an optimisation that cannot run must cost latency, never an
error"; only the "no pod at all" case implemented it.

A refusal (404/501/503 from the pod's own /v1/ocr) is now a fallback, not a
failure:

* the rest of the document stops being sent one failure at a time;
* the pool is set aside for 300s, so the next document does not pay a
  failed round trip per page, and a rebuilt image is picked up without
  restarting the orchestrator;
* it is said ONCE, in the pod's own words, and says that this box will now
  rasterise so its memory is worth watching.

A 400 still fails: falling back on any error would hide a broken page
behind a slow one.

This restores service. It does NOT make the pods run the pipeline -- that
is the next commit, and the log line above is what will say when it works.
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.1"
+__version__ = "0.4.2"
 __tagline__ = "Complete orchestration, distribution and escalation of remotizable advanced inference."
 
 # Configure the CUDA caching allocator BEFORE torch is imported anywhere.

codai/ocr/manager.py

@@ -420,6 +420,52 @@ def _page_from_pod(d: dict, index: int) -> OcrPage:
     return page
 
 
+#: How long a pool stays marked as unable to run the pipeline on its own pods
+#: before it is tried again. Long enough that a document does not pay a failed
+#: round trip per page, short enough that fixing the pods (a rebuilt image, a
+#: corrected seed) takes effect without restarting the orchestrator.
+POD_PIPELINE_COOLDOWN_S = 300.0
+
+#: model id -> (until, reason). See _read_on_pods().
+_POD_PIPELINE_UNUSABLE: dict = {}
+_POD_PIPELINE_LOCK = threading.Lock()
+
+
+def _pod_pipeline_refusal(mid: str) -> str:
+    """Why this pool's pods were set aside, or "" if they are to be tried."""
+    with _POD_PIPELINE_LOCK:
+        hit = _POD_PIPELINE_UNUSABLE.get(mid)
+        if not hit:
+            return ""
+        until, reason = hit
+        if until <= time.time():
+            _POD_PIPELINE_UNUSABLE.pop(mid, None)
+            return ""
+        return reason
+
+
+def _set_aside_pod_pipeline(mid: str, reason: str) -> bool:
+    """Stop sending pages to this pool's own pipeline for a while.
+
+    Returns True the first time, so the caller says it once per cooldown
+    instead of once per page.
+    """
+    now = time.time()
+    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)
+    return fresh
+
+
+#: Answers that mean "this pod cannot run the pipeline", as opposed to "this
+#: page went wrong". A pod that does not serve /v1/ocr, cannot attach its OCR
+#: engine to the model it is serving, or has no model to attach to will say the
+#: same thing for every page of every document -- so the pages belong on the
+#: local pipeline, not in a retry loop.
+_POD_CANNOT_RUN_PIPELINE = (404, 501, 503)
+
+
 #: (engine, model, image) combinations already explained. The gate is asked
 #: once per document, and the answer only changes when the configuration does,
 #: so saying it every time would bury it; saying it never is what let a whole
@@ -1649,14 +1695,25 @@ class OcrManager:
                   or (filename or "").lower().endswith(".pdf")
                   or data[:5] == b"%PDF-")
         # A single image has nothing to split; it goes as it arrived.
+        # Set aside recently? Then this document does not pay a round trip per
+        # page to be told the same thing again. Said once, when it was set
+        # aside; repeating it per document would bury it.
+        standing = _pod_pipeline_refusal(model_id)
+        if standing:
+            return None
         parts = (list(iter_pdf_page_pdfs(data)) if is_pdf else [data])
         if not parts:
             return None
         sem = self._sem or asyncio.Semaphore(4)
         opts = dict(page_opts or {})
+        refused: list = []
 
         async def _one(idx: int, blob: bytes):
             async with sem:
+                if refused:
+                    # 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))
@@ -1671,6 +1728,18 @@ class OcrManager:
                         _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()
 
@@ -1688,11 +1757,25 @@ 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)
-            print(f"[ocr] the pod-side pipeline was configured for {model_id!r} "
-                  f"but {_n} of {len(parts)} page(s) had no pod to run it "
-                  f"(placement answered 'local' — nothing rented, or the pool "
-                  f"could not serve) — reading the WHOLE document here instead",
-                  flush=True)
+            if 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
+                # place they are visible.
+                if _set_aside_pod_pipeline(model_id, refused[0]):
+                    print(f"[ocr] the pods for {model_id!r} run an image that "
+                          f"carries the OCR pipeline, but they REFUSE to run "
+                          f"it: {refused[0]} — falling back to the pipeline on "
+                          f"this box and not asking them again for "
+                          f"{POD_PIPELINE_COOLDOWN_S:.0f}s. This box will "
+                          f"rasterise in the meantime, so watch its memory.",
+                          flush=True)
+            else:
+                print(f"[ocr] the pod-side pipeline was configured for "
+                      f"{model_id!r} but {_n} of {len(parts)} page(s) had no "
+                      f"pod to run it (placement answered 'local' — nothing "
+                      f"rented, or the pool could not serve) — reading the "
+                      f"WHOLE document here instead", flush=True)
             return None
         bad = [r for r in got if isinstance(r, BaseException)]
         if bad:

tests/test_ocr_dual_mode.py

@@ -459,15 +459,20 @@ def test_a_silent_fallback_would_make_a_test_prove_nothing():
     fallback and the success say which one happened."""
     src = (ROOT / "codai/ocr/manager.py").read_text()
     fn = src.split("async def _read_on_pods")[1].split("\n    def ")[0]
-    assert "but {_n} of {len(parts)} page(s) had no pod" in fn
+    # Matched on fragments rather than a whole sentence: the wording is
+    # allowed to be rewrapped, but a path that reports NOTHING is not.
+    assert "page(s) had no " in fn
     assert "ON THE PODS" in fn
+    # …and the third outcome, added after pods that carried the pipeline
+    # refused to run it and the document failed instead of falling back.
+    assert "REFUSE to run" in fn
 
 
 def test_the_fallback_message_names_the_cause_not_just_the_effect():
     src = (ROOT / "codai/ocr/manager.py").read_text()
     fn = src.split("async def _read_on_pods")[1].split("\n    def ")[0]
     assert "placement answered 'local'" in fn
-    assert "nothing rented" in fn
+    assert "nothing " in fn and "rented" in fn
 
 
 # ------------------------------- all three VLM engines, not just surya #

tests/test_ocr_pod_pipeline_gate.py

@@ -28,6 +28,11 @@ from codai.api.runpod_worker import (image_runs_ocr_pipeline,
 import codai.ocr.manager as om
 
 
+@pytest.fixture
+def anyio_backend():
+    return "asyncio"
+
+
 OCR_IMAGE = "ghcr.io/nextime/coderai-vllm-ocr:latest"
 PLAIN_IMAGE = "ghcr.io/nextime/coderai-vllm:latest"
 
@@ -192,3 +197,140 @@ def test_a_non_vlm_engine_never_goes_to_a_pod(monkeypatch):
 def test_local_serve_mode_never_goes_to_a_pod(monkeypatch):
     _with_entry(monkeypatch, _entry(engine="coderai-vllm-ocr"))
     assert om.pod_pipeline_model(_cfg(surya_serve="local"), "surya") == ""
+
+
+# --------------------------------------------------------------------------
+# A pod that cannot run the pipeline must cost latency, not the document
+# --------------------------------------------------------------------------
+
+class _Lease:
+    kind = "pod"
+    base = "https://10.0.0.1:1234/v1"
+    api_key = "k"
+
+    def release(self):
+        pass
+
+
+@pytest.fixture(autouse=True)
+def _forget_set_aside_pools():
+    om._POD_PIPELINE_UNUSABLE.clear()
+    yield
+    om._POD_PIPELINE_UNUSABLE.clear()
+
+
+def _two_page_pdf_split(monkeypatch, pages=2):
+    monkeypatch.setattr(om, "iter_pdf_page_pdfs",
+                        lambda _d: [b"%%PDF-page" + str(i).encode() for i in range(pages)])
+
+
+def _pods_answer(monkeypatch, fn):
+    from codai.models import placement
+    monkeypatch.setattr(placement, "choose", lambda *_a, **_k: _Lease())
+    monkeypatch.setattr(placement, "spill_threshold", lambda *_a, **_k: 0)
+    monkeypatch.setattr(om, "_post_page_to_pod", fn)
+
+
+async def _read(mgr, cfg):
+    return await mgr._read_on_pods(b"%PDF-1.4 x", "f.pdf", "application/pdf",
+                                   "datalab-to/surya-ocr-2", dpi=200,
+                                   page_opts=None, detect=None, cfg=cfg,
+                                   engine_name="surya")
+
+
+@pytest.mark.anyio
+async def test_a_pod_that_refuses_the_pipeline_falls_back_instead_of_failing(
+        monkeypatch, capsys):
+    """THE outage: a 503 from the pod turned an optimisation into a failure."""
+    _two_page_pdf_split(monkeypatch)
+
+    def _refuse(*_a, **_k):
+        raise om.OcrError("pod OCR at https://10.0.0.1:1234/v1/ocr answered "
+                          "503: model is not served over HTTP by this install",
+                          status=503)
+
+    _pods_answer(monkeypatch, _refuse)
+    mgr = om.OcrManager()
+    assert await _read(mgr, types.SimpleNamespace()) is None
+    out = capsys.readouterr().out
+    assert "REFUSE to run it" in out
+    assert "not served over HTTP" in out          # the POD's own words
+
+
+@pytest.mark.anyio
+async def test_a_refusal_stops_the_rest_of_the_document(monkeypatch):
+    """One failed round trip, not one per page of an 1800-page filing."""
+    _two_page_pdf_split(monkeypatch, pages=40)
+    calls = []
+
+    def _refuse(*_a, **_k):
+        calls.append(1)
+        raise om.OcrError("pod OCR answered 503: nope", status=503)
+
+    _pods_answer(monkeypatch, _refuse)
+    mgr = om.OcrManager()
+    assert await _read(mgr, types.SimpleNamespace()) is None
+    assert len(calls) < 40
+
+
+@pytest.mark.anyio
+async def test_the_pool_is_set_aside_so_the_next_document_does_not_retry(
+        monkeypatch):
+    _two_page_pdf_split(monkeypatch)
+    calls = []
+
+    def _refuse(*_a, **_k):
+        calls.append(1)
+        raise om.OcrError("pod OCR answered 503: nope", status=503)
+
+    _pods_answer(monkeypatch, _refuse)
+    mgr = om.OcrManager()
+    await _read(mgr, types.SimpleNamespace())
+    before = len(calls)
+    await _read(mgr, types.SimpleNamespace())
+    assert len(calls) == before, "second document re-asked a pool already set aside"
+
+
+@pytest.mark.anyio
+async def test_the_pool_is_tried_again_once_the_cooldown_passes(monkeypatch):
+    """A rebuilt image must take effect without restarting the orchestrator."""
+    _two_page_pdf_split(monkeypatch)
+    calls = []
+
+    def _refuse(*_a, **_k):
+        calls.append(1)
+        raise om.OcrError("pod OCR answered 503: nope", status=503)
+
+    _pods_answer(monkeypatch, _refuse)
+    mgr = om.OcrManager()
+    await _read(mgr, types.SimpleNamespace())
+    om._POD_PIPELINE_UNUSABLE["datalab-to/surya-ocr-2"] = (0.0, "stale")
+    before = len(calls)
+    await _read(mgr, types.SimpleNamespace())
+    assert len(calls) > before
+
+
+@pytest.mark.anyio
+async def test_a_real_page_failure_still_fails(monkeypatch):
+    """Falling back on ANY error would hide a broken page behind a slow one."""
+    _two_page_pdf_split(monkeypatch)
+
+    def _boom(*_a, **_k):
+        raise om.OcrError("pod OCR answered 400: that is not a PDF", status=400)
+
+    _pods_answer(monkeypatch, _boom)
+    mgr = om.OcrManager()
+    with pytest.raises(om.OcrError):
+        await _read(mgr, types.SimpleNamespace())
+
+
+@pytest.mark.anyio
+async def test_a_working_pod_is_untouched(monkeypatch):
+    _two_page_pdf_split(monkeypatch)
+    _pods_answer(monkeypatch,
+                 lambda *_a, **_k: {"engine": "surya",
+                                    "pages": [{"text": "ok", "index": 0}]})
+    mgr = om.OcrManager()
+    got = await _read(mgr, types.SimpleNamespace())
+    assert got is not None
+    assert got[0] == "surya"