three quarters of rasterisation was zlib (v0.4.7)

Measured on A4 at 200 DPI (3.4-3.9 Mpix), best of three:

    pdfium page.render()               21-46 ms
    PNG encode, PIL default (level 6)  86-353 ms   1846 KB
    PNG encode, level 1 (lossless)        123 ms   2117 KB
    JPEG q95                              8.4 ms    596 KB
    the page's own embedded JPEG           0.1 ms    147 KB

So most of what is called rasterisation is not rasterisation, and on the surya
path it was paid TWICE per page: once in subprocess_engine to hand the page to
the isolated-venv worker, once inside surya's own client to build the data URI
(surya/inference/backends/openai_client.py:32, with "image/png" hard-coded into
the URI it builds). Against roughly 5 s on the GPU per page -- 12 pages/min/pod
at Running: 0-3 reqs, measured on two live pods -- that is CPU the pods need in
order to hold 48 pages in flight.

Two ways out, both shipped, both default:

* A page that already IS an image is sent as the PDF holds it. A scan is
  usually one full-page DCTDecode object, and extract_embedded() takes it in
  0.1 ms with no render, no encode and no resample. Every guard in there is a
  way those bytes would have been a WRONG page rather than a slower one: one
  image object and nothing else (a stamp or a text layer must not be dropped),
  DCTDecode alone (Flate is raw pixels, CCITT and JBIG2 are bitonal, JPX is
  JPEG 2000), grey or RGB only (a CMYK JPEG out of a PDF decodes inverted),
  /Rotate 0 and an axis-aligned positive matrix (otherwise the bytes are a
  sideways page, which reads as gibberish rather than as an error), 98 %
  coverage anchored at the origin, and an effective DPI within
  [min_dpi, max_dpi_factor x target].

* When a page must be encoded, it is encoded once, at zlib level 1. Lossless,
  2.6x faster, and the bytes are carried forward so the request reuses them.

The carrier is an attribute on the PIL image, recorded with the size it was
true for. Any PIL transform returns a NEW object and so drops it, which is the
design: a missed passthrough costs milliseconds, a wrong one shows the model a
plausible page at the wrong resolution. chandra and olmOCR therefore scale
BEFORE converting, since convert() was quietly invalidating every page.

End to end, one page from PDF bytes to data URI:

    a page that must be rendered   610 ms -> 234 ms   (jpeg: 68 ms)
    a scanned page                 610 ms ->  6.9 ms, 1895 KB -> 147 KB

New: ocr.page_image_format (auto | png | jpeg), page_image_jpeg_quality,
page_image_min_dpi, page_image_max_dpi_factor, with the admin API and the
settings form. `png` always re-encodes, which is what a controlled comparison
needs; `jpeg` is lossy and stays opt-in, because whether it costs accuracy is a
question about a corpus and not about a codec.

The rasteriser worker loads page_image by path: it may not import codai.ocr
(377 MB via litellm), and the guards must be the SAME ones the parent applies or
a pool outage would change what the model is shown. ocr_worker.py may not
import even that -- it runs in the engine's own venv -- so its copy of the
carrier convention is tested against page_image's to keep the two from drifting.

42 new tests; 3306 passed with the 13 pre-existing failures (broker, studio,
manual client) untouched.
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.6"
+__version__ = "0.4.7"
 __tagline__ = "Complete orchestration, distribution and escalation of remotizable advanced inference."
 
 # Configure the CUDA caching allocator BEFORE torch is imported anywhere.

codai/admin/routes.py

@@ -4138,6 +4138,10 @@ def build_settings_dict(c, gpu_cards):
             "max_pages": c.ocr.max_pages,
             "pod_pipeline": c.ocr.pod_pipeline,
             "pod_pipeline_wait_s": c.ocr.pod_pipeline_wait_s,
+            "page_image_format": c.ocr.page_image_format,
+            "page_image_jpeg_quality": c.ocr.page_image_jpeg_quality,
+            "page_image_min_dpi": c.ocr.page_image_min_dpi,
+            "page_image_max_dpi_factor": c.ocr.page_image_max_dpi_factor,
             "max_resident_mb": c.ocr.max_resident_mb,
             "page_window": c.ocr.page_window,
             "raster_workers": c.ocr.raster_workers,
@@ -4997,6 +5001,21 @@ async def api_save_settings(request: Request, username: str = Depends(require_ad
                 o.pod_pipeline_wait_s = max(0.0, float(d.get("pod_pipeline_wait_s") or 0))
             except (TypeError, ValueError):
                 pass
+        if "page_image_format" in d:
+            _pf = str(d.get("page_image_format") or "").strip().lower()
+            # An unknown value leaves the setting alone rather than silently
+            # becoming the default: a typo in a scripted PUT must not quietly
+            # change how every page is encoded.
+            o.page_image_format = _pf if _pf in ("auto", "png", "jpeg") else o.page_image_format
+        if "page_image_jpeg_quality" in d:
+            o.page_image_jpeg_quality = max(1, min(100, _i(
+                "page_image_jpeg_quality", o.page_image_jpeg_quality, 1)))
+        for _k, _lo in (("page_image_min_dpi", 0.0), ("page_image_max_dpi_factor", 1.0)):
+            if _k in d:
+                try:
+                    setattr(o, _k, max(_lo, float(d.get(_k) 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

@@ -1150,6 +1150,41 @@
         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:190px">
+        <label class="form-label">Page image sent to the model</label>
+        <select id="s-ocr-page-image-format" class="form-input">
+          <option value="auto">Auto &mdash; pass scans through, else lossless PNG</option>
+          <option value="png">Always re-encode as PNG</option>
+          <option value="jpeg">JPEG (lossy, fastest)</option>
+        </select>
+        <span class="form-hint">Measured on A4 at 200&nbsp;DPI: rendering a page costs
+        21&ndash;46&nbsp;ms and <b>encoding it 86&ndash;353&nbsp;ms</b>, so most of what
+        looks like rasterisation is zlib. <b>Auto</b> sends a scanned page as the JPEG the
+        PDF already holds &mdash; no render, no encode, no resample (0.1&nbsp;ms, 147&nbsp;KB)
+        &mdash; and encodes everything else losslessly at zlib level&nbsp;1. <b>PNG</b>
+        always re-encodes, which is what a controlled comparison needs. <b>JPEG</b> is
+        eight times faster again but lossy.</span>
+      </div>
+      <div class="form-group" style="flex:1;min-width:120px">
+        <label class="form-label">JPEG quality</label>
+        <input type="number" id="s-ocr-page-image-quality" class="form-input" min="1" max="100" placeholder="95">
+        <span class="form-hint">Only used when the format above is JPEG.</span>
+      </div>
+      <div class="form-group" style="flex:1;min-width:150px">
+        <label class="form-label">Pass a scan through above (DPI)</label>
+        <input type="number" id="s-ocr-page-image-min-dpi" class="form-input" min="0" placeholder="150">
+        <span class="form-hint">A scan coarser than this is rendered at the configured DPI
+        instead, so the model still gets the pixel density the rest of the pipeline was
+        tuned for. A scan is also rendered when it is more than
+        <span id="s-ocr-page-image-factor-label">3</span>&times; the target, where
+        rendering down is cheaper than shipping it whole.</span>
+      </div>
+      <div class="form-group" style="flex:1;min-width:130px">
+        <label class="form-label">&hellip;and below this &times; DPI</label>
+        <input type="number" id="s-ocr-page-image-max-factor" class="form-input" min="1" step="0.5" placeholder="3">
+        <span class="form-hint">Upper bound for passing a scan through, as a multiple of
+        the configured DPI.</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">
@@ -2276,6 +2311,12 @@ async function loadSettings(){
     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-page-image-format').value = ocr.page_image_format ?? 'auto';
+    document.getElementById('s-ocr-page-image-quality').value = ocr.page_image_jpeg_quality ?? 95;
+    document.getElementById('s-ocr-page-image-min-dpi').value = ocr.page_image_min_dpi ?? 150;
+    document.getElementById('s-ocr-page-image-max-factor').value = ocr.page_image_max_dpi_factor ?? 3;
+    document.getElementById('s-ocr-page-image-factor-label').textContent =
+      String(ocr.page_image_max_dpi_factor ?? 3);
     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;
@@ -2583,6 +2624,10 @@ async function saveSettings(silent){
       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,
+      page_image_format: document.getElementById('s-ocr-page-image-format').value || 'auto',
+      page_image_jpeg_quality: parseInt(document.getElementById('s-ocr-page-image-quality').value) || 95,
+      page_image_min_dpi: parseFloat(document.getElementById('s-ocr-page-image-min-dpi').value) || 0,
+      page_image_max_dpi_factor: parseFloat(document.getElementById('s-ocr-page-image-max-factor').value) || 3,
       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

@@ -1071,6 +1071,34 @@ class OcrConfig:
     # 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 a page is handed to the model ---
+    # Measured on A4 at 200 DPI (3.4-3.9 Mpix): rendering a page costs 21-46 ms
+    # and ENCODING it 86-353 ms, so three quarters of what is called
+    # rasterisation is zlib, and on the surya path it was paid twice per page.
+    #
+    #   auto   (default) a page that already IS an image -- a scan held as one
+    #          full-page JPEG -- is sent to the model exactly as the PDF holds
+    #          it: no render, no encode, no resample (0.1 ms and 147 KB against
+    #          46 + 352 ms and 1846 KB). Everything else is encoded as PNG at
+    #          zlib level 1, which is lossless and 2.6x faster than PIL's
+    #          default level 6.
+    #   png    always re-encode. The one setting under which two runs of the
+    #          same document produce byte-identical requests whatever the PDF
+    #          held, which is what a controlled comparison needs.
+    #   jpeg   lossy, 8 ms and a third of the bytes. Whether it costs accuracy
+    #          is a question about a corpus rather than about a codec, so it is
+    #          opt-in and not a default.
+    page_image_format: str = "auto"
+    page_image_jpeg_quality: int = 95
+    # The bounds within which a scan may be passed through. Below the floor the
+    # embedded image is genuinely coarser than the configured DPI and rendering
+    # it up at least gives the model the pixel density the rest of the pipeline
+    # was tuned for; far above the target, rendering DOWN is cheaper than
+    # shipping the whole thing for the engine to scale. Expressed in real DPI
+    # rather than as a ratio because "is 150 DPI enough to read" is a question
+    # about documents, which an operator can answer.
+    page_image_min_dpi: float = 150.0
+    page_image_max_dpi_factor: float = 3.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
@@ -2078,6 +2106,10 @@ class ConfigManager:
                 "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_image_format": self.config.ocr.page_image_format,
+                "page_image_jpeg_quality": self.config.ocr.page_image_jpeg_quality,
+                "page_image_min_dpi": self.config.ocr.page_image_min_dpi,
+                "page_image_max_dpi_factor": self.config.ocr.page_image_max_dpi_factor,
                 "page_window": self.config.ocr.page_window,
                 "raster_workers": self.config.ocr.raster_workers,
                 "lang": self.config.ocr.lang,

codai/ocr/base.py

@@ -283,7 +283,8 @@ def ink_coverage(image, threshold: int = 200) -> float:
         return -1.0
 
 
-def load_pages(data: bytes, filename: str = "", content_type: str = "", dpi: int = 200) -> List["Any"]:
+def load_pages(data: bytes, filename: str = "", content_type: str = "",
+               dpi: int = 200, policy=None) -> List["Any"]:
     """Decode ``data`` (an image or a PDF) into a list of RGB ``PIL.Image`` pages.
 
     PDFs are rasterised at ``dpi`` via pypdfium2 (preferred) or pdf2image/poppler.
@@ -302,7 +303,7 @@ def load_pages(data: bytes, filename: str = "", content_type: str = "", dpi: int
         or data[:5] == b"%PDF-"
 
     if is_pdf:
-        return _rasterize_pdf(data, dpi)
+        return _rasterize_pdf(data, dpi, policy)
 
     try:
         img = Image.open(io.BytesIO(data))
@@ -484,7 +485,8 @@ def iter_pdf_page_pdfs(data: bytes):
 
 
 def render_pages(data: bytes, filename: str = "", content_type: str = "",
-                 dpi: int = 200, raster_workers: int = 0, chunk: int = 8):
+                 dpi: int = 200, raster_workers: int = 0, chunk: int = 8,
+                 policy=None):
     """Yield RGB ``PIL.Image`` pages ONE AT A TIME instead of returning them all.
 
     Same decoding rules as :func:`load_pages` — this is that function with the list
@@ -514,13 +516,14 @@ def render_pages(data: bytes, filename: str = "", content_type: str = "",
             try:
                 from codai.ocr.rasterize import iter_pdf_pages
                 pooled = iter_pdf_pages(data, dpi, chunk=chunk,
-                                        raster_workers=raster_workers)
+                                        raster_workers=raster_workers,
+                                        policy=policy)
             except Exception:
                 pooled = None
             if pooled is not None:
                 yield from pooled
                 return
-        yield from _render_pdf_pages(data, dpi)
+        yield from _render_pdf_pages(data, dpi, policy)
         return
 
     try:
@@ -538,7 +541,30 @@ def render_pages(data: bytes, filename: str = "", content_type: str = "",
         yield img.convert("RGB")
 
 
-def render_pdf_range(data: bytes, dpi: int, start: int, stop: int):
+def _page_or_render(page, dpi: float, scale: float, policy):
+    """One page as a PIL image: its own bytes when it may be, else rendered.
+
+    The in-process twin of the rasteriser worker's fast path. Both call the same
+    guards (``page_image.extract_embedded``) so the two never drift -- a page
+    passed through in one and rendered in the other would make a pool outage
+    change what the model is shown, not just how fast it arrives.
+    """
+    from codai.ocr.workers import page_image
+    got = None
+    if policy is not None:
+        got = policy.extract(page, dpi)
+    elif page_image.DEFAULT.passthrough:
+        got = page_image.DEFAULT.extract(page, dpi)
+    if got is not None:
+        import io as _io
+        from PIL import Image as _Image
+        mime, raw, _size = got
+        return page_image.attach(_Image.open(_io.BytesIO(raw)), mime, raw)
+    return page.render(scale=scale).to_pil().convert("RGB")
+
+
+def render_pdf_range(data: bytes, dpi: int, start: int, stop: int,
+                     policy=None):
     """Yield in-process RGB pages for ``[start, stop)`` of a PDF, under the lock.
 
     The fallback for a pooled read whose pool stopped serving midway: re-rendering
@@ -551,7 +577,7 @@ def render_pdf_range(data: bytes, dpi: int, start: int, stop: int):
         pdfium = None
     if pdfium is None:
         # No pypdfium2 at all: the whole-document fallback owns this case.
-        for i, img in enumerate(_rasterize_pdf(data, dpi)):
+        for i, img in enumerate(_rasterize_pdf(data, dpi, policy)):
             if start <= i < stop:
                 yield img
         return
@@ -565,13 +591,13 @@ def render_pdf_range(data: bytes, dpi: int, start: int, stop: int):
     for i in range(max(0, start), min(n, stop)):
         try:
             with _PDFIUM_LOCK:
-                img = pdf[i].render(scale=scale).to_pil().convert("RGB")
+                img = _page_or_render(pdf[i], dpi, scale, policy)
         except Exception as e:
             raise OcrError(f"failed to rasterise PDF (pypdfium2): {e}", status=400)
         yield img
 
 
-def _render_pdf_pages(data: bytes, dpi: int):
+def _render_pdf_pages(data: bytes, dpi: int, policy=None):
     """Yield one rasterised RGB page at a time (pypdfium2, else pdf2image)."""
     try:
         import pypdfium2 as pdfium
@@ -592,9 +618,7 @@ def _render_pdf_pages(data: bytes, dpi: int):
                 # and holding pdfium while an engine reads a page would serialise
                 # the whole subsystem behind one document.
                 with _PDFIUM_LOCK:
-                    page = pdf[i]
-                    bitmap = page.render(scale=scale)
-                    img = bitmap.to_pil().convert("RGB")
+                    img = _page_or_render(pdf[i], dpi, scale, policy)
             except Exception as e:
                 raise OcrError(f"failed to rasterise PDF (pypdfium2): {e}", status=400)
             yield img
@@ -602,11 +626,11 @@ def _render_pdf_pages(data: bytes, dpi: int):
 
     # pdf2image has no per-page API that avoids the whole document, so this path
     # keeps the old behaviour: correctness first, and pypdfium2 is the shipped dep.
-    for im in _rasterize_pdf(data, dpi):
+    for im in _rasterize_pdf(data, dpi, policy):
         yield im
 
 
-def _rasterize_pdf(data: bytes, dpi: int) -> List["Any"]:
+def _rasterize_pdf(data: bytes, dpi: int, policy=None) -> List["Any"]:
     """Rasterise a PDF to a list of RGB PIL pages (pypdfium2 first, then pdf2image)."""
     from PIL import Image  # noqa: F401  (imported for type parity / already validated)
 
@@ -623,9 +647,7 @@ def _rasterize_pdf(data: bytes, dpi: int) -> List["Any"]:
                 scale = max(0.5, dpi / 72.0)
                 pages = []
                 for i in range(len(pdf)):
-                    page = pdf[i]
-                    bitmap = page.render(scale=scale)
-                    pages.append(bitmap.to_pil().convert("RGB"))
+                    pages.append(_page_or_render(pdf[i], dpi, scale, policy))
             return pages
         except Exception as e:
             raise OcrError(f"failed to rasterise PDF (pypdfium2): {e}", status=400)

codai/ocr/chandra.py

@@ -204,18 +204,24 @@ def scale_to_fit(img, max_size=(3072, 2048), min_size=(1792, 28), grid_size: int
     return img.resize((new_width, new_height), resample=Image.Resampling.LANCZOS)
 
 
-def prepare_image(image) -> (str, tuple):
+def prepare_image(image, policy=None) -> (str, tuple):
     """``(data URI, (w, h))`` for the image as the MODEL will see it.
 
     The size is returned because the boxes come back normalised against it, and
     de-normalising them against the pre-scaling size would put every box in the wrong
     place on any page that was scaled.
+
+    ``scale_to_fit`` FIRST, and ``convert`` only if the mode needs it: both
+    return a new object, which is what drops a scanned page's own JPEG bytes --
+    exactly when they stop being the image the model will see. A page that needs
+    neither keeps them and is sent without being encoded at all.
     """
-    img = scale_to_fit(image.convert("RGB"))
-    buf = io.BytesIO()
-    img.save(buf, format="PNG")
-    uri = "data:image/png;base64," + base64.b64encode(buf.getvalue()).decode("ascii")
-    return uri, img.size
+    from codai.ocr.workers import page_image
+    img = scale_to_fit(image)
+    if img.mode not in ("RGB", "L"):
+        img = img.convert("RGB")
+    mime, raw = (policy or page_image.DEFAULT).encode(img)
+    return page_image.data_uri(mime, raw), img.size
 
 
 # ---------------------------------------------------------------------------
@@ -579,7 +585,9 @@ class ChandraEngine(OcrEngine):
 
     def recognize_image(self, image) -> OcrPage:
         src_w, src_h = image.size
-        data_uri, (w, h) = prepare_image(image)
+        from codai.ocr.workers import page_image
+        data_uri, (w, h) = prepare_image(
+            image, page_image.Policy.from_cfg(self.cfg))
         content, meta, conf = self._ask_with_retries(data_uri)
         page = parse_page(
             content, w, h,

codai/ocr/manager.py

@@ -2186,9 +2186,11 @@ class OcrManager:
         # `chunk` is the page window: a pooled read borrows a worker per chunk,
         # so matching the two means the document is re-opened once per window
         # rather than once per few pages.
+        from codai.ocr.workers import page_image
         stream = render_pages(data, filename=filename, content_type=content_type,
                               dpi=dpi, raster_workers=self._raster_workers(cfg),
-                              chunk=window)
+                              chunk=window,
+                              policy=page_image.Policy.from_cfg(cfg))
         pages_all: List[OcrPage] = []
         blank_all: list = []
         stamps_all: list = []

codai/ocr/olmocr.py

@@ -153,15 +153,22 @@ def parse_front_matter(content: str) -> (dict, str):
 # image prep
 # ---------------------------------------------------------------------------
 
-def prepare_image(image, longest_side: int = DEFAULT_LONGEST_SIDE) -> str:
-    """Return a ``data:image/png;base64,…`` URI of ``image`` scaled to ``longest_side``.
+def prepare_image(image, longest_side: int = DEFAULT_LONGEST_SIDE,
+                  policy=None) -> str:
+    """Return a ``data:<mime>;base64,…`` URI of ``image`` scaled to ``longest_side``.
 
     Only ever scales DOWN past the target: upsampling a small crop buys nothing and costs
     vision tokens.
+
+    A page that needs no scaling and arrived already encoded is sent exactly as
+    it arrived -- hence the resize BEFORE any ``convert``, so the carried bytes
+    are dropped by the one operation that invalidates them rather than by a
+    mode conversion that was a no-op anyway.
     """
     from PIL import Image
+    from codai.ocr.workers import page_image
 
-    img = image.convert("RGB")
+    img = image
     target = max(64, int(longest_side or DEFAULT_LONGEST_SIDE))
     w, h = img.size
     longest = max(w, h)
@@ -169,9 +176,10 @@ def prepare_image(image, longest_side: int = DEFAULT_LONGEST_SIDE) -> str:
         scale = target / float(longest)
         img = img.resize((max(1, int(round(w * scale))), max(1, int(round(h * scale)))),
                          Image.LANCZOS)
-    buf = io.BytesIO()
-    img.save(buf, format="PNG")
-    return "data:image/png;base64," + base64.b64encode(buf.getvalue()).decode("ascii")
+    if img.mode not in ("RGB", "L"):
+        img = img.convert("RGB")
+    mime, raw = (policy or page_image.DEFAULT).encode(img)
+    return page_image.data_uri(mime, raw)
 
 
 # ---------------------------------------------------------------------------
@@ -352,9 +360,11 @@ class OlmOcrEngine(OcrEngine):
         and olmOCR's front matter made that worse: a reply cut off before the
         body closes parses to metadata and no text at all.
         """
+        from codai.ocr.workers import page_image
         data_uri = prepare_image(
             image, int(getattr(self.cfg, "olmocr_longest_side", DEFAULT_LONGEST_SIDE) or
-                       DEFAULT_LONGEST_SIDE))
+                       DEFAULT_LONGEST_SIDE),
+            page_image.Policy.from_cfg(self.cfg))
         messages = [{
             "role": "user",
             "content": [

codai/ocr/rasterize.py

@@ -41,6 +41,7 @@ import threading
 from typing import Optional
 
 from codai.ocr.base import OcrError, render_pdf_range
+from codai.ocr.workers import page_image
 
 _WORKER = os.path.join(os.path.dirname(__file__), "workers", "raster_worker.py")
 
@@ -97,8 +98,9 @@ class _Worker:
         self.proc.stdin.flush()
 
     # -- commands --------------------------------------------------------- #
-    def open(self, data: bytes, dpi: int) -> "tuple[int, int]":
-        self._send({"cmd": "open", "dpi": int(dpi), "nbytes": len(data)}, data)
+    def open(self, data: bytes, dpi: int, policy=None) -> "tuple[int, int]":
+        self._send({"cmd": "open", "dpi": int(dpi), "nbytes": len(data),
+                    "page_image": (policy or page_image.DEFAULT).as_dict()}, data)
         r = self._reply()
         if not r.get("ok"):
             # A bad document is the document's problem, not the worker's: raise
@@ -118,6 +120,18 @@ class _Worker:
         out = []
         for m in metas:
             raw = self._read_exactly(int(m["nbytes"]))
+            mime = m.get("mime")
+            if mime:
+                # The worker never rendered this page: these are the bytes the
+                # PDF itself holds. Image.open is lazy, so the pixels are only
+                # decoded if something downstream actually looks at them -- and
+                # the bytes ride along so the engine can send them onwards
+                # without an encode. Carried, not trusted: any transform drops
+                # them (see page_image.attach).
+                import io as _io
+                img = Image.open(_io.BytesIO(raw))
+                out.append(page_image.attach(img, mime, raw))
+                continue
             out.append(Image.frombytes("RGB", (int(m["w"]), int(m["h"])), raw))
         return out
 
@@ -269,7 +283,7 @@ def reset_pool() -> None:
         _pool_size = 0
 
 
-def probe_pdf(data: bytes, dpi: int, raster_workers: int = 0):
+def probe_pdf(data: bytes, dpi: int, raster_workers: int = 0, policy=None):
     """``(pages, retained_bytes)`` for a PDF, in a worker when there is one.
 
     Returns None when the pool could not serve, so the caller uses its own
@@ -282,7 +296,7 @@ def probe_pdf(data: bytes, dpi: int, raster_workers: int = 0):
         with pool.borrow() as w:
             if w is None:
                 return None
-            return w.open(data, dpi)
+            return w.open(data, dpi, policy)
     except OcrError:
         raise                       # a bad document, not a pool problem
     except Exception as exc:
@@ -292,7 +306,7 @@ def probe_pdf(data: bytes, dpi: int, raster_workers: int = 0):
 
 
 def iter_pdf_pages(data: bytes, dpi: int, chunk: int = 8,
-                   raster_workers: int = 0):
+                   raster_workers: int = 0, policy=None):
     """Yield RGB pages from worker processes, or None if the pool cannot serve.
 
     A generator that yields nothing and returns is indistinguishable from an empty
@@ -317,7 +331,7 @@ def iter_pdf_pages(data: bytes, dpi: int, chunk: int = 8,
     if w is None:
         return None
     try:
-        n, _est = w.open(data, dpi)
+        n, _est = w.open(data, dpi, policy)
     finally:
         pool.release(w)
 
@@ -329,11 +343,11 @@ def iter_pdf_pages(data: bytes, dpi: int, chunk: int = 8,
             if worker is None:
                 # The pool stopped serving mid-document. Rasterise the rest here
                 # rather than fail a document that is already half read.
-                for img in render_pdf_range(data, dpi, start, n):
+                for img in render_pdf_range(data, dpi, start, n, policy):
                     yield img
                 return
             try:
-                worker.open(data, dpi)
+                worker.open(data, dpi, policy)
                 pages = worker.render(start, count)
             finally:
                 pool.release(worker)

codai/ocr/subprocess_engine.py

@@ -288,14 +288,19 @@ class SubprocessOcrEngine(OcrEngine):
 
     def recognize_image(self, image) -> OcrPage:
         import base64
-        import io
-        buf = io.BytesIO()
-        image.convert("RGB").save(buf, "PNG")
-        b64 = base64.b64encode(buf.getvalue()).decode("ascii")
+        from codai.ocr.workers import page_image
+        # The page goes down the pipe as it already is whenever it can: a
+        # scanned page arrives as the JPEG the PDF held, and re-encoding it here
+        # cost 323 ms of zlib and 1.8 MB of pipe per page -- then surya's own
+        # client did it a SECOND time to build the request. `image.convert`
+        # used to be unconditional, which also quietly dropped the carried
+        # bytes on every page, since convert returns a copy.
+        mime, raw = page_image.Policy.from_cfg(self.cfg).encode(image)
+        b64 = base64.b64encode(raw).decode("ascii")
         # Per-request overrides travel WITH the page, not with the load: a pool
         # is built once and then serves every caller, so anything set at load
         # time is the same for all of them.
-        msg = {"cmd": "ocr", "image": b64}
+        msg = {"cmd": "ocr", "image": b64, "mime": mime}
         opts = page_options()
         if opts:
             msg["opts"] = opts

codai/ocr/surya.py

@@ -92,7 +92,14 @@ class SuryaEngine(SubprocessOcrEngine):
                 "repeat_window": max(0, int(getattr(
                     self.cfg, "surya_repeat_window", 0) or 0)),
                 "repeat_scaling": max(0.0, float(getattr(
-                    self.cfg, "surya_repeat_scaling", 0.0) or 0.0))}
+                    self.cfg, "surya_repeat_scaling", 0.0) or 0.0)),
+                # How the WORKER encodes a page for the model, when it has to
+                # encode one at all. Sent at load because the surya function it
+                # replaces is module-level and must be in place before page one.
+                "page_image_format": str(getattr(
+                    self.cfg, "page_image_format", "auto") or "auto"),
+                "page_image_jpeg_quality": max(1, min(100, int(getattr(
+                    self.cfg, "page_image_jpeg_quality", 95) or 95)))}
 
     #: Modes where the recognition VLM lives in a model server we only talk to.
     SERVED_MODES = ("model", "vllm", "llamacpp", "server")

codai/ocr/workers/ocr_worker.py

@@ -11,7 +11,13 @@
 # Protocol: newline-delimited JSON on stdin/stdout.
 #   startup            → worker prints {"ready": true}
 #   {"cmd":"load","opts":{...}}   → {"ok":true} | {"ok":false,"error":..,"status":..}
-#   {"cmd":"ocr","image":"<b64 png>","opts":{...}} → {"ok":true,"page":{...}} | {...}
+#   {"cmd":"ocr","image":"<b64>","mime":"image/png","opts":{...}}
+#                      → {"ok":true,"page":{...}} | {...}
+#       `mime` is what `image` actually is ("image/png" when absent). It matters
+#       because a page that arrived already encoded can be sent ONWARDS to the
+#       model without being encoded again: surya's own client PNGs every page
+#       (86-353 ms on A4 at 200 DPI), and with the pipe doing the same the
+#       surya path paid zlib twice per page against ~5 s on the GPU.
 #       `opts` are PER-REQUEST overrides for this page only (max_tokens,
 #       max_continuations). Applied thread-locally, so lanes cannot overwrite
 #       each other's budgets. Absent = the values sent at load.
@@ -327,6 +333,131 @@ def _no_resume(reason: str) -> None:
         pass
 
 
+#: What a page's bytes were when they arrived, attached to the PIL image.
+#:
+#: Deliberately the same shape as ``codai/ocr/workers/page_image.py`` writes,
+#: and deliberately NOT imported from it: this file runs inside the engine's own
+#: virtualenv, where coderai is not importable at all. The convention is the
+#: protocol, so it is repeated here rather than shared.
+_PAGE_BYTES = "_coderai_page_bytes"
+
+#: ``(format, quality)`` for pages this process must encode itself. Module-level
+#: because the surya function it feeds is.
+_IMAGE_FORMAT = ["auto", 95]
+
+
+def _carry(img, mime, raw):
+    """Remember that ``raw`` already encodes ``img`` (see page_image.attach)."""
+    try:
+        setattr(img, _PAGE_BYTES, (mime, raw, tuple(img.size)))
+    except Exception:
+        pass
+    return img
+
+
+def _carried(img):
+    """``(mime, raw)`` when this image IS those bytes, else None.
+
+    The size check is what makes this safe: surya scales a page to fit the
+    model, and a scaled page is a new object -- but an in-place transform would
+    leave the attribute pointing at bytes of the wrong size, which the model
+    would read as a perfectly plausible wrong page.
+    """
+    got = getattr(img, _PAGE_BYTES, None)
+    if not got:
+        return None
+    try:
+        mime, raw, size = got
+    except Exception:
+        return None
+    if not raw or tuple(img.size) != tuple(size):
+        return None
+    return mime, raw
+
+
+def _encode_page(img):
+    """``(mime, bytes)`` for one page: its own bytes, else the cheap encoding.
+
+    PIL's PNG default is ``compress_level=6`` (323 ms on a 3.45 Mpix page);
+    level 1 is lossless too and takes 123 ms. JPEG is 8 ms but lossy, so it is
+    only used when the operator asked for it.
+    """
+    fmt, quality = _IMAGE_FORMAT[0], _IMAGE_FORMAT[1]
+    if fmt != "png":
+        got = _carried(img)
+        if got is not None:
+            return got
+    buf = io.BytesIO()
+    if fmt == "jpeg":
+        if img.mode not in ("RGB", "L"):
+            img = img.convert("RGB")
+        img.save(buf, format="JPEG", quality=int(quality))
+        return "image/jpeg", buf.getvalue()
+    if img.mode not in ("RGB", "L"):
+        img = img.convert("RGB")
+    img.save(buf, format="PNG", compress_level=1)
+    return "image/png", buf.getvalue()
+
+
+def _install_image_encoder():
+    """Make surya's client stop re-encoding pages it was handed encoded.
+
+    ``surya/inference/backends/openai_client.py`` builds every request with
+
+        buf = io.BytesIO(); image.save(buf, format="PNG")        # 86-353 ms
+        ... f"data:image/png;base64,{image_b64}"                 # hard-coded
+
+    so both functions are replaced: the encoder, to reuse the page's own bytes
+    (or at least a cheaper encoding), and the message builder, because the mime
+    type in that URI is a literal and a JPEG announced as a PNG is the kind of
+    thing that works until the day a server believes it.
+
+    Patched once per process, failure is silent BY DESIGN: a surya release that
+    renames either function must cost throughput, never pages. Verified by
+    reading the installed module, not assumed -- if ``_build_messages`` no
+    longer calls the encoder by name, the encoder patch alone still works.
+    """
+    try:
+        from surya.inference.backends import openai_client as _oc
+    except Exception as e:
+        print(f"[surya] page encoder unavailable: {e}", file=sys.stderr)
+        return
+    if getattr(_oc, "_coderai_image_patched", False):
+        return
+    if getattr(_oc, "encode_image_b64", None) is None:
+        print("[surya] page encoder: no encode_image_b64 to replace",
+              file=sys.stderr)
+        return
+
+    def _encode_b64(image):
+        mime, raw = _encode_page(image)
+        _LAST_MIME.value = mime
+        return base64.b64encode(raw).decode("ascii")
+
+    def _build_messages(image, prompt):
+        b64 = _encode_b64(image)
+        mime = getattr(_LAST_MIME, "value", "image/png")
+        return [{"role": "user",
+                 "content": [{"type": "image_url",
+                              "image_url": {"url": f"data:{mime};base64,{b64}"}},
+                             {"type": "text", "text": prompt}]}]
+
+    _oc.encode_image_b64 = _encode_b64
+    if getattr(_oc, "_build_messages", None) is not None:
+        _oc._build_messages = _build_messages
+    _oc._coderai_image_patched = True
+
+
+#: The mime of the last page this THREAD encoded, so the message builder can
+#: declare what it is actually sending. Per-thread for the same reason the token
+#: trace is: lanes put several pages in one process.
+class _LastMime(threading.local):
+    value = "image/png"
+
+
+_LAST_MIME = _LastMime()
+
+
 def _note_loop(raw: str) -> None:
     """Record that this page's generation ended still repeating itself.
 
@@ -1199,6 +1330,16 @@ class SuryaWorker:
         _RETRY_BUDGET[0] = None if budget is None else max(0, int(budget))
         #: 0 leaves surya's own 2048 in place.
         _MAX_TOKENS[0] = max(0, int(opts.get("max_tokens", 0) or 0))
+        #: How a page is encoded for the model, when this process must encode
+        #: it at all. Module cell, like the rest: the surya function it feeds is
+        #: module-level.
+        fmt = str(opts.get("page_image_format", "auto") or "auto").strip().lower()
+        _IMAGE_FORMAT[0] = fmt if fmt in ("auto", "png", "jpeg") else "auto"
+        try:
+            _IMAGE_FORMAT[1] = max(1, min(100, int(
+                opts.get("page_image_jpeg_quality", 95) or 95)))
+        except (TypeError, ValueError):
+            _IMAGE_FORMAT[1] = 95
         #: Resumptions allowed for one page. 0 = return a cut-off page partial.
         _CONTINUE_BUDGET[0] = max(0, int(opts.get("max_continuations", 2) or 0))
         #: 0 asks the server for its own `max_model_len`. Set it when the server
@@ -1249,6 +1390,7 @@ class SuryaWorker:
         self._det = DetectionPredictor()
         self._mode = "predictor"
         self._install_trace_hook()
+        _install_image_encoder()
         return
 
     def _install_trace_hook(self):
@@ -1730,6 +1872,7 @@ def _serve_page(lanes: "_Lanes", msg: dict, engine: str) -> None:
         from PIL import Image
         data = base64.b64decode(msg["image"])
         img = Image.open(io.BytesIO(data)); img.load()
+        _carry(img, msg.get("mime") or "image/png", data)
         # On THIS thread, before the page is read. _trace_begin() inside ocr()
         # resets the page's trace but not the options, so they survive it -- and
         # being thread-local they cannot leak into another lane's page.
@@ -1784,6 +1927,7 @@ def main():
                     from PIL import Image
                     data = base64.b64decode(msg["image"])
                     img = Image.open(io.BytesIO(data)); img.load()
+                    _carry(img, msg.get("mime") or "image/png", data)
                     _trace_opts(msg.get("opts"))
                     _emit({"ok": True, "page": worker.ocr(img)})
                 except Exception as e:

codai/ocr/workers/raster_worker.py

@@ -21,23 +21,46 @@ Protocol, newline-delimited JSON with raw binary payloads, on stdin/stdout:
     <- {"ok": true, "pages": [{"w": 2066, "h": 2924, "nbytes": 18122952}, ...]}
        + <concatenated raw RGB bytes, in that order>
 
+A page may instead come back ALREADY ENCODED, with a "mime" on its meta:
+
+    <- {"ok": true, "pages": [{"w": 1268, "h": 1752, "nbytes": 150740,
+                               "mime": "image/jpeg"}]}
+       + <the JPEG the PDF itself holds, untouched>
+
+which is what a scanned page is: one full-page DCTDecode image that needs no
+render and no encode (0.1 ms and 147 KB against 46 ms, 352 ms and 1846 KB). The
+guards live in page_image.extract_embedded; `open` carries the policy.
+
     -> {"cmd": "close"}                     release the document, keep the process
     <- {"ok": true}
 
     -> {"cmd": "shutdown"}
     <- {"ok": true}                         then exit
 
-Raw RGB rather than PNG: the parent rebuilds the page with Image.frombytes, and
-a page that is about to be read by an OCR engine must not pass through a lossy
-codec, while PNG would cost more CPU than the pipe it saves.
+Raw RGB rather than PNG for a RENDERED page: the parent rebuilds it with
+Image.frombytes, and a page about to be read by an OCR engine must not pass
+through a lossy codec, while PNG would cost more CPU than the pipe it saves. A
+page that was never rendered is a different matter -- its bytes are already a
+JPEG, and re-rendering them to ship them raw would be work in both directions.
 
 Errors answer {"ok": false, "error": str, "status": int} and leave the process
 usable -- one undecodable document must not cost the pool a worker.
 """
 
 import json
+import os
 import sys
 
+# This script runs with -I, so its own directory is NOT on sys.path. The policy
+# that decides when a page may be passed through has to be shared with the
+# parent -- duplicating those guards is how the two sides drift -- and it is the
+# one import allowed here beyond pypdfium2 and PIL: stdlib + PIL only, no codai.
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+try:
+    import page_image
+except Exception:                      # pragma: no cover - keep rendering
+    page_image = None
+
 
 class _Doc:
     """The one document this worker holds open."""
@@ -46,6 +69,8 @@ class _Doc:
         self.pdf = None
         self.scale = 1.0
         self.n = 0
+        self.dpi = 200
+        self.policy = None
 
     def close(self):
         pdf, self.pdf = self.pdf, None
@@ -112,6 +137,13 @@ def main() -> int:
                 _reply({"ok": False, "error": str(e), "status": 500})
                 return 1          # the frame is lost; the parent must replace us
             dpi = int(msg.get("dpi") or 200)
+            doc.dpi = dpi
+            doc.policy = None
+            if page_image is not None and msg.get("page_image") is not None:
+                try:
+                    doc.policy = page_image.Policy.from_dict(msg.get("page_image"))
+                except Exception:
+                    doc.policy = None
             try:
                 doc.pdf = pdfium.PdfDocument(data)
                 doc.scale = max(0.5, dpi / 72.0)
@@ -140,6 +172,19 @@ def main() -> int:
             try:
                 for i in range(start, min(doc.n, start + count)):
                     page = doc.pdf[i]
+                    # A page that already IS an image is sent as it is: no
+                    # render, no encode, and a pipe carrying 147 KB instead of
+                    # 11.6 MB. None from here means "render it", always.
+                    got = None
+                    if doc.policy is not None:
+                        got = doc.policy.extract(page, doc.dpi)
+                    if got is not None:
+                        mime, raw, (w, h) = got
+                        metas.append({"w": w, "h": h, "nbytes": len(raw),
+                                      "mime": mime})
+                        blobs.append(raw)
+                        del page
+                        continue
                     bitmap = page.render(scale=doc.scale)
                     img = bitmap.to_pil().convert("RGB")
                     raw = img.tobytes()

tests/test_ocr_raster_pool.py

@@ -235,7 +235,35 @@ def test_the_worker_does_not_import_the_ocr_package():
         elif isinstance(node, ast.ImportFrom) and node.module:
             names.add(node.module.split(".")[0])
     assert "codai" not in names, f"worker imports the package: {sorted(names)}"
-    assert names <= {"json", "sys", "pypdfium2", "PIL"}, sorted(names)
+    # `page_image` is the one local module it loads, by path, because the guards
+    # deciding when a page may be passed through unrendered must be the SAME
+    # ones the parent applies -- two copies of those would drift, and a drifted
+    # guard changes what the model is shown. It is held to the same rule as the
+    # worker itself, below.
+    assert names <= {"json", "os", "sys", "pypdfium2", "PIL", "page_image"}, \
+        sorted(names)
+
+
+def test_the_policy_module_is_as_light_as_the_worker():
+    """The worker may import it, so it carries the worker's constraint too:
+    stdlib and PIL, nothing of coderai, and no top-level pypdfium2 (which the
+    engine venvs do not have)."""
+    import ast
+    src = (ROOT / "codai/ocr/workers/page_image.py").read_text()
+    names = set()
+    for node in ast.walk(ast.parse(src)):
+        if isinstance(node, ast.Import):
+            names.update(a.name.split(".")[0] for a in node.names)
+        elif isinstance(node, ast.ImportFrom) and node.module:
+            names.add(node.module.split(".")[0])
+    assert "codai" not in names, sorted(names)
+    top = set()
+    for node in ast.parse(src).body:
+        if isinstance(node, ast.Import):
+            top.update(a.name.split(".")[0] for a in node.names)
+        elif isinstance(node, ast.ImportFrom) and node.module:
+            top.add(node.module.split(".")[0])
+    assert top == {"io"}, sorted(top)
 
 
 def test_the_worker_is_run_with_isolated_mode():

tests/test_ocr_resident_budget.py

@@ -403,9 +403,10 @@ def _manager(monkeypatch, n_pages, cfg=None, engine_for=None):
     trace = {"pulled": 0, "windows": [], "reserved": []}
 
     def _fake_render(data, filename="", content_type="", dpi=200,
-                     raster_workers=0, chunk=8):
+                     raster_workers=0, chunk=8, policy=None):
         trace["chunk"] = chunk
         trace["raster_workers"] = raster_workers
+        trace["policy"] = policy
         for i in range(n_pages):
             trace["pulled"] += 1
             yield f"page-{i}"                     # a stand-in for a PIL image
@@ -741,6 +742,19 @@ def test_the_pool_chunk_matches_the_page_window(monkeypatch):
     assert trace["chunk"] == 4, trace.get("chunk")      # _Cfg.page_window
 
 
+def test_the_page_image_policy_reaches_the_renderer(monkeypatch):
+    """It decides whether a scanned page is rendered at all, so a read that
+    does not carry it would silently render every scan."""
+    class _C(_Cfg):
+        page_image_format = "jpeg"
+        page_image_jpeg_quality = 70
+    m, trace = _manager(monkeypatch, n_pages=2, cfg=_C())
+    asyncio.run(m.ocr_document(b"%PDF-x", filename="x.pdf"))
+    assert trace["policy"] is not None
+    assert trace["policy"].fmt == "jpeg"
+    assert trace["policy"].jpeg_quality == 70
+
+
 def test_the_raster_worker_setting_reaches_the_renderer(monkeypatch):
     class _C(_Cfg):
         raster_workers = 3

tests/test_ocr_worker_lanes.py

@@ -45,12 +45,20 @@ def _png_bytes():
 
 
 class _Img:
-    """The minimum of a PIL image the engine touches before sending it."""
+    """The minimum of a PIL image the engine touches before sending it.
+
+    ``save`` takes ``format`` as a keyword because that is how the encoder
+    writes it now -- it also passes ``compress_level``, which PIL accepts and
+    this stand-in ignores.
+    """
+
+    mode = "RGB"
+    size = (1, 1)
 
     def convert(self, mode):
         return self
 
-    def save(self, buf, fmt):
+    def save(self, buf, format=None, **kw):
         buf.write(_png_bytes())