ocr: a ruined page is not an error, and nothing was catching it (v0.3.79)

From last night's run: ~97% of a 1,000-document batch came out truncated,
with no error anywhere. The engine answered 200 every time. Its decode had
fallen into a loop, run to the token cap emitting the same fragment, and
the resumer correctly refused to continue a reply that was already
repeating itself -- so the page was flagged, returned, and indexed.

`fallback_engines` never fired, because it only fires on an exception.
The same pages read cleanly on another engine (38,055 characters against a
truncated stub), so the recovery was there the whole time and nothing
reached for it.

`ocr.retry_degraded` (default "auto") re-reads ONLY the degraded pages on
another engine and keeps the replacement only if it is not degraded in
turn. Bounded by construction: a clean document costs nothing, a ruined
one costs at most a second pass over its ruined pages, and
`retry_degraded_max_pages` caps even that. The page keeps its own index
and carries `degraded_retry` saying who re-read it and why; the response
reports `recovered_pages`.

Degraded means looping or truncated -- the two the engine itself attests.
NOT a short page, a blank page or an unreadable one: those have their own
meaning and re-reading them spends a card to be told the same thing.

Which leaves the other half of the report: olmOCR returned 4,046
characters where chandra read 30,983 on the same page, ending cleanly well
short of the cap, flagging nothing. The only evidence an early stop leaves
is the company it keeps, so `retry_short_pages_below` compares a page with
the DOCUMENT'S OWN median rather than a character count -- a threshold in
characters would need re-tuning for every language, DPI and layout.
Default 0, OFF, and deliberately: this signal is inferred where the others
are attested, and a genuine title or signature page looks identical. A
short page is also only replaced by a LONGER one.

Off on pods: only the orchestrator can reach another engine's pods, so a
pod would re-read on the engine that just produced the flag.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HXRMe9ZPUQCCCtBai9uqt3

codai/__init__.py

@@ -16,7 +16,7 @@
 
 # Canonical product version for CoderAI — single source of truth. Both the API
 # metadata and the admin web UI read from here.
-__version__ = "0.3.78"
+__version__ = "0.3.79"
 __tagline__ = "Complete orchestration, distribution and escalation of remotizable advanced inference."
 
 # Configure the CUDA caching allocator BEFORE torch is imported anywhere.

codai/admin/routes.py

@@ -4180,6 +4180,10 @@ def build_settings_dict(c, gpu_cards):
             "vlm_max_num_seqs": c.ocr.vlm_max_num_seqs,
             "vlm_max_model_len": c.ocr.vlm_max_model_len,
             "fallback_engines": c.ocr.fallback_engines,
+            "retry_degraded": c.ocr.retry_degraded,
+            "retry_degraded_max_pages": c.ocr.retry_degraded_max_pages,
+            "retry_short_pages_below": c.ocr.retry_short_pages_below,
+            "retry_short_pages_min_doc": c.ocr.retry_short_pages_min_doc,
             "olmocr_enabled": c.ocr.olmocr_enabled,
             "olmocr_instances": c.ocr.olmocr_instances,
             "olmocr_serve": c.ocr.olmocr_serve,
@@ -5043,6 +5047,10 @@ async def api_save_settings(request: Request, username: str = Depends(require_ad
         if "build_retry_cooldown_s" in d: o.build_retry_cooldown_s = _f("build_retry_cooldown_s", o.build_retry_cooldown_s, 0.0)
         if "vlm_gpu_memory_utilization" in d: o.vlm_gpu_memory_utilization = _f("vlm_gpu_memory_utilization", o.vlm_gpu_memory_utilization, 0.0)
         if "fallback_engines" in d: o.fallback_engines = (d.get("fallback_engines") or "auto").strip()
+        if "retry_degraded" in d: o.retry_degraded = (d.get("retry_degraded") or "auto").strip()
+        if "retry_degraded_max_pages" in d: o.retry_degraded_max_pages = max(0, int(d.get("retry_degraded_max_pages") or 0))
+        if "retry_short_pages_below" in d: o.retry_short_pages_below = max(0.0, min(1.0, float(d.get("retry_short_pages_below") or 0.0)))
+        if "retry_short_pages_min_doc" in d: o.retry_short_pages_min_doc = max(3, int(d.get("retry_short_pages_min_doc") or 4))
         if "vlm_max_num_batched_tokens" in d: o.vlm_max_num_batched_tokens = _i("vlm_max_num_batched_tokens", o.vlm_max_num_batched_tokens, 0)
         if "vlm_max_num_seqs" in d: o.vlm_max_num_seqs = _i("vlm_max_num_seqs", o.vlm_max_num_seqs, 0)
         if "vlm_max_model_len" in d: o.vlm_max_model_len = _i("vlm_max_model_len", o.vlm_max_model_len, 0)

codai/admin/templates/settings.html

@@ -1200,6 +1200,28 @@
         <div class="form-hint">Tried when the chosen engine can't serve, so a document
           is never refused: <code>auto</code>, <code>off</code>, or an explicit order.</div>
       </div>
+      <div class="form-group" style="flex:1;min-width:170px">
+        <label class="form-label">Re-read degraded pages on</label>
+        <input type="text" id="s-ocr-retry-degraded" class="form-input" placeholder="auto">
+        <div class="form-hint">A page can come back looping or cut off at the token cap
+          with the engine answering <code>200</code>, so <em>Fallback engines</em> never
+          fires. These re-read only those pages elsewhere and keep whichever answer
+          isn't degraded: <code>auto</code>, <code>off</code>, or an explicit order.</div>
+      </div>
+      <div class="form-group" style="flex:1;min-width:150px">
+        <label class="form-label">Re-read cap (pages/doc)</label>
+        <input type="number" id="s-ocr-retry-degraded-max" class="form-input" min="0" placeholder="0">
+        <div class="form-hint">0 = no limit. A scan the whole fleet reads badly would
+          otherwise pay twice for every page to arrive in the same place.</div>
+      </div>
+      <div class="form-group" style="flex:1;min-width:150px">
+        <label class="form-label">Re-read pages under × median</label>
+        <input type="number" id="s-ocr-retry-short" class="form-input" min="0" max="1" step="0.05" placeholder="0">
+        <div class="form-hint">0 = off. Catches an engine that stops early without
+          flagging anything, by comparing a page with the document's own median.
+          <strong>Inferred, not attested</strong> — a real title or signature page looks
+          the same, so leave it off unless your documents have none.</div>
+      </div>
     </div>
 
     <div class="card-title" style="font-size:13px;margin-top:.4rem">PaddleOCR</div>
@@ -2232,6 +2254,9 @@ async function loadSettings(){
     document.getElementById('s-ocr-retry-cooldown').value = ocr.build_retry_cooldown_s ?? 60;
     document.getElementById('s-ocr-vlm-gmu').value = ocr.vlm_gpu_memory_utilization ?? 0;
     document.getElementById('s-ocr-fallback').value = ocr.fallback_engines ?? 'auto';
+    document.getElementById('s-ocr-retry-degraded').value = ocr.retry_degraded ?? 'auto';
+    document.getElementById('s-ocr-retry-degraded-max').value = ocr.retry_degraded_max_pages ?? 0;
+    document.getElementById('s-ocr-retry-short').value = ocr.retry_short_pages_below ?? 0;
     document.getElementById('s-ocr-vlm-mnbt').value = ocr.vlm_max_num_batched_tokens ?? 4096;
     document.getElementById('s-ocr-vlm-mns').value = ocr.vlm_max_num_seqs ?? 16;
     document.getElementById('s-ocr-vlm-mml').value = ocr.vlm_max_model_len ?? 0;
@@ -2533,6 +2558,9 @@ async function saveSettings(silent){
       build_retry_cooldown_s: parseFloat(document.getElementById('s-ocr-retry-cooldown').value) || 0,
       vlm_gpu_memory_utilization: parseFloat(document.getElementById('s-ocr-vlm-gmu').value) || 0,
       fallback_engines: document.getElementById('s-ocr-fallback').value.trim() || 'auto',
+      retry_degraded: document.getElementById('s-ocr-retry-degraded').value.trim() || 'auto',
+      retry_degraded_max_pages: parseInt(document.getElementById('s-ocr-retry-degraded-max').value) || 0,
+      retry_short_pages_below: parseFloat(document.getElementById('s-ocr-retry-short').value) || 0,
       vlm_max_num_batched_tokens: parseInt(document.getElementById('s-ocr-vlm-mnbt').value) || 0,
       vlm_max_num_seqs: parseInt(document.getElementById('s-ocr-vlm-mns').value) || 0,
       vlm_max_model_len: parseInt(document.getElementById('s-ocr-vlm-mml').value) || 0,

codai/api/runpod_worker.py

@@ -1336,6 +1336,11 @@ def _plan(engine, image, args, mcfg, api_key, entry, served,
         # This image carries NO docTR -- `auto` would route a failed page to an
         # engine that is not there. This engine or nothing, said out loud.
         env["CODERAI_OCR_FALLBACK_ENGINES"] = "off"
+        # Same reasoning for the degraded-page re-read, and one more: only the
+        # ORCHESTRATOR can reach another engine's pods. A pod re-reading its own
+        # degraded page would re-read it on the one engine that just produced
+        # it, which is the retry that already failed.
+        env["CODERAI_OCR_RETRY_DEGRADED"] = "off"
         # And the TUNING. Every one of these was set in response to a measured
         # failure, and a pod reading pages at surya's defaults instead would be
         # indistinguishable from the "pre-fix pod image" that got blamed three

codai/config.py

@@ -1084,6 +1084,30 @@ class OcrConfig:
     # "off"/"none" = no fallback (fail with the original error); or an explicit
     # comma/space separated order.
     fallback_engines: str = "auto"
+    # A page can come back DEGRADED without the engine ever failing: the reply
+    # looped until it hit the token cap, or it stopped at the cap and could not
+    # be resumed. `fallback_engines` never fires for those -- the call returned
+    # 200 -- so a run can come out almost entirely truncated with no error
+    # anywhere. These re-read ONLY the degraded pages on another engine and keep
+    # whichever answer is not degraded. "auto" = the fallback order; "off" = keep
+    # what the first engine produced; or an explicit comma/space separated order.
+    retry_degraded: str = "auto"
+    # Stop re-reading after this many pages of one document, 0 = no limit. A
+    # scan the whole fleet reads badly would otherwise pay twice for every page
+    # to arrive at the same place.
+    retry_degraded_max_pages: int = 0
+    # Re-read a page whose text is below this FRACTION of the document's median
+    # page, even though the engine flagged nothing. 0 = off, which is the
+    # default, because unlike a looping or truncated page this one is INFERRED:
+    # the engine says the page is fine and we are disagreeing with it from the
+    # shape of the document around it. A real early stop looks exactly like a
+    # title page or a signature page, and only the operator knows which their
+    # corpus has. 0.25 catches an engine that returned an eighth of what its
+    # neighbours did without touching a document of genuinely uneven pages.
+    retry_short_pages_below: float = 0.0
+    # ...and only in a document with at least this many pages, since a median
+    # taken over two pages is not a median.
+    retry_short_pages_min_doc: int = 4
 
     # --- PaddleOCR / PP-Structure ---
     paddle_enabled: bool = True
@@ -2011,6 +2035,10 @@ class ConfigManager:
                 "vlm_max_num_seqs": self.config.ocr.vlm_max_num_seqs,
                 "vlm_max_model_len": self.config.ocr.vlm_max_model_len,
                 "fallback_engines": self.config.ocr.fallback_engines,
+                "retry_degraded": self.config.ocr.retry_degraded,
+                "retry_degraded_max_pages": self.config.ocr.retry_degraded_max_pages,
+                "retry_short_pages_below": self.config.ocr.retry_short_pages_below,
+                "retry_short_pages_min_doc": self.config.ocr.retry_short_pages_min_doc,
                 "olmocr_enabled": self.config.ocr.olmocr_enabled,
                 "olmocr_instances": self.config.ocr.olmocr_instances,
                 "olmocr_serve": self.config.ocr.olmocr_serve,
@@ -2162,7 +2190,8 @@ class ConfigManager:
                                 ("CODERAI_OCR_CHANDRA_MODEL", "chandra_model"),
                                 ("CODERAI_OCR_CHANDRA_MODEL_ID", "chandra_model_id"),
                                 ("CODERAI_OCR_CHANDRA_SERVER_URL", "chandra_server_url"),
-                                ("CODERAI_OCR_FALLBACK_ENGINES", "fallback_engines")):
+                                ("CODERAI_OCR_FALLBACK_ENGINES", "fallback_engines"),
+                                ("CODERAI_OCR_RETRY_DEGRADED", "retry_degraded")):
                 _val = (_os.environ.get(_env) or "").strip()
                 if _val:
                     setattr(self.config.ocr, _attr, _val)

codai/ocr/manager.py

@@ -35,7 +35,7 @@ from typing import Dict, List, Optional
 
 from codai.ocr.base import (OcrEngine, OcrPage, OcrError, ink_coverage,
                             PAGE_OPTIONS, page_estimate, render_pages,
-                            iter_pdf_page_pdfs, pdf_page_count,
+                            iter_pdf_page_pdfs, pdf_page_count, split_pdf_page,
                             unreadable_report)
 from codai.ocr.budget import get_budget
 
@@ -1247,6 +1247,14 @@ class OcrManager:
                     data, filename, content_type, use_dpi, window,
                     engine=engine, lang=lang, detect=detect,
                     page_opts=page_opts, cfg=cfg)
+        # A page can be ruined without anything failing -- see `_retry_degraded`.
+        # Done here rather than inside either pipeline, because the two produce
+        # the same degraded pages for the same reasons and a reader of the
+        # result cannot tell which path it came from.
+        recovered = await self._retry_degraded(
+            data, filename, content_type, name, pages, dpi=use_dpi,
+            page_opts=page_opts, detect=detect, lang=lang, cfg=cfg,
+            window=window)
         full_text = "\n\n".join(p.text for p in pages)
 
         structured_out = None
@@ -1280,6 +1288,11 @@ class OcrManager:
         report = unreadable_report(pages)
         if report:
             out["unreadable"] = report
+        if recovered:
+            # Top level, because a caller deciding whether to trust this
+            # document should not have to walk the pages to find out that part
+            # of it was read by a different engine than the one it asked for.
+            out["recovered_pages"] = recovered
         return out
 
     def _raster_workers(self, cfg) -> int:
@@ -1289,6 +1302,238 @@ class OcrManager:
         except Exception:
             return 0
 
+    # ------------------------------------------------------------------
+    # Degraded pages
+    # ------------------------------------------------------------------
+    @staticmethod
+    def _page_degraded(page) -> str:
+        """Why this page is not usable, or "" when it is.
+
+        A degraded page is NOT an error. The engine answered 200 with
+        plausible-looking text, and the only evidence is in the diagnostics it
+        attached. Two causes, both of which make the page incomplete:
+
+        * ``repeat_detected`` -- the decode fell into a loop and ran to the cap
+          emitting the same fragment. Retrying the SAME engine at a higher
+          temperature is what produced this flag; it does not clear it.
+        * ``truncated`` -- it stopped at the cap and could not be continued.
+
+        Deliberately NOT degraded: a short page, a blank page, or a page the
+        engine reported it could not read. Those have their own handling and
+        their own meaning, and re-reading them on a second engine would spend a
+        card to learn what the first engine already said correctly.
+        """
+        meta = getattr(page, "meta", None) or {}
+        if meta.get("repeat_detected"):
+            return "looping"
+        info = meta.get("model_attempts") or {}
+        if isinstance(info, dict) and info.get("truncated"):
+            return "truncated"
+        return ""
+
+    @staticmethod
+    def _short_pages(pages: List, cfg) -> dict:
+        """Pages far shorter than the document around them → {index: reason}.
+
+        Reported from the same run: olmOCR returned 4,046 characters where
+        chandra read 30,983 on the SAME page, and its fastest calls returned
+        the least text. Nothing was flagged -- the reply ended cleanly, well
+        short of the cap, so there is no `truncated` and no loop. The only
+        evidence an early stop leaves is the company it keeps.
+
+        So the measure is the document's OWN median, not a character count:
+        a threshold in characters would have to be re-tuned for every language,
+        DPI, layout and font, and would be wrong for all of them. A page at an
+        eighth of its neighbours is an early stop whatever the corpus.
+
+        OFF by default (`retry_short_pages_below = 0`) and deliberately so.
+        This signal is INFERRED, where looping and truncation are attested by
+        the engine itself -- and a genuine title page, index or signature page
+        looks exactly the same from here. Only the operator knows whether their
+        documents have those.
+        """
+        try:
+            frac = float(getattr(cfg, "retry_short_pages_below", 0.0) or 0.0)
+            min_doc = int(getattr(cfg, "retry_short_pages_min_doc", 4) or 4)
+        except Exception:
+            return {}
+        if frac <= 0 or len(pages) < max(3, min_doc):
+            return {}
+        lens = sorted(len((getattr(p, "text", "") or "").strip()) for p in pages)
+        mid = len(lens) // 2
+        median = lens[mid] if len(lens) % 2 else (lens[mid - 1] + lens[mid]) / 2
+        if median <= 0:
+            return {}
+        floor = median * frac
+        out = {}
+        for i, page in enumerate(pages):
+            n = len((getattr(page, "text", "") or "").strip())
+            # A page the blank check already explained is not an early stop:
+            # there was nothing on it. Re-reading it buys the same answer from
+            # a second card.
+            if (getattr(page, "meta", None) or {}).get("blank"):
+                continue
+            if n < floor:
+                out[i] = (f"short ({n} chars against a document median of "
+                          f"{int(median)})")
+        return out
+
+    def _retry_degraded_order(self, failed: str) -> List[str]:
+        """Engines to re-read a degraded page with, in order."""
+        cfg = self._require_cfg()
+        raw = str(getattr(cfg, "retry_degraded", "auto") or "auto").strip().lower()
+        if raw in ("off", "none", "no", "false", "0"):
+            return []
+        if raw in ("auto", "", "on", "true", "1"):
+            return self._fallback_order(failed)
+        names = [n.strip() for n in raw.replace(",", " ").split() if n.strip()]
+        return [n for n in names
+                if n != failed and n in _ENGINE_FACTORIES and _engine_enabled(cfg, n)]
+
+    async def _read_page_blobs(self, blobs, engine: str, *, filename: str,
+                               content_type: str, dpi: int, page_opts, detect,
+                               lang, cfg, window: int):
+        """Read a handful of one-page documents with ONE named engine.
+
+        Takes whichever of the two pipelines that engine's configuration
+        selects, exactly as a whole document would -- a retry that quietly used
+        the other path would be measuring the path, not the engine.
+        """
+        out = []
+        remote = pod_pipeline_model(cfg, engine)
+        for blob in blobs:
+            got = None
+            if remote:
+                try:
+                    got = await self._read_on_pods(
+                        blob, filename, content_type, remote, dpi=dpi,
+                        page_opts=page_opts, detect=detect, cfg=cfg,
+                        engine_name=engine)
+                except Exception as exc:
+                    print(f"[ocr] re-reading on '{engine}' pods failed: {exc}",
+                          flush=True)
+                    got = None
+            if got is not None:
+                pages = got[1]
+            else:
+                _n, pages, _b, _s, _g = await self._read_windowed(
+                    blob, filename, content_type, dpi, window, engine=engine,
+                    lang=lang, detect=detect, page_opts=page_opts, cfg=cfg)
+            out.append(pages[0] if pages else None)
+        return out
+
+    async def _retry_degraded(self, data: bytes, filename: str, content_type: str,
+                              name: str, pages: List, *, dpi: int, page_opts,
+                              detect, lang, cfg, window: int) -> str:
+        """Re-read every degraded page on another engine. Returns a note, or "".
+
+        Mutates `pages` in place, replacing only those pages that come back
+        clean. This exists because an OCR run can fail almost completely with
+        no error anywhere: a repeating decode answers 200, `fallback_engines`
+        never fires, and a thousand documents land in the index truncated.
+
+        Bounded by construction -- only pages already known to be unusable are
+        re-read, so a clean document costs nothing and a ruined one costs at
+        most a second pass over its ruined pages.
+        """
+        why = {}
+        for i, page in enumerate(pages):
+            reason = self._page_degraded(page)
+            if reason:
+                why[i] = reason
+        for i, reason in self._short_pages(pages, cfg).items():
+            why.setdefault(i, reason)
+        bad = sorted(why)
+        if not bad:
+            return ""
+        chain = self._retry_degraded_order(name)
+        if not chain:
+            return ""
+        try:
+            cap = max(0, int(getattr(cfg, "retry_degraded_max_pages", 0) or 0))
+        except Exception:
+            cap = 0
+        dropped = 0
+        if cap and len(bad) > cap:
+            dropped = len(bad) - cap
+            bad = bad[:cap]
+        is_pdf = ((content_type or "").lower() == "application/pdf"
+                  or (filename or "").lower().endswith(".pdf")
+                  or data[:5] == b"%PDF-")
+        if not is_pdf:
+            if len(pages) != 1:
+                return ""
+            blobs = {0: data}
+        else:
+            blobs = {}
+            for i in bad:
+                try:
+                    blobs[i] = split_pdf_page(data, i)
+                except Exception as exc:
+                    print(f"[ocr] cannot isolate page {i} to re-read it: {exc}",
+                          flush=True)
+        bad = [i for i in bad if i in blobs]
+        if not bad:
+            return ""
+        print(f"[ocr] {len(bad)} of {len(pages)} page(s) came back DEGRADED from "
+              f"'{name}' ("
+              + ", ".join(f"p{i}:{why[i]}" for i in bad[:8])
+              + (" ..." if len(bad) > 8 else "")
+              + f") — the engine answered 200, so this is not a fallback: "
+              f"re-reading those pages on {' -> '.join(chain)}", flush=True)
+        if dropped:
+            print(f"[ocr] and leaving {dropped} more as they are "
+                  f"(ocr.retry_degraded_max_pages)", flush=True)
+        fixed: dict = {}
+        for alt in chain:
+            todo = [i for i in bad if i not in fixed]
+            if not todo:
+                break
+            try:
+                got = await self._read_page_blobs(
+                    [blobs[i] for i in todo], alt, filename=filename,
+                    content_type=content_type, dpi=dpi, page_opts=page_opts,
+                    detect=detect, lang=lang, cfg=cfg, window=window)
+            except Exception as exc:
+                print(f"[ocr] re-read on '{alt}' failed outright: {exc}", flush=True)
+                continue
+            for i, page in zip(todo, got):
+                if page is None or self._page_degraded(page):
+                    continue
+                if why[i].startswith("short"):
+                    # The complaint was length, so the answer has to be longer.
+                    # A second engine that stops just as early has told us the
+                    # page really is short, and taking its text would hide that
+                    # behind a different engine's name.
+                    if len((page.text or "").strip()) <= \
+                            len((pages[i].text or "").strip()):
+                        continue
+                # Keep the ORIGINAL page index: the caller indexes by it, and a
+                # page re-read on its own always comes back as page 0.
+                page.index = pages[i].index
+                meta = dict(getattr(page, "meta", None) or {})
+                meta["degraded_retry"] = {
+                    "from": name, "to": alt, "reason": why[i],
+                    "why": (f"'{name}' returned this page {why[i]}, which is a "
+                            f"200 and not a failure, so it was re-read on "
+                            f"'{alt}' and that answer is the one here"),
+                }
+                page.meta = meta
+                pages[i] = page
+                fixed[i] = alt
+        if not fixed:
+            print(f"[ocr] no engine in {' -> '.join(chain)} read those pages any "
+                  f"better — keeping what '{name}' produced, still flagged",
+                  flush=True)
+            return ""
+        by_engine: dict = {}
+        for alt in fixed.values():
+            by_engine[alt] = by_engine.get(alt, 0) + 1
+        note = ", ".join(f"{n} page(s) by {e}" for e, n in sorted(by_engine.items()))
+        print(f"[ocr] recovered {len(fixed)} of {len(bad)} degraded page(s): "
+              f"{note}", flush=True)
+        return note
+
     async def _read_on_pods(self, data: bytes, filename: str, content_type: str,
                             model_id: str, *, dpi: int, page_opts, detect, cfg,
                             engine_name: str = "surya"):

tests/test_ocr_degraded_retry.py

+"""A page can be RUINED without anything ever failing.
+
+Reported from a 1,000-document run: ~97% came out `troncato`. Nothing raised,
+nothing was logged as an error, and `fallback_engines` never fired -- because
+the engine answered 200 every time. Its decode had fallen into a loop, run to
+the token cap emitting the same fragment, and the resumer correctly refused to
+continue a reply that was already repeating itself. The flag was set, the page
+was returned, and the index swallowed it.
+
+The same pages read cleanly on another engine (38,055 characters against a
+truncated stub). So the recovery was available the whole time and nothing
+reached for it: `fallback_engines` only fires on an exception, and there was
+none.
+
+These tests hold the shape of the fix:
+
+* a degraded page is re-read on ANOTHER engine, not retried on the one that
+  produced it (that retry is what set the flag);
+* only degraded pages are re-read, so a clean document costs nothing;
+* the replacement is kept only if it is NOT degraded in turn;
+* the page keeps its own index and says who re-read it and why;
+* it is OFF on a pod, because only the orchestrator can reach another engine.
+"""
+
+import asyncio
+import io
+import pathlib
+import sys
+
+import pytest
+
+ROOT = pathlib.Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT))
+
+from codai.ocr import manager as om                            # noqa: E402
+from codai.ocr.base import OcrPage                             # noqa: E402
+
+
+def _pdf(n=3):
+    pdfium = pytest.importorskip("pypdfium2")
+    d = pdfium.PdfDocument.new()
+    for _ in range(n):
+        d.new_page(595, 842)
+    b = io.BytesIO()
+    d.save(b)
+    return b.getvalue()
+
+
+def _looping(i, text="riga riga riga"):
+    return OcrPage(index=i, text=text, meta={"repeat_detected": True})
+
+
+def _truncated(i, text="meta pagina"):
+    return OcrPage(index=i, text=text,
+                   meta={"model_attempts": {"truncated": True, "complete": False}})
+
+
+def _clean(i, text="una pagina intera, letta per bene."):
+    return OcrPage(index=i, text=text)
+
+
+class _Cfg:
+    retry_degraded = "auto"
+    retry_degraded_max_pages = 0
+    fallback_engines = "auto"
+
+
+def _mgr(cfg=None):
+    m = om.OcrManager.__new__(om.OcrManager)
+    m._cfg = cfg or _Cfg()
+    m._sem = None
+    m._require_cfg = lambda: m._cfg
+    return m
+
+
+# ------------------------------------------------------- what counts as bad #
+def test_a_looping_page_is_degraded():
+    assert om.OcrManager._page_degraded(_looping(0)) == "looping"
+
+
+def test_a_truncated_page_is_degraded():
+    assert om.OcrManager._page_degraded(_truncated(0)) == "truncated"
+
+
+def test_a_clean_page_is_not():
+    assert om.OcrManager._page_degraded(_clean(0)) == ""
+
+
+def test_a_short_page_is_not_degraded():
+    """A short page has four causes and they need opposite handling. Only the
+    two the engine itself flags are re-read; `unreadable_report` keeps the rest,
+    because re-reading a genuinely blank page spends a card to be told again
+    that it is blank."""
+    assert om.OcrManager._page_degraded(OcrPage(index=0, text="")) == ""
+    assert om.OcrManager._page_degraded(OcrPage(index=0, text="x")) == ""
+
+
+def test_a_resumed_but_complete_page_is_not_degraded():
+    """`resumed` without `truncated` means the reply ran past the cap and WAS
+    finished. That page is whole and must not be read twice."""
+    p = OcrPage(index=0, text="lunga ma intera",
+                meta={"model_attempts": {"resumed": 3, "calls": 4}})
+    assert om.OcrManager._page_degraded(p) == ""
+
+
+def test_degradation_survives_a_meta_that_is_not_a_dict():
+    p = OcrPage(index=0, text="x", meta={"model_attempts": "oops"})
+    assert om.OcrManager._page_degraded(p) == ""
+
+
+# ------------------------------------------------------------- which engines #
+def test_auto_uses_the_fallback_order():
+    m = _mgr()
+    m._fallback_order = lambda failed: ["chandra", "doctr"]
+    assert m._retry_degraded_order("surya") == ["chandra", "doctr"]
+
+
+def test_off_means_keep_what_the_first_engine_produced():
+    cfg = _Cfg()
+    cfg.retry_degraded = "off"
+    m = _mgr(cfg)
+    m._fallback_order = lambda failed: ["chandra"]
+    assert m._retry_degraded_order("surya") == []
+
+
+def test_an_explicit_order_never_includes_the_engine_that_failed():
+    """Re-reading on the same engine is the retry that already produced the
+    flag -- chandra raises the temperature itself and still looped."""
+    cfg = _Cfg()
+    cfg.retry_degraded = "surya, chandra"
+    m = _mgr(cfg)
+    m._fallback_order = lambda failed: []
+    order = m._retry_degraded_order("surya")
+    assert "surya" not in order
+
+
+# ------------------------------------------------------------ the re-read #
+def _run(coro):
+    return asyncio.get_event_loop_policy().new_event_loop().run_until_complete(coro)
+
+
+def _wire(m, answers, seen):
+    """`answers` maps engine name -> callable(i) -> OcrPage or None."""
+    async def _read(blobs, engine, **kw):
+        seen.append((engine, len(blobs)))
+        return [answers[engine](i) for i in range(len(blobs))]
+    m._read_page_blobs = _read
+
+
+def test_only_the_degraded_pages_are_re_read():
+    m = _mgr()
+    m._fallback_order = lambda f: ["chandra"]
+    pages = [_clean(0), _looping(1), _clean(2)]
+    seen = []
+    _wire(m, {"chandra": lambda i: _clean(99)}, seen)
+    note = _run(m._retry_degraded(_pdf(3), "a.pdf", "application/pdf", "surya",
+                                  pages, dpi=200, page_opts={}, detect=None,
+                                  lang="", cfg=m._cfg, window=8))
+    assert seen == [("chandra", 1)], seen
+    assert note and "chandra" in note
+    assert pages[0].text == "una pagina intera, letta per bene."
+    assert pages[1].text == "una pagina intera, letta per bene."
+
+
+def test_a_clean_document_costs_nothing():
+    m = _mgr()
+    m._fallback_order = lambda f: ["chandra"]
+    seen = []
+    _wire(m, {"chandra": lambda i: _clean(0)}, seen)
+    pages = [_clean(0), _clean(1)]
+    note = _run(m._retry_degraded(_pdf(2), "a.pdf", "application/pdf", "surya",
+                                  pages, dpi=200, page_opts={}, detect=None,
+                                  lang="", cfg=m._cfg, window=8))
+    assert seen == [] and note == ""
+
+
+def test_the_replaced_page_keeps_its_own_index():
+    """A page re-read on its own always comes back as page 0. Callers index
+    into `pages` by page number, so the original index has to survive."""
+    m = _mgr()
+    m._fallback_order = lambda f: ["chandra"]
+    pages = [_clean(0), _clean(1), _looping(2)]
+    _wire(m, {"chandra": lambda i: _clean(0)}, [])
+    _run(m._retry_degraded(_pdf(3), "a.pdf", "application/pdf", "surya", pages,
+                           dpi=200, page_opts={}, detect=None, lang="",
+                           cfg=m._cfg, window=8))
+    assert [p.index for p in pages] == [0, 1, 2]
+
+
+def test_the_replaced_page_says_who_read_it_and_why():
+    m = _mgr()
+    m._fallback_order = lambda f: ["chandra"]
+    pages = [_looping(0)]
+    _wire(m, {"chandra": lambda i: _clean(0)}, [])
+    _run(m._retry_degraded(_pdf(1), "a.pdf", "application/pdf", "surya", pages,
+                           dpi=200, page_opts={}, detect=None, lang="",
+                           cfg=m._cfg, window=8))
+    got = pages[0].meta["degraded_retry"]
+    assert got["from"] == "surya" and got["to"] == "chandra"
+    assert got["reason"] == "looping"
+    assert "200" in got["why"]
+
+
+def test_a_replacement_that_is_also_degraded_is_refused():
+    """Keeping a second truncated page would hide the problem behind a
+    different engine's name."""
+    m = _mgr()
+    m._fallback_order = lambda f: ["chandra"]
+    pages = [_looping(0, "originale")]
+    _wire(m, {"chandra": lambda i: _truncated(0, "altrettanto rotta")}, [])
+    note = _run(m._retry_degraded(_pdf(1), "a.pdf", "application/pdf", "surya",
+                                  pages, dpi=200, page_opts={}, detect=None,
+                                  lang="", cfg=m._cfg, window=8))
+    assert note == ""
+    assert pages[0].text == "originale"
+    assert "degraded_retry" not in pages[0].meta
+    assert pages[0].meta["repeat_detected"] is True
+
+
+def test_the_chain_is_walked_until_one_engine_reads_the_page():
+    m = _mgr()
+    m._fallback_order = lambda f: ["olmocr", "chandra"]
+    pages = [_looping(0)]
+    seen = []
+    _wire(m, {"olmocr": lambda i: _truncated(0),
+              "chandra": lambda i: _clean(0)}, seen)
+    _run(m._retry_degraded(_pdf(1), "a.pdf", "application/pdf", "surya", pages,
+                           dpi=200, page_opts={}, detect=None, lang="",
+                           cfg=m._cfg, window=8))
+    assert [e for e, _ in seen] == ["olmocr", "chandra"]
+    assert pages[0].meta["degraded_retry"]["to"] == "chandra"
+
+
+def test_a_later_engine_is_only_asked_for_the_pages_still_broken():
+    m = _mgr()
+    m._fallback_order = lambda f: ["olmocr", "chandra"]
+    pages = [_looping(0), _looping(1)]
+    seen = []
+    _wire(m, {"olmocr": lambda i: _clean(0) if i == 0 else _truncated(0),
+              "chandra": lambda i: _clean(0)}, seen)
+    _run(m._retry_degraded(_pdf(2), "a.pdf", "application/pdf", "surya", pages,
+                           dpi=200, page_opts={}, detect=None, lang="",
+                           cfg=m._cfg, window=8))
+    assert seen == [("olmocr", 2), ("chandra", 1)], seen
+
+
+def test_an_engine_that_raises_outright_does_not_lose_the_document():
+    m = _mgr()
+    m._fallback_order = lambda f: ["olmocr", "chandra"]
+    pages = [_looping(0, "quel che c'e")]
+
+    async def _read(blobs, engine, **kw):
+        if engine == "olmocr":
+            raise RuntimeError("no pod")
+        return [_clean(0)]
+    m._read_page_blobs = _read
+    note = _run(m._retry_degraded(_pdf(1), "a.pdf", "application/pdf", "surya",
+                                  pages, dpi=200, page_opts={}, detect=None,
+                                  lang="", cfg=m._cfg, window=8))
+    assert "chandra" in note
+
+
+def test_the_cap_bounds_what_one_document_may_spend():
+    cfg = _Cfg()
+    cfg.retry_degraded_max_pages = 2
+    m = _mgr(cfg)
+    m._fallback_order = lambda f: ["chandra"]
+    pages = [_looping(i) for i in range(5)]
+    seen = []
+    _wire(m, {"chandra": lambda i: _clean(0)}, seen)
+    _run(m._retry_degraded(_pdf(5), "a.pdf", "application/pdf", "surya", pages,
+                           dpi=200, page_opts={}, detect=None, lang="",
+                           cfg=cfg, window=8))
+    assert seen == [("chandra", 2)], seen
+    assert om.OcrManager._page_degraded(pages[4]) == "looping"
+
+
+def test_a_single_image_is_re_read_whole():
+    """No PDF to split: the one page IS the document."""
+    m = _mgr()
+    m._fallback_order = lambda f: ["chandra"]
+    pages = [_looping(0)]
+    seen = []
+    _wire(m, {"chandra": lambda i: _clean(0)}, seen)
+    note = _run(m._retry_degraded(b"\x89PNG\r\n\x1a\n...", "p.png", "image/png",
+                                  "surya", pages, dpi=200, page_opts={},
+                                  detect=None, lang="", cfg=m._cfg, window=8))
+    assert seen == [("chandra", 1)] and "chandra" in note
+
+
+# ------------------------------------------------------------ on the pods #
+def test_a_pod_never_re_reads_its_own_degraded_page():
+    """Only the orchestrator can reach another engine's pods. A pod would
+    re-read on the engine that just produced the flag."""
+    from codai.api import runpod_worker as rw
+    src = (ROOT / "codai/api/runpod_worker.py").read_text()
+    assert 'env["CODERAI_OCR_RETRY_DEGRADED"] = "off"' in src
+
+
+def test_the_knob_exists_and_defaults_to_auto():
+    from codai.config import OcrConfig
+    assert OcrConfig().retry_degraded == "auto"
+    assert OcrConfig().retry_degraded_max_pages == 0
+
+
+# ------------------------------------------------- the inferred early stop #
+class _ShortCfg(_Cfg):
+    retry_short_pages_below = 0.25
+    retry_short_pages_min_doc = 4
+
+
+def _long(i, n=30000):
+    return OcrPage(index=i, text="x" * n)
+
+
+def test_short_page_detection_is_off_by_default():
+    """It is INFERRED, not attested. A corpus with real title pages would pay
+    for a second read of every one of them."""
+    from codai.config import OcrConfig
+    assert OcrConfig().retry_short_pages_below == 0.0
+    pages = [_long(0), _long(1), _long(2), OcrPage(index=3, text="x" * 100)]
+    assert om.OcrManager._short_pages(pages, _Cfg()) == {}
+
+
+def test_a_page_an_eighth_the_length_of_its_neighbours_is_short():
+    pages = [_long(0), _long(1), _long(2), OcrPage(index=3, text="x" * 4046)]
+    got = om.OcrManager._short_pages(pages, _ShortCfg())
+    assert list(got) == [3]
+    assert "4046 chars" in got[3] and "median of 30000" in got[3]
+
+
+def test_the_measure_is_the_documents_own_median_not_a_character_count():
+    """A document of uniformly short pages has no short page. A threshold in
+    characters would call every page of it an early stop."""
+    pages = [OcrPage(index=i, text="x" * 900) for i in range(6)]
+    assert om.OcrManager._short_pages(pages, _ShortCfg()) == {}
+
+
+def test_a_two_page_document_has_no_median_worth_trusting():
+    pages = [_long(0), OcrPage(index=1, text="x" * 10)]
+    assert om.OcrManager._short_pages(pages, _ShortCfg()) == {}
+
+
+def test_a_page_already_explained_as_blank_is_not_an_early_stop():
+    pages = [_long(0), _long(1), _long(2),
+             OcrPage(index=3, text="", meta={"blank": True, "ink": 0.0001})]
+    assert om.OcrManager._short_pages(pages, _ShortCfg()) == {}
+
+
+def test_a_short_page_is_only_replaced_by_a_LONGER_one():
+    """The complaint was length. A second engine stopping just as early has
+    told us the page really is short."""
+    m = _mgr(_ShortCfg())
+    m._fallback_order = lambda f: ["chandra"]
+    pages = [_long(0), _long(1), _long(2), OcrPage(index=3, text="x" * 4046)]
+    _wire(m, {"chandra": lambda i: OcrPage(index=0, text="y" * 900)}, [])
+    note = _run(m._retry_degraded(_pdf(4), "a.pdf", "application/pdf", "olmocr",
+                                  pages, dpi=200, page_opts={}, detect=None,
+                                  lang="", cfg=m._cfg, window=8))
+    assert note == ""
+    assert pages[3].text == "x" * 4046
+
+
+def test_a_short_page_IS_replaced_when_the_other_engine_reads_more():
+    m = _mgr(_ShortCfg())
+    m._fallback_order = lambda f: ["chandra"]
+    pages = [_long(0), _long(1), _long(2), OcrPage(index=3, text="x" * 4046)]
+    _wire(m, {"chandra": lambda i: OcrPage(index=0, text="y" * 30983)}, [])
+    note = _run(m._retry_degraded(_pdf(4), "a.pdf", "application/pdf", "olmocr",
+                                  pages, dpi=200, page_opts={}, detect=None,
+                                  lang="", cfg=m._cfg, window=8))
+    assert "chandra" in note
+    assert pages[3].text == "y" * 30983
+    assert pages[3].meta["degraded_retry"]["reason"].startswith("short")
+
+
+def test_a_looping_page_is_replaced_on_its_own_terms_not_on_length():
+    """Length is only the test for a page flagged SHORT. A looping page emits
+    the same fragment to the cap, so it is usually the LONGER of the two."""
+    m = _mgr(_ShortCfg())
+    m._fallback_order = lambda f: ["chandra"]
+    pages = [_long(0), _long(1), _long(2), _looping(3, "riga " * 4000)]
+    _wire(m, {"chandra": lambda i: OcrPage(index=0, text="buono")}, [])
+    note = _run(m._retry_degraded(_pdf(4), "a.pdf", "application/pdf", "surya",
+                                  pages, dpi=200, page_opts={}, detect=None,
+                                  lang="", cfg=m._cfg, window=8))
+    assert "chandra" in note and pages[3].text == "buono"