Image Hopper router v2-fast: faster serving and default-route image cap (same weights and calibration)

Dockerfile

@@ -9,4 +9,4 @@ COPY requirements.lock requirements.lock.sha256 /app/
 RUN sha256sum -c requirements.lock.sha256 && python -m pip install --no-cache-dir -r requirements.lock
 COPY . /app
 EXPOSE 8080
-ENTRYPOINT ["python", "-m", "open_decisions.image_jev.release.server", "--host", "0.0.0.0", "--port", "8080", "--offline", "--adapter", "adapter"]
+ENTRYPOINT ["python", "-m", "open_decisions.image_jev.release.server", "--host", "0.0.0.0", "--port", "8080", "--offline", "--adapter", "adapter", "--serving-options", "lean_lora,uint8_pixels,device_readout,image_token_cap_default=768"]

README.md

@@ -24,7 +24,8 @@ in one forward pass and without generating text. It is built on `Qwen/Qwen3.5-9B
   for those letters at the last prompt position.
 - **Calibration.** Logits are divided by a temperature chosen by route and answer type (fitted by maximum likelihood
   with shrinkage toward a global temperature). Calibration never changes which option ranks first.
-- **Images** use the base processor's native resolution route; up to four images per request.
+- **Images** use the base processor's native resolution route; on the default route each image is capped at 768
+  image tokens (screen/geometry questions keep full resolution); up to four images per request.
 
 | Route / answer type | Temperature |
 |---|---:|
@@ -58,10 +59,16 @@ checked against the benchmark's reference images for near-duplicates.
 The default-route temperatures were fitted on a held-out calibration split of our own data; the screen/geometry
 temperatures on held-out real screenshots and geometry diagrams that were never used for training.
 
+## Serving
+
+The Dockerfile starts the server with `--serving-options lean_lora,uint8_pixels,device_readout,image_token_cap_default=768`.
+The first three options change only how the computation runs, not the outputs; the image cap on the default route
+reduces cost per decision.
+
 ## Usage
 
     pip install -r requirements.txt
-    python -m open_decisions.image_jev.release.server --offline --adapter adapter --port 8080
+    python -m open_decisions.image_jev.release.server --offline --adapter adapter --port 8080 --serving-options "lean_lora,uint8_pixels,device_readout,image_token_cap_default=768"
 
 `POST /v1/systemone`:
 

RELEASE-MANIFEST.json

@@ -15,14 +15,14 @@
   },
   "code": {
     "dirty": false,
-    "git_commit": "b94f0c4507c2463f6569974cfbb36a05ccdbb764"
+    "git_commit": "78124a4fcb2123b8f5b3abb08d5751155c7d44be"
   },
   "files": {
     "BASE-MODEL-LICENSE": "c71d239df91726fc519c6eb72d318ec65820627232b2f796219e87dcf35d0ab4",
-    "Dockerfile": "904c4630d2b50f577e3e6cf0e9d71f25c06e4fb64ef1a41238b299ba0c8a5d7a",
+    "Dockerfile": "46c5a925edee39410700d801ec0156904e8b5808373d5c77a3600f45bce29691",
     "LICENSE": "c71d239df91726fc519c6eb72d318ec65820627232b2f796219e87dcf35d0ab4",
     "NOTICE": "1d5e63ea05ed3c846593f1f4185b4355a25b2b25682bccac24995e094141b6c7",
-    "README.md": "2b1b313bcf80dff7aaeea5904c82d3adec8e0c8079a42a28093f7538c794b8b9",
+    "README.md": "8486ae6e58abfeeed8aa515bb87277c88a36ba1ee7ea14c139cc9614a4f8481a",
     "adapter/README.md": "bb29528044c9409e183ce924e4ea6f29f2b705d749287781b8ff2777b5209a55",
     "adapter/adapter_config.json": "4f6e55c1d151070a6e8444feff44bdcfbe4b87e14c3fde855f69159066162487",
     "adapter/adapter_model.safetensors": "6201fcb1c08144e465a03fd6f5355c4394b3bffaad659440fa64063dd4f71509",
@@ -31,12 +31,13 @@
     "open_decisions/__init__.py": "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855",
     "open_decisions/image_jev/__init__.py": "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855",
     "open_decisions/image_jev/release/__init__.py": "4a29d6b23b6021c0cc27d3fa5163ebbede7ab531b1a13814dfddf1fedcd71af4",
-    "open_decisions/image_jev/release/predictor.py": "b4a469714f2cfb5b6cf8551975dc76e2b6139b6dae61b1ebcc4f31335dddb407",
-    "open_decisions/image_jev/release/server.py": "3815ab07ca929b9e226b0a73521b2dc6cc6c426da1d0d8942ab9e372163f3ca9",
-    "open_decisions/image_jev/release/smoke.py": "80c6ed250089724a332ea0341ad6324757b19a5407d219f539d28bcdf7864ea4",
+    "open_decisions/image_jev/release/fastpath.py": "4ad3e5e869e80393b30fd1fd90a377cae244a25ceca249c8aeb5eb0d44d61236",
+    "open_decisions/image_jev/release/predictor.py": "bbc178c08fb2b9a92684d759968b8792391a87ff21b804c3b723c68af61e76a3",
+    "open_decisions/image_jev/release/server.py": "bb994fe1abe7d29a425c4357eb80973565d6105130307f04597e8803ee97c168",
+    "open_decisions/image_jev/release/smoke.py": "773c339d4393d12953264918bfd3decfb46a67cee06feb2e61d29611c0313e3b",
     "open_decisions/image_jev/router.py": "e99db6dde898e7f7be4babcec2f4e545eedff51b04db8c7fe98e47e884a27c47",
     "open_decisions/image_jev/speed.py": "5671a56aa16d4647dca8f3c360e83f98747641d7722ae184110806375448c5cb",
-    "open_decisions/image_jev/vision.py": "f9e988cae5184b09b0ebacec7f9baeb3db06e971862c18229dea1ee5034e2be2",
+    "open_decisions/image_jev/vision.py": "de4ed892c0513ec86f954514476560cf9b20b7dd1d0e0afca548fcdf576aa882",
     "open_decisions/scoring/__init__.py": "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855",
     "open_decisions/scoring/prompt.py": "131916c014979500e5d2daceb94817a8942de8148966d3309c8ecbc8547027ee",
     "requirements.lock": "974e3a19773d67b4de84842ffa4a3a537e2886240cdfb97d94216159804aa7e2",
@@ -44,11 +45,12 @@
     "requirements.txt": "321bb81a1bd43b4858cf724d22944ac3e5e93504a9bf231be88b6752b437a63c"
   },
   "image_route": "native",
-  "manifest_sha256": "1403eee576ec886cf6c3bcd2dbb02761fd9cd9200af118d2fff05da6eb4ecbd3",
+  "manifest_sha256": "be3f472213dffc9678155bdd91480bb87f63edd7bebcb8737c960646a1ed23ff",
   "measurement_artifact": null,
   "name": "HopitAI/image-hopper",
   "scorer": "one-prefill-option-letter",
-  "status": "owner-review",
+  "serving_options": "lean_lora,uint8_pixels,device_readout,image_token_cap_default=768",
+  "status": "DRAFT - not published; public text needs owner approval",
   "system_kind": "routed",
   "version": "image-jev-release/v1"
 }

open_decisions/image_jev/release/fastpath.py

@@ -0,0 +1,620 @@
+"""Opt-in serving optimisations for the Image Hopper release server (cost audit, step 1).
+
+Every option is off by default, so the published path is unchanged unless the server is started
+with ``--serving-options``.  Options are either
+
+* **exact**: the served function is bit-identical by construction (same operations in the same
+  order, only Python overhead, copies or host transfers removed); or
+* **tolerance-bound**: numerically close but not bit-identical.  The cost audit registers the
+  shipping rule before measuring: argmax identical on every audit item and max |dp| <= 1e-3.
+
+Options (``OPTIONS`` holds the one-line descriptions):
+
+``lean_lora``      exact.  PEFT's LoRA wrappers are replaced by plain modules that call the
+                   original base layer, plus the screen_geometry LoRA term computed with PEFT's
+                   exact operations (input cast to the adapter dtype, B(A(x)) * scaling, add,
+                   cast back) only while that route is active.  Route change flips one flag.
+``lora_bf16``      tolerance.  As lean_lora, LoRA factors held in the base dtype (bf16).
+``merged_views``   tolerance.  The screen_geometry route reads a second, merged bf16 copy of each
+                   targeted weight (W + s*B@A rounded once); the default route reads W.
+``uint8_pixels``   exact.  The processor resizes and patchifies as before but returns uint8
+                   patches; they are uploaded from pinned memory and normalised on the device with
+                   the processor's own fused mean/std arithmetic (elementwise, so identical).
+``gpu_preprocess`` tolerance.  uint8_pixels plus the resize itself on the device.
+``device_readout`` exact.  Option rows gathered with a cached device index tensor; one host copy.
+``cuda_graphs``    tolerance.  The language model (and the last-position vocabulary row) is
+                   replayed from CUDA graphs recorded at padded lengths, one set per route.
+                   Right padding after the decision position cannot change a causal model's
+                   output there; kernels chosen for the padded shape can round differently.
+``sdpa_flash``     tolerance.  Scaled-dot-product attention restricted to the flash kernel.
+``compile_text``   tolerance.  ``torch.compile(dynamic=True)`` on the language model.
+
+One parameterised option changes outputs by design (it is neither exact nor tolerance-bound):
+
+``image_token_cap_default=N`` output-changing.  As below, but only for questions whose
+                   router-rules/v1 text route is ``default``; ``screen_geometry`` images stay
+                   native.  The two caps are alternatives.
+``image_token_cap=N`` output-changing.  The native route's processor max pixels drop to
+                   ``N * 32 * 32`` so every image keeps at most N merged image tokens
+                   (``vision.native_image_tokens``); N in [256, 16384].  Weights, router and
+                   calibration are unchanged.  Off (no cap) unless given.
+"""
+
+from __future__ import annotations
+
+import contextlib
+import dataclasses
+import time
+from typing import Any, Callable, Iterable, Mapping, Sequence
+
+OPTIONS: Mapping[str, str] = {
+    "lean_lora": "exact: plain base layers + route-switched LoRA term with PEFT's operations",
+    "lora_bf16": "tolerance: route-switched LoRA term with bf16 factors",
+    "merged_views": "tolerance: screen_geometry route reads merged bf16 weight copies",
+    "uint8_pixels": "exact: uint8 patches, pinned upload, device normalisation",
+    "gpu_preprocess": "tolerance: uint8_pixels plus device resize",
+    "device_readout": "exact: cached device index for option rows, one host copy",
+    "cuda_graphs": "tolerance: language model replayed from CUDA graphs at padded lengths",
+    "sdpa_flash": "tolerance: SDPA restricted to the flash kernel",
+    "compile_text": "tolerance: torch.compile(dynamic=True) on the language model",
+}
+EXACT_OPTIONS = frozenset({"lean_lora", "uint8_pixels", "device_readout"})
+IMAGE_TOKEN_CAP = "image_token_cap"
+IMAGE_TOKEN_CAP_DEFAULT = "image_token_cap_default"
+PARAMETERISED_OPTIONS: Mapping[str, str] = {
+    IMAGE_TOKEN_CAP: "output-changing: native-route processor max pixels = N x 32 x 32 (N tokens)",
+    IMAGE_TOKEN_CAP_DEFAULT: "output-changing: as image_token_cap, only for questions the text "
+                             "rule routes to default (screen_geometry stays native)",
+}
+ROUTE_MODES = ("lean_lora", "lora_bf16", "merged_views")
+# Padded lengths for the recorded graphs: 64-token steps to 1,024, then 128, 256 and 512.
+GRAPH_LENGTHS = tuple(list(range(64, 1025, 64)) + list(range(1152, 2049, 128))
+                      + list(range(2304, 4097, 256)) + list(range(4608, 8193, 512)))
+
+
+class ServingOptionError(ValueError):
+    """Unknown or conflicting serving options."""
+
+
+def _parameter(item: str) -> tuple[str, int] | None:
+    name, sep, raw = item.partition("=")
+    if not sep:
+        return None
+    name, raw = name.strip(), raw.strip()
+    if name not in PARAMETERISED_OPTIONS:
+        raise ServingOptionError(f"unknown serving option {name!r}; parameterised options: "
+                                 f"{sorted(PARAMETERISED_OPTIONS)}")
+    if not raw.isdigit():
+        raise ServingOptionError(f"{name} needs a positive integer, got {raw!r}")
+    from open_decisions.image_jev import vision
+
+    try:
+        return name, vision.validate_image_token_cap(int(raw))
+    except ValueError as error:
+        raise ServingOptionError(str(error)) from None
+
+
+def parse_options(value: str | Iterable[str] | None) -> tuple[str, ...]:
+    """Validate options; return them in the canonical (``OPTIONS``) order.
+
+    A parameterised option (``image_token_cap=N``) follows the named options, normalised to
+    ``name=N``; giving it twice is refused.
+    """
+    if value is None:
+        return ()
+    items = [item.strip() for item in value.split(",")] if isinstance(value, str) else list(value)
+    items = [item for item in items if item]
+    parameters: dict[str, int] = {}
+    named = []
+    for item in items:
+        parsed = _parameter(item)
+        if parsed is None:
+            named.append(item)
+            continue
+        name, number = parsed
+        if name in parameters and parameters[name] != number:
+            raise ServingOptionError(f"{name} given twice ({parameters[name]} and {number})")
+        parameters[name] = number
+    if len(parameters) > 1:
+        raise ServingOptionError("image_token_cap and image_token_cap_default are alternatives")
+    items = named
+    unknown = sorted(set(items) - set(OPTIONS))
+    if unknown:
+        raise ServingOptionError(f"unknown serving options {unknown}; known: {sorted(OPTIONS)}"
+                                 f" and {sorted(f'{k}=N' for k in PARAMETERISED_OPTIONS)}")
+    chosen = set(items)
+    modes = [name for name in ROUTE_MODES if name in chosen]
+    if "lora_bf16" in chosen and "merged_views" in chosen:
+        raise ServingOptionError("lora_bf16 and merged_views are alternative route modes")
+    if "compile_text" in chosen and "cuda_graphs" in chosen:
+        raise ServingOptionError("compile_text and cuda_graphs are alternatives")
+    if "gpu_preprocess" in chosen:
+        chosen.add("uint8_pixels")
+    if modes and "lean_lora" not in chosen:
+        chosen.add("lean_lora")  # lora_bf16/merged_views use the same route modules
+    return (tuple(name for name in OPTIONS if name in chosen)
+            + tuple(f"{name}={parameters[name]}" for name in PARAMETERISED_OPTIONS
+                    if name in parameters))
+
+
+def image_token_cap(options: Iterable[str], name: str = IMAGE_TOKEN_CAP) -> int | None:
+    """The ``image_token_cap=N`` value among parsed options, or None (no cap: the default)."""
+    for item in options:
+        parsed = _parameter(item)
+        if parsed is not None and parsed[0] == name:
+            return parsed[1]
+    return None
+
+
+def image_token_cap_default(options: Iterable[str]) -> int | None:
+    """The route-conditioned ``image_token_cap_default=N`` value, or None."""
+    return image_token_cap(options, IMAGE_TOKEN_CAP_DEFAULT)
+
+
+def is_exact(options: Iterable[str]) -> bool:
+    """True only for options that keep the served function bit-identical (no cap)."""
+    return set(options) <= EXACT_OPTIONS
+
+
+def route_mode(options: Iterable[str]) -> str | None:
+    chosen = set(options)
+    if "merged_views" in chosen:
+        return "merged"
+    if "lora_bf16" in chosen:
+        return "bf16"
+    if "lean_lora" in chosen:
+        return "exact"
+    return None
+
+
+# ---------------------------------------------------------------------------------- route modules
+class RouteState:
+    """One shared flag: True while the screen_geometry route is served."""
+
+    __slots__ = ("on",)
+
+    def __init__(self, on: bool = True):
+        self.on = bool(on)
+
+
+def _route_linear_class():
+    import torch
+    from torch import nn
+    from torch.nn import functional as F
+
+    class RouteLinear(nn.Module):
+        """A LoRA-targeted linear layer with the adapter switched by ``RouteState``.
+
+        ``exact`` replicates PEFT's vanilla LoRA forward operation by operation:
+        ``result = base(x); x = x.to(A.dtype); result = result + B(A(x)) * scaling;
+        result.to(result_dtype)``.  Off-route it calls the original base layer only.
+        """
+
+        def __init__(self, base, lora_a, lora_b, scaling: float, state: RouteState, mode: str):
+            super().__init__()
+            if mode not in ("exact", "bf16", "merged"):
+                raise ValueError("unknown route mode")
+            self.base = base
+            self.state = state
+            self.mode = mode
+            self.scaling = float(scaling)
+            weight = base.weight
+            if mode == "merged":
+                delta = (lora_b.detach().float() @ lora_a.detach().float()) * self.scaling
+                merged = (weight.detach().float() + delta.to(weight.device)).to(weight.dtype)
+                self.register_buffer("merged_weight", merged, persistent=False)
+                self.lora_a = self.lora_b = None
+            else:
+                dtype = weight.dtype if mode == "bf16" else lora_a.dtype
+                self.register_buffer("lora_a", lora_a.detach().to(dtype), persistent=False)
+                self.register_buffer("lora_b", lora_b.detach().to(dtype), persistent=False)
+
+        @property
+        def weight(self):  # modules that inspect .weight keep working
+            return self.base.weight
+
+        def forward(self, x):
+            if not self.state.on:
+                return self.base(x)
+            if self.mode == "merged":
+                return F.linear(x, self.merged_weight, self.base.bias)
+            result = self.base(x)
+            result_dtype = result.dtype
+            if x.dtype != self.lora_a.dtype:
+                x = x.to(dtype=self.lora_a.dtype)
+            result = result + F.linear(F.linear(x, self.lora_a), self.lora_b) * self.scaling
+            return result.to(result_dtype)
+
+    return RouteLinear, torch
+
+
+def _lora_layers(root) -> list[tuple[str, Any]]:
+    return [(name, module) for name, module in root.named_modules()
+            if hasattr(module, "base_layer") and hasattr(module, "lora_A")
+            and hasattr(module, "lora_B") and hasattr(module, "scaling")]
+
+
+def install_route_modules(peft_model, *, adapter_name: str, mode: str):
+    """Replace PEFT LoRA wrappers by ``RouteLinear``; return (plain model, state, record)."""
+    from torch import nn
+
+    RouteLinear, torch = _route_linear_class()
+    root = peft_model.get_base_model() if hasattr(peft_model, "get_base_model") else peft_model
+    state = RouteState(True)
+    layers = _lora_layers(root)
+    if not layers:
+        raise ServingOptionError("no LoRA layers found to replace")
+    extra_bytes = 0
+    for name, module in layers:
+        if not isinstance(module.base_layer, nn.Linear):
+            raise ServingOptionError(f"{name}: LoRA on a non-linear layer is not supported")
+        if adapter_name not in module.lora_A or adapter_name not in module.lora_B:
+            raise ServingOptionError(f"{name}: adapter {adapter_name!r} missing")
+        if adapter_name in getattr(module, "lora_variant", {}) or getattr(module, "merged", False):
+            raise ServingOptionError(f"{name}: DoRA/variant or merged LoRA is not supported")
+        if getattr(module, "fan_in_fan_out", False):
+            raise ServingOptionError(f"{name}: fan_in_fan_out is not supported")
+        dropout = module.lora_dropout[adapter_name]
+        if not isinstance(dropout, nn.Identity) and getattr(dropout, "p", 0.0) != 0.0:
+            raise ServingOptionError(f"{name}: LoRA dropout must be zero at serving")
+        lora_a, lora_b = module.lora_A[adapter_name], module.lora_B[adapter_name]
+        if getattr(lora_a, "bias", None) is not None or getattr(lora_b, "bias", None) is not None:
+            raise ServingOptionError(f"{name}: LoRA bias is not supported")
+        replacement = RouteLinear(module.base_layer, lora_a.weight, lora_b.weight,
+                                  module.scaling[adapter_name], state, mode)
+        if mode == "merged":
+            extra_bytes += replacement.merged_weight.numel() * replacement.merged_weight.element_size()
+        parent_name, _, child = name.rpartition(".")
+        parent = root.get_submodule(parent_name) if parent_name else root
+        setattr(parent, child, replacement)
+    del peft_model
+    if torch.cuda.is_available():
+        torch.cuda.empty_cache()
+    record = {"route_mode": mode, "replaced_layers": len(layers),
+              "merged_view_bytes": extra_bytes}
+    return root.eval(), state, record
+
+
+class RouteSwitch:
+    """Adapter context for ``RouteLinear`` models: one flag assignment per route change."""
+
+    def __init__(self, state: RouteState):
+        self.state = state
+
+    def __call__(self, route):
+        from open_decisions.image_jev import router
+
+        if route not in router.ROUTES:
+            raise ValueError("unknown adapter route")
+        self.state.on = route == router.SCREEN_GEOMETRY
+        return contextlib.nullcontext()
+
+
+# ------------------------------------------------------------------------------------ processor
+class PixelProcessor:
+    """Processor proxy returning device-normalised pixels from uint8 patches.
+
+    The image processor normalises before patchifying; normalisation is per channel and
+    elementwise and patchify is a permutation plus temporal duplication, so normalising the
+    uint8 patches with the same fused mean/std tensors gives identical float32 values.
+    """
+
+    def __init__(self, processor, device, *, gpu_resize: bool = False):
+        object.__setattr__(self, "_inner", processor)
+        object.__setattr__(self, "_device", device)
+        object.__setattr__(self, "_gpu_resize", bool(gpu_resize))
+        object.__setattr__(self, "_norm", None)
+
+    def __getattr__(self, name):
+        return getattr(self._inner, name)
+
+    def __setattr__(self, name, value):
+        setattr(self._inner, name, value)
+
+    def _mean_std(self, device):
+        import torch
+
+        if self._norm is None or self._norm[0] != device:
+            ip = self._inner.image_processor
+            mean = torch.tensor(ip.image_mean, device=device) * (1.0 / ip.rescale_factor)
+            std = torch.tensor(ip.image_std, device=device) * (1.0 / ip.rescale_factor)
+            per = int(ip.temporal_patch_size) * int(ip.patch_size) * int(ip.patch_size)
+            object.__setattr__(self, "_norm", (device, mean.to(torch.float32).repeat_interleave(per),
+                                               std.to(torch.float32).repeat_interleave(per)))
+        return self._norm[1], self._norm[2]
+
+    def normalize(self, patches):
+        import torch
+
+        ip = self._inner.image_processor
+        if not (getattr(ip, "do_rescale", True) and getattr(ip, "do_normalize", True)):
+            raise RuntimeError("uint8_pixels expects the processor's default rescale+normalize")
+        mean, std = self._mean_std(patches.device)
+        return patches.to(dtype=torch.float32).sub_(mean).div_(std)
+
+    def __call__(self, *args, **kwargs):
+        import torch
+
+        if kwargs.get("images") is None and len(args) < 2:
+            return self._inner(*args, **kwargs)
+        images_kwargs = dict(kwargs.pop("images_kwargs", None) or {})
+        images_kwargs.update(do_rescale=False, do_normalize=False)
+        if self._gpu_resize and self._device is not None:
+            images_kwargs["device"] = self._device
+        output = self._inner(*args, images_kwargs=images_kwargs, **kwargs)
+        pixels = output["pixel_values"]
+        if pixels.dtype != torch.uint8:
+            raise RuntimeError(f"expected uint8 patches, got {pixels.dtype}")
+        device = self._device
+        if device is not None and str(device).startswith("cuda") and pixels.device.type == "cpu":
+            pixels = pixels.pin_memory().to(device, non_blocking=True)
+        elif device is not None and pixels.device != torch.device(device):
+            pixels = pixels.to(device)
+        output["pixel_values"] = self.normalize(pixels)
+        return output
+
+
+# --------------------------------------------------------------------------------------- readout
+class DeviceReadout:
+    """``vision.option_probs`` with a cached device index (same operations after the gather)."""
+
+    def __init__(self):
+        self._index: dict[tuple, Any] = {}
+
+    def __call__(self, logits, letter_token_ids, temperature: float) -> list[float]:
+        import math
+
+        import torch
+
+        if not math.isfinite(temperature) or temperature <= 0:
+            raise ValueError("temperature must be finite and greater than zero")
+        key = (str(logits.device), tuple(letter_token_ids))
+        index = self._index.get(key)
+        if index is None:
+            if any(not 0 <= int(i) < logits.numel() for i in letter_token_ids):
+                raise ValueError("letter_token_ids must be valid vocabulary indices")
+            index = torch.tensor(list(letter_token_ids), dtype=torch.long, device=logits.device)
+            self._index[key] = index
+        selected = logits.index_select(0, index).to(dtype=torch.float32) / temperature
+        values = torch.log_softmax(selected, dim=-1).exp().detach().cpu().tolist()
+        if not all(math.isfinite(value) for value in values):
+            raise ValueError("option probabilities are non-finite")
+        return values
+
+
+# ---------------------------------------------------------------------------------------- graphs
+def hf_model(model):
+    return model.get_base_model() if hasattr(model, "get_base_model") else model
+
+
+def _move(value, device):
+    from open_decisions.image_jev.speed import move_to_device
+
+    return move_to_device(value, device)
+
+
+def embeds_positions(model, prepared, device):
+    """The multimodal forward up to the language model (the model's own methods, same order)."""
+    import torch
+
+    top = hf_model(model)
+    core = top.model
+    ids = _move(prepared.input_ids, device)
+    pixels = _move(prepared.pixel_values, device)
+    grid = _move(prepared.image_grid_thw, device)
+    mm = getattr(prepared, "mm_token_type_ids", None)
+    mm = None if mm is None else _move(mm, device)
+    embeds = core.get_input_embeddings()(ids)
+    if pixels is not None:
+        features = core.get_image_features(pixels, grid, return_dict=True).pooler_output
+        features = torch.cat(features, dim=0).to(embeds.device, embeds.dtype)
+        mask, _ = core.get_placeholder_mask(ids, inputs_embeds=embeds, image_features=features)
+        embeds = embeds.masked_scatter(mask, features)
+    positions = core.compute_3d_position_ids(
+        input_ids=ids, inputs_embeds=embeds, image_grid_thw=grid, video_grid_thw=None,
+        attention_mask=None, past_key_values=None, mm_token_type_ids=mm)
+    return embeds, positions
+
+
+def pad_positions(positions, size: int):
+    """Right-pad (3, 1, n) position ids to ``size`` by continuing each axis from its last value."""
+    import torch
+
+    n = positions.shape[-1]
+    if size < n:
+        raise ValueError("padded size is shorter than the sequence")
+    if size == n:
+        return positions
+    tail = positions[..., -1:] + torch.arange(1, size - n + 1, device=positions.device,
+                                              dtype=positions.dtype)
+    return torch.cat((positions, tail), dim=-1)
+
+
+def eager_last_logits(model, embeds, positions):
+    top = hf_model(model)
+    hidden = top.model.language_model(input_ids=None, position_ids=positions, attention_mask=None,
+                                      past_key_values=None, inputs_embeds=embeds,
+                                      use_cache=False).last_hidden_state
+    return top.lm_head(hidden[:, -1:, :])[0, 0]
+
+
+class DecoderGraphs:
+    """CUDA graphs of language model + last-row vocabulary projection at padded lengths."""
+
+    def __init__(self, model, *, lengths: Sequence[int], routes: Sequence[Any],
+                 set_route: Callable[[Any], Any] | None, device, log=print):
+        import torch
+
+        top = hf_model(model)
+        self.model = model
+        lm, head = top.model.language_model, top.lm_head
+        hidden = int(top.config.text_config.hidden_size)
+        dtype = next(lm.parameters()).dtype
+        self.graphs: dict[tuple[Any, int], tuple] = {}
+        self.lengths = sorted(set(int(n) for n in lengths))
+        started = time.monotonic()
+        pool = None
+        with torch.inference_mode():
+            for route in routes:
+                if set_route is not None:
+                    set_route(route)
+                for n in sorted(self.lengths, reverse=True):  # longest first; shorter reuse pool
+                    embeds = torch.zeros(1, n, hidden, device=device, dtype=dtype)
+                    positions = torch.arange(n, device=device).view(1, 1, n).expand(3, 1, n).contiguous()
+                    last = torch.zeros(1, dtype=torch.long, device=device)
+
+                    def step(embeds=embeds, positions=positions, last=last):
+                        state = lm(input_ids=None, position_ids=positions, attention_mask=None,
+                                   past_key_values=None, inputs_embeds=embeds,
+                                   use_cache=False).last_hidden_state
+                        return head(state[0].index_select(0, last))[0]
+
+                    side = torch.cuda.Stream()
+                    side.wait_stream(torch.cuda.current_stream())
+                    with torch.cuda.stream(side):
+                        for _ in range(3):
+                            step()
+                    torch.cuda.current_stream().wait_stream(side)
+                    graph = torch.cuda.CUDAGraph()
+                    with torch.cuda.graph(graph, pool=pool):
+                        out = step()
+                    pool = graph.pool()
+                    self.graphs[(route, n)] = (graph, embeds, positions, last, out)
+        self.capture_seconds = time.monotonic() - started
+        self.routes = list(routes)
+        log(f"captured {len(self.graphs)} graphs in {self.capture_seconds:.1f} s")
+
+    def fits(self, n: int) -> bool:
+        return bool(self.lengths) and n <= self.lengths[-1]
+
+    def run(self, route, embeds, positions):
+        n = embeds.shape[1]
+        size = next(x for x in self.lengths if x >= n)
+        graph, static_embeds, static_positions, last, out = self.graphs[(route, size)]
+        static_embeds.zero_()
+        static_embeds[:, :n].copy_(embeds)
+        static_positions.copy_(pad_positions(positions, size))
+        last.fill_(n - 1)
+        graph.replay()
+        return out.clone()
+
+
+class GraphLogits:
+    """``logits_fn`` for the Predictor: eager multimodal front, graph-replayed language model."""
+
+    def __init__(self, graphs: DecoderGraphs | None, device, *, route_key: Callable[[Any], Any]):
+        self.graphs, self.device, self.route_key = graphs, device, route_key
+        self.replayed = self.eager = 0
+
+    def __call__(self, model, prepared, *, timer, route=None):
+        from open_decisions.image_jev.speed import PhaseTimer
+
+        timer = timer or PhaseTimer()
+        with timer.phase("forward"):
+            embeds, positions = embeds_positions(model, prepared, self.device)
+            n = embeds.shape[1]
+            if self.graphs is not None and self.graphs.fits(n):
+                self.replayed += 1
+                return self.graphs.run(self.route_key(route), embeds, positions)
+            self.eager += 1
+            return eager_last_logits(model, embeds, positions)
+
+
+# ------------------------------------------------------------------------------------------ apply
+@dataclasses.dataclass
+class Serving:
+    model: Any
+    processor: Any
+    adapter_context: Callable | None
+    inference_context: Callable[[], Any] | None
+    logits_fn: Callable | None
+    readout_fn: Callable | None
+    record: dict
+    image_token_cap: int | None = None
+    image_token_cap_default: int | None = None
+
+
+def _sdpa_flash_context():
+    from torch.nn.attention import SDPBackend, sdpa_kernel
+
+    return sdpa_kernel([SDPBackend.FLASH_ATTENTION])
+
+
+def apply_serving_options(model, processor, options: Sequence[str], *, device, routed: bool,
+                          adapter_name: str | None, adapter_context: Callable | None,
+                          inference_context: Callable[[], Any] | None,
+                          graph_lengths: Sequence[int] = GRAPH_LENGTHS, log=print) -> Serving:
+    """Apply validated ``options`` to a loaded (PEFT or plain) model and its processor."""
+    import torch
+
+    options = parse_options(options)
+    record: dict[str, Any] = {"serving_options": list(options), "exact": is_exact(options)}
+    cap = image_token_cap(options)
+    if cap is not None:
+        from open_decisions.image_jev import vision
+
+        record["image_token_cap"] = cap
+        record["native_max_pixels"] = vision.native_max_pixels(cap)
+    cap_default = image_token_cap_default(options)
+    if cap_default is not None:
+        from open_decisions.image_jev import vision
+
+        record["image_token_cap_default"] = cap_default
+        record["native_max_pixels_default_route"] = vision.native_max_pixels(cap_default)
+    started = time.monotonic()
+    mode = route_mode(options)
+    if mode is not None:
+        if not routed or adapter_name is None:
+            raise ServingOptionError("route modes need the routed single-adapter system")
+        model, state, route_record = install_route_modules(model, adapter_name=adapter_name,
+                                                           mode=mode)
+        adapter_context = RouteSwitch(state)
+        record.update(route_record)
+    if "uint8_pixels" in options:
+        processor = PixelProcessor(processor, device, gpu_resize="gpu_preprocess" in options)
+    readout_fn = DeviceReadout() if "device_readout" in options else None
+    if "sdpa_flash" in options:
+        base_context = inference_context
+
+        @contextlib.contextmanager
+        def flash_context():
+            with (base_context() if base_context is not None else contextlib.nullcontext()):
+                with _sdpa_flash_context():
+                    yield
+
+        inference_context = flash_context
+    if "compile_text" in options:
+        core = hf_model(model).model
+        core.language_model = torch.compile(core.language_model, dynamic=True)
+        record["compile"] = "torch.compile(language_model, dynamic=True)"
+    logits_fn = None
+    if "cuda_graphs" in options:
+        if not str(device).startswith("cuda"):
+            raise ServingOptionError("cuda_graphs needs a CUDA device")
+        from open_decisions.image_jev import router
+
+        routes = list(router.ROUTES) if routed else [None]
+        set_route = adapter_context if routed else None
+        context = inference_context() if inference_context is not None else contextlib.nullcontext()
+        with context:
+            graphs = DecoderGraphs(model, lengths=graph_lengths, routes=routes,
+                                   set_route=set_route, device=device, log=log)
+        logits_fn = GraphLogits(graphs, device, route_key=(lambda route: route) if routed
+                                else (lambda route: None))
+        record.update({"graph_count": len(graphs.graphs), "graph_lengths": list(graphs.lengths),
+                       "graph_capture_seconds": round(graphs.capture_seconds, 2)})
+    record["apply_seconds"] = round(time.monotonic() - started, 2)
+    return Serving(model=model, processor=processor, adapter_context=adapter_context,
+                   inference_context=inference_context, logits_fn=logits_fn,
+                   readout_fn=readout_fn, record=record, image_token_cap=cap,
+                   image_token_cap_default=cap_default)
+
+
+__all__ = [
+    "DecoderGraphs", "DeviceReadout", "EXACT_OPTIONS", "GRAPH_LENGTHS", "GraphLogits",
+    "IMAGE_TOKEN_CAP", "IMAGE_TOKEN_CAP_DEFAULT", "OPTIONS", "PARAMETERISED_OPTIONS",
+    "image_token_cap", "image_token_cap_default",
+    "PixelProcessor", "RouteState", "RouteSwitch", "Serving", "ServingOptionError",
+    "apply_serving_options", "install_route_modules", "is_exact", "pad_positions",
+    "parse_options", "route_mode",
+]

open_decisions/image_jev/release/predictor.py

@@ -228,16 +228,134 @@ class RoutedCalibration:
         return self.route_type_temperatures[route_name].get(bucket, self.global_temperature)
 
 
+_ROUTED_V2_CALIBRATION_FIELDS = {
+    "adapter_sha256_by_route", "artifact_sha256", "base", "image_route",
+    "image_route_config_sha256", "regularization", "route_global_temperatures",
+    "route_type_counts", "route_type_temperatures", "rules_sha256", "rules_version",
+    "sources", "version",
+}
+ROUTED_V2_CALIBRATION_VERSION = "image-jev/router-calibration/v2"
+
+
+def _sha256_text(value: Any) -> bool:
+    return (isinstance(value, str) and len(value) == 64
+            and all(char in "0123456789abcdef" for char in value))
+
+
+@dataclass(frozen=True)
+class RoutedCalibrationV2:
+    """Serving view of ``image-jev/router-calibration/v2`` (one adapter per route)."""
+
+    route_global_temperatures: Mapping[str, float]
+    route_type_temperatures: Mapping[str, Mapping[str, float]]
+    artifact_sha256: str
+    identity_sha256: str
+    adapter_sha256_by_route: Mapping[str, str]
+    base: Mapping[str, str]
+    rules_sha256: str
+
+    @classmethod
+    def from_artifact(
+        cls, value: Mapping[str, Any], *,
+        adapter_sha256_by_route: Mapping[str, str] | None = None,
+        base: Mapping[str, str] | None = None,
+    ) -> "RoutedCalibrationV2":
+        if not isinstance(value, Mapping) or set(value) != _ROUTED_V2_CALIBRATION_FIELDS:
+            raise ValueError("routed calibration v2 artifact fields differ")
+        body = {key: item for key, item in value.items() if key != "artifact_sha256"}
+        if value["artifact_sha256"] != hashlib.sha256(_compact(body)).hexdigest():
+            raise ValueError("routed calibration artifact hash differs")
+        if value["version"] != ROUTED_V2_CALIBRATION_VERSION:
+            raise ValueError("routed calibration artifact version differs")
+        if value["rules_version"] != router.RULES_VERSION \
+                or value["rules_sha256"] != router.RULES_SHA256:
+            raise ValueError("routed calibration rules binding differs")
+        if value["image_route"] != IMAGE_ROUTE \
+                or value["image_route_config_sha256"] != vision.IMAGE_ROUTE_CONFIG_SHA256[IMAGE_ROUTE]:
+            raise ValueError("routed calibration image-route binding differs")
+        hashes = value["adapter_sha256_by_route"]
+        if (not isinstance(hashes, Mapping) or set(hashes) != set(router.ROUTES)
+                or not all(_sha256_text(hashes[name]) for name in router.ROUTES)):
+            raise ValueError("routed calibration adapter hashes are invalid")
+        if adapter_sha256_by_route is not None and dict(hashes) != dict(adapter_sha256_by_route):
+            raise ValueError("routed calibration adapter binding differs")
+        if not isinstance(value["base"], Mapping) or set(value["base"]) != {"repo", "revision"}:
+            raise ValueError("routed calibration base pin differs")
+        if base is not None and value["base"] != dict(base):
+            raise ValueError("routed calibration base binding differs")
+        globals_ = value["route_global_temperatures"]
+        route_temperatures = value["route_type_temperatures"]
+        route_counts = value["route_type_counts"]
+        for name, mapping in (("global", globals_), ("temperature", route_temperatures),
+                              ("count", route_counts), ("source", value["sources"])):
+            if not isinstance(mapping, Mapping) or set(mapping) != set(router.ROUTES):
+                raise ValueError(f"routed calibration {name} routes differ")
+        checked = {}
+        for route_name in router.ROUTES:
+            values = route_temperatures[route_name]
+            counts = route_counts[route_name]
+            if (
+                not isinstance(values, Mapping)
+                or set(values) != set(TYPE_BUCKET_BY_OPTION_COUNT.values())
+                or not isinstance(counts, Mapping)
+                or set(counts) != set(TYPE_BUCKET_BY_OPTION_COUNT.values())
+            ):
+                raise ValueError("routed calibration temperatures must be objects")
+            if any(
+                isinstance(count, bool) or not isinstance(count, int) or count < 0
+                for count in counts.values()
+            ):
+                raise ValueError("routed calibration counts are invalid")
+            checked[route_name] = {
+                str(kind): _temperature(item, f"route_type_temperatures[{route_name}][{kind}]")
+                for kind, item in values.items()
+            }
+        return cls(
+            route_global_temperatures={
+                name: _temperature(globals_[name], f"route_global_temperatures[{name}]")
+                for name in router.ROUTES
+            },
+            route_type_temperatures=checked,
+            artifact_sha256=str(value["artifact_sha256"]),
+            identity_sha256=hashlib.sha256(_compact(dict(value))).hexdigest(),
+            adapter_sha256_by_route=dict(hashes), base=dict(value["base"]),
+            rules_sha256=str(value["rules_sha256"]),
+        )
+
+    def temperature(self, n_options: int, route_name: str) -> float:
+        if route_name not in router.ROUTES:
+            raise ValueError("unknown routed calibration route")
+        bucket = TYPE_BUCKET_BY_OPTION_COUNT.get(int(n_options))
+        if bucket is None:
+            return self.route_global_temperatures[route_name]
+        return self.route_type_temperatures[route_name].get(
+            bucket, self.route_global_temperatures[route_name])
+
+
+ROUTED_CALIBRATIONS = (RoutedCalibration, RoutedCalibrationV2)
+
+
 def serving_calibration_from_artifact(
     value: Mapping[str, Any], *, checkpoint_sha256: str | None = "",
     adapter_sha256: str | None = None, base: Mapping[str, str] | None = None,
-) -> Calibration | RoutedCalibration:
+    adapter_sha256_by_route: Mapping[str, str] | None = None,
+) -> Calibration | RoutedCalibration | RoutedCalibrationV2:
     if not isinstance(value, Mapping):
         raise ValueError("calibration artifact must be an object")
     if value.get("version") == "image-jev/router-calibration/v1":
+        if adapter_sha256_by_route is not None:
+            raise ValueError("router-calibration/v1 binds one adapter, not one per route")
         return RoutedCalibration.from_artifact(
             value, adapter_sha256=adapter_sha256, base=base
         )
+    if value.get("version") == ROUTED_V2_CALIBRATION_VERSION:
+        if adapter_sha256 is not None:
+            raise ValueError("router-calibration/v2 binds one adapter per route")
+        return RoutedCalibrationV2.from_artifact(
+            value, adapter_sha256_by_route=adapter_sha256_by_route, base=base
+        )
+    if adapter_sha256_by_route is not None:
+        raise ValueError("single-route calibration cannot bind route adapters")
     return Calibration.from_artifact(value, checkpoint_sha256=checkpoint_sha256)
 
 
@@ -353,13 +471,29 @@ def _length(value: Any) -> int:
 class Predictor:
     """One prefill per question; calibrated option probabilities in the board's shape."""
 
-    def __init__(self, model, processor, calibration: Calibration | RoutedCalibration, *, model_name: str,
+    def __init__(self, model, processor, calibration: Calibration | RoutedCalibration | RoutedCalibrationV2, *, model_name: str,
                  context_policy: str = "prepend", max_input_tokens: int | None = None,
                  inference_context: Callable[[], Any] | None = None,
                  adapter_context: Callable[[str], Any] | None = None,
-                 synchronize: Callable[[], Any] | None = None, forward_call=None) -> None:
+                 synchronize: Callable[[], Any] | None = None, forward_call=None,
+                 logits_fn: Callable[..., Any] | None = None,
+                 readout_fn: Callable[..., list[float]] | None = None,
+                 image_token_cap: int | None = None,
+                 image_token_cap_default: int | None = None) -> None:
         if context_policy not in CONTEXT_POLICIES:
             raise ValueError(f"context_policy must be one of {CONTEXT_POLICIES}")
+        # Optional native-route image-token cap (``image_token_cap=N`` serving option); None keeps
+        # the processor's own maximum, i.e. the published behaviour.
+        self.image_token_cap = (None if image_token_cap is None
+                                else vision.validate_image_token_cap(image_token_cap))
+        # Route-conditioned variant (``image_token_cap_default=N``): the cap applies only when the
+        # question's text rule (router-rules/v1) routes it to ``default``; screen_geometry stays
+        # native.  Decided on the text route even under ``route_override``.
+        self.image_token_cap_default = (
+            None if image_token_cap_default is None
+            else vision.validate_image_token_cap(image_token_cap_default))
+        if self.image_token_cap is not None and self.image_token_cap_default is not None:
+            raise ValueError("image_token_cap and image_token_cap_default are alternatives")
         self.model, self.processor, self.calibration = model, processor, calibration
         self.model_name = model_name
         self.context_policy = context_policy
@@ -368,18 +502,28 @@ class Predictor:
         self._adapter_context = adapter_context
         self._synchronize = synchronize
         self._forward_call = forward_call
+        # Optional serving fast paths (``release.fastpath``); None keeps the published path.
+        self._logits_fn = logits_fn
+        self._readout_fn = readout_fn
         self._cache = NativeCache(processor)
         try:
             self.device = next(model.parameters()).device
         except (StopIteration, AttributeError):
             self.device = None
 
-    def _read(self, images: Sequence, question: Question, *, loaded_images=None, timer=None) -> tuple[list[float], int]:
+    def _read(self, images: Sequence, question: Question, *, loaded_images=None, timer=None,
+              route_override: str | None = None) -> tuple[list[float], int]:
         example = {"question": question.instructions, "options": list(question.options),
                    "images": list(images)}
+        cap = {} if self.image_token_cap is None else {"image_token_cap": self.image_token_cap}
+        if self.image_token_cap_default is not None and router.route(
+                question.route_instructions or question.instructions,
+                question.options) == router.DEFAULT:
+            cap = {"image_token_cap": self.image_token_cap_default}
         try:
             prepared = vision.prepare_native(example, self.processor, loaded_images=loaded_images,
-                                             cache=self._cache, timer=timer, device=self.device)
+                                             cache=self._cache, timer=timer, device=self.device,
+                                             **cap)
         except (TypeError, ValueError) as error:
             raise RequestError(str(error)) from error
         tokens = _length(prepared.input_ids)
@@ -390,8 +534,8 @@ class Predictor:
             )
         with timer.phase("route_adapter_switch"):
             route_name = None
-            if isinstance(self.calibration, RoutedCalibration):
-                route_name = router.route(
+            if isinstance(self.calibration, ROUTED_CALIBRATIONS):
+                route_name = route_override or router.route(
                     question.route_instructions or question.instructions, question.options
                 )
                 temperature = self.calibration.temperature(len(question.options), route_name)
@@ -406,11 +550,18 @@ class Predictor:
         )
         with route_context:
             with inference_context:
-                logits = vision.last_position_logits(
-                    self.model, prepared, timer=timer, forward_call=self._forward_call)
+                if self._logits_fn is None:
+                    logits = vision.last_position_logits(
+                        self.model, prepared, timer=timer, forward_call=self._forward_call)
+                else:
+                    logits = self._logits_fn(self.model, prepared, timer=timer, route=route_name)
                 with timer.phase("readout_calibration"):
-                    probabilities = vision.option_probs(
-                        logits, prepared.letter_token_ids, temperature=temperature)
+                    if self._readout_fn is None:
+                        probabilities = vision.option_probs(
+                            logits, prepared.letter_token_ids, temperature=temperature)
+                    else:
+                        probabilities = self._readout_fn(
+                            logits, prepared.letter_token_ids, temperature)
         with timer.phase("readout_calibration"):
             total = math.fsum(probabilities)
             return [value / total for value in probabilities], tokens
@@ -421,7 +572,16 @@ class Predictor:
             images, questions = parse_request(body, context_policy=self.context_policy)
         return self.predict_questions(images, questions, timer=timer)
 
-    def predict_questions(self, images, questions, *, timer=None) -> dict[str, Any]:
+    def predict_questions(self, images, questions, *, timer=None,
+                          route_override: str | None = None) -> dict[str, Any]:
+        """Answer parsed questions.  ``route_override`` (evaluation only; never set by the
+        server) serves a routed system on one named route, e.g. ``default`` = the frozen base
+        with its default-route temperatures."""
+        if route_override is not None:
+            if route_override not in router.ROUTES:
+                raise ValueError(f"route_override must be one of {router.ROUTES}")
+            if not isinstance(self.calibration, ROUTED_CALIBRATIONS):
+                raise ValueError("route_override needs a routed system")
         timer = timer or PhaseTimer()
         with timer.phase("decode_load"):
             try:
@@ -431,7 +591,8 @@ class Predictor:
                 raise RequestError(str(error)) from error
         answers, input_tokens = {}, 0
         for question in questions:
-            probabilities, tokens = self._read(images, question, loaded_images=loaded, timer=timer)
+            probabilities, tokens = self._read(images, question, loaded_images=loaded, timer=timer,
+                                               route_override=route_override)
             with timer.phase("response_build"):
                 input_tokens += tokens
                 names = [option["name"] for option in question.options]
@@ -507,9 +668,21 @@ def load_predictor(*, calibration_path: str | Path, model_name: str,
                    adapter_path: str | Path | None = None,
                    base_hashes_path: str | Path | None = None, device: str = "cuda",
                    context_policy: str = "prepend", max_input_tokens: int | None = None,
-                   allow_slow_path: bool = False) -> tuple[Predictor, dict[str, Any]]:
-    """Load the pinned frozen base in bf16 and return the predictor and its provenance."""
+                   allow_slow_path: bool = False, adapters_path: str | Path | None = None,
+                   serving_form: str = "U",
+                   adapter_work_dir: str | Path | None = None,
+                   serving_options: Sequence[str] | str | None = None,
+                   ) -> tuple[Predictor, dict[str, Any]]:
+    """Load the pinned frozen base in bf16 and return the predictor and its provenance.
+
+    ``adapters_path`` (router v3) is a directory holding ``default/`` and
+    ``screen_geometry/`` adapters bound by a ``router-calibration/v2`` artifact;
+    ``serving_form`` selects form U (both attached) or M (merged default).
+    ``serving_options`` (``release.fastpath``) are off by default; any option is recorded in
+    the provenance and none changes weights or calibration.
+    """
     import platform
+    import sys
 
     import torch
     import transformers
@@ -518,19 +691,35 @@ def load_predictor(*, calibration_path: str | Path, model_name: str,
 
     if not allow_slow_path:
         check_fast_path()
+    if adapters_path is not None and adapter_path is not None:
+        raise ValueError("pass either one adapter or a per-route adapters directory, not both")
     adapter = None if adapter_path is None else Path(adapter_path)
     adapter_sha256 = "" if adapter is None else tree_sha256(adapter)
+    route_dirs = None
+    if adapters_path is not None:
+        route_dirs = {name: Path(adapters_path) / name for name in router.ROUTES}
     try:
         calibration_value = json.loads(Path(calibration_path).read_text(encoding="utf-8"))
     except (OSError, json.JSONDecodeError) as error:
         raise ValueError(f"cannot load calibration {calibration_path}: {error}") from error
     base_pin = {"repo": BASE_REPO, "revision": BASE_REVISION}
-    calibration = serving_calibration_from_artifact(
-        calibration_value, checkpoint_sha256=adapter_sha256,
-        adapter_sha256=adapter_sha256 or None, base=base_pin,
-    )
-    routed = isinstance(calibration, RoutedCalibration)
-    if routed and adapter is None:
+    if route_dirs is not None:
+        route_hashes = {name: tree_sha256(path) for name, path in route_dirs.items()}
+        calibration = serving_calibration_from_artifact(
+            calibration_value, base=base_pin, adapter_sha256_by_route=route_hashes,
+        )
+        if not isinstance(calibration, RoutedCalibrationV2):
+            raise ValueError("per-route adapters require router-calibration/v2")
+    else:
+        route_hashes = None
+        calibration = serving_calibration_from_artifact(
+            calibration_value, checkpoint_sha256=adapter_sha256,
+            adapter_sha256=adapter_sha256 or None, base=base_pin,
+        )
+        if isinstance(calibration, RoutedCalibrationV2):
+            raise ValueError("router-calibration/v2 requires the per-route adapters directory")
+    routed = isinstance(calibration, ROUTED_CALIBRATIONS)
+    if isinstance(calibration, RoutedCalibration) and adapter is None:
         raise ValueError("routed calibration requires the screen_geometry adapter")
     snapshot = Path(snapshot_download(BASE_REPO, revision=BASE_REVISION, cache_dir=model_cache,
                                       local_files_only=offline))
@@ -545,20 +734,53 @@ def load_predictor(*, calibration_path: str | Path, model_name: str,
         transformers, "AutoModelForImageTextToText")
     model = auto_model.from_pretrained(str(snapshot), torch_dtype=torch.bfloat16,
                                        device_map=device, trust_remote_code=False, attn_implementation="sdpa")
-    if adapter is not None:
-        from peft import PeftModel
+    route_record = None
+    if route_dirs is not None:
+        import tempfile
 
-        loaded = PeftModel.from_pretrained(
-            model, str(adapter), **({"adapter_name": router.SCREEN_GEOMETRY} if routed else {})
+        from open_decisions.image_jev.release.adapters import attach_route_adapters
+
+        model, adapter_context, route_record = attach_route_adapters(
+            model, default_dir=route_dirs[router.DEFAULT],
+            screen_dir=route_dirs[router.SCREEN_GEOMETRY], form=serving_form,
+            work_dir=adapter_work_dir or tempfile.mkdtemp(prefix="image-hopper-adapters-"),
         )
-        model = loaded if routed else loaded.merge_and_unload()
-    model = model.eval()
-    adapter_context = AdapterSwitch(model) if routed else None
+        model = model.eval()
+    else:
+        if adapter is not None:
+            from peft import PeftModel
+
+            loaded = PeftModel.from_pretrained(
+                model, str(adapter), **({"adapter_name": router.SCREEN_GEOMETRY} if routed else {})
+            )
+            model = loaded if routed else loaded.merge_and_unload()
+        model = model.eval()
+        adapter_context = AdapterSwitch(model) if routed else None
+    from open_decisions.image_jev.release import fastpath
+
+    options = fastpath.parse_options(serving_options)
+    inference_context, logits_fn, readout_fn, serving_record = torch.inference_mode, None, None, None
+    token_cap = token_cap_default = None
+    if options:
+        if route_dirs is not None:
+            raise ValueError("serving options are implemented for the single-adapter systems")
+        serving = fastpath.apply_serving_options(
+            model, processor, options, device=device, routed=routed,
+            adapter_name=router.SCREEN_GEOMETRY if routed else None,
+            adapter_context=adapter_context, inference_context=torch.inference_mode,
+            log=lambda text: print(text, file=sys.stderr))
+        model, processor = serving.model, serving.processor
+        adapter_context, inference_context = serving.adapter_context, serving.inference_context
+        logits_fn, readout_fn, serving_record = serving.logits_fn, serving.readout_fn, serving.record
+        token_cap = serving.image_token_cap
+        token_cap_default = serving.image_token_cap_default
     predictor = Predictor(model, processor, calibration, model_name=model_name,
                           context_policy=context_policy, max_input_tokens=max_input_tokens,
-                          inference_context=torch.inference_mode,
+                          inference_context=inference_context,
                           adapter_context=adapter_context,
-                          synchronize=(torch.cuda.synchronize if str(device).startswith("cuda") else None))
+                          synchronize=(torch.cuda.synchronize if str(device).startswith("cuda") else None),
+                          logits_fn=logits_fn, readout_fn=readout_fn, image_token_cap=token_cap,
+                          image_token_cap_default=token_cap_default)
     provenance = {
         "base_repo": BASE_REPO, "base_revision": BASE_REVISION,
         "base_files_verified": verified,
@@ -570,17 +792,33 @@ def load_predictor(*, calibration_path: str | Path, model_name: str,
         "system_kind": "routed" if routed else "single-route",
         "router_rules_version": router.RULES_VERSION if routed else None,
         "router_rules_sha256": router.RULES_SHA256 if routed else None,
-        "adapter_switching": "PEFT layer enable/disable on route transition" if routed else None,
+        "adapter_switching": (
+            route_record["adapter_switching"] if route_record is not None
+            else "PEFT layer enable/disable on route transition" if routed else None
+        ),
+        "adapters_by_route": (
+            None if route_dirs is None else {
+                name: {"path": str(path), "sha256": route_hashes[name]}
+                for name, path in route_dirs.items()
+            }
+        ),
+        "serving_form": None if route_record is None else route_record["serving_form"],
+        "difference_adapter": (
+            None if route_record is None else route_record.get("difference_adapter")
+        ),
         "precision": "bfloat16", "attention": "sdpa", "processor_use_fast": True, "scorer": "one-prefill-option-letter",
         "context_policy": context_policy, "max_input_tokens": max_input_tokens,
         "torch_version": torch.__version__, "transformers_version": transformers.__version__,
         "python_version": platform.python_version(),
     }
+    if options:  # absent unless enabled: the default provenance is unchanged
+        provenance["serving_options"] = serving_record
     return predictor, provenance
 
 
 __all__ = [
-    "BASE_REPO", "BASE_REVISION", "Calibration", "RoutedCalibration", "Predictor", "Question", "RequestError",
+    "BASE_REPO", "BASE_REVISION", "Calibration", "ROUTED_CALIBRATIONS", "RoutedCalibration",
+    "RoutedCalibrationV2", "Predictor", "Question", "RequestError",
     "TYPE_BUCKET_BY_OPTION_COUNT", "check_fast_path", "load_predictor", "parse_request",
     "serving_calibration_from_artifact", "tree_sha256", "verify_snapshot",
 ]

open_decisions/image_jev/release/server.py

@@ -74,6 +74,12 @@ def build_parser() -> argparse.ArgumentParser:
     parser.add_argument("--calibration", default="calibration.json")
     parser.add_argument("--adapter", default=None,
                         help="optional local adapter directory; calibration must bind its tree hash")
+    parser.add_argument("--adapters", default=None,
+                        help="router v3: directory with default/ and screen_geometry/ adapters; "
+                             "calibration must be router-calibration/v2 bound to both")
+    parser.add_argument("--serving-form", choices=("U", "M"), default="U",
+                        help="router v3: U keeps both adapters attached (set_adapter on route "
+                             "change); M merges default and serves an exact difference adapter")
     parser.add_argument("--base-hashes", default="base-model-files.json",
                         help="hash listing of the pinned base snapshot; '' skips the check")
     parser.add_argument("--model-cache", default=None, help="Hugging Face cache directory")
@@ -87,6 +93,9 @@ def build_parser() -> argparse.ArgumentParser:
     parser.add_argument("--max-input-tokens", type=int, default=None,
                         help="refuse (HTTP 400) requests above this many processed tokens")
     parser.add_argument("--allow-slow-path", action="store_true")
+    parser.add_argument("--serving-options", default="",
+                        help="comma-separated release.fastpath options (default: none, the "
+                             "published path); recorded in /health")
     parser.add_argument("--phase-timing", action="store_true",
                         help="diagnostic synchronized phase_seconds in responses; default off")
     return parser
@@ -98,10 +107,10 @@ def main(argv=None) -> int:
     args = build_parser().parse_args(argv)
     predictor, provenance = load_predictor(
         calibration_path=args.calibration, model_name=args.name, model_cache=args.model_cache,
-        adapter_path=args.adapter,
+        adapter_path=args.adapter, adapters_path=args.adapters, serving_form=args.serving_form,
         offline=args.offline, base_hashes_path=args.base_hashes or None, device=args.device,
         context_policy=args.context_policy, max_input_tokens=args.max_input_tokens,
-        allow_slow_path=args.allow_slow_path,
+        allow_slow_path=args.allow_slow_path, serving_options=args.serving_options or None,
     )
     health = {"model": args.name, **provenance}
     print(json.dumps(health, indent=2, sort_keys=True), file=sys.stderr)

open_decisions/image_jev/release/smoke.py

@@ -127,7 +127,7 @@ def _exchange(handler_type, method: str, path: str, body=None) -> tuple[int, dic
 def run(calibration_path: str | Path) -> dict:
     from open_decisions.image_jev import router
     from open_decisions.image_jev.release.predictor import (
-        Predictor, RoutedCalibration, serving_calibration_from_artifact,
+        ROUTED_CALIBRATIONS, Predictor, serving_calibration_from_artifact,
     )
     from open_decisions.image_jev.release.server import handler
 
@@ -139,6 +139,12 @@ def run(calibration_path: str | Path) -> dict:
     )
     model = _fake_model()
     predictor = Predictor(model, FakeProcessor(), calibration, model_name="smoke")
+
+    def temperature(n_options: int) -> float:
+        # Every smoke question routes to ``default`` (no screen/geometry cues).
+        if isinstance(calibration, ROUTED_CALIBRATIONS):
+            return calibration.temperature(n_options, router.DEFAULT)
+        return calibration.temperature(n_options)
     handler_type = handler(predictor, {"model": "smoke"})
 
     def post(body) -> tuple[int, dict]:
@@ -172,19 +178,13 @@ def run(calibration_path: str | Path) -> dict:
         four = answers["four"]["probabilities"]
         assert list(four) == ["top", "middle", "bottom", "none"]
         assert answers["four"]["choice"] == "top"
-        four_temperature = (
-            calibration.temperature(4, router.DEFAULT)
-            if isinstance(calibration, RoutedCalibration) else calibration.temperature(4)
-        )
+        four_temperature = temperature(4)
         for name, got, want in zip(four, four.values(), _expected(4, four_temperature)):
             assert abs(got - want) < 1e-6, (name, got, want)
         three = answers["three"]["probabilities"]
-        for got, want in zip(three.values(), _expected(3, calibration.global_temperature)):
+        for got, want in zip(three.values(), _expected(3, temperature(3))):
             assert abs(got - want) < 1e-6
-        binary_temperature = (
-            calibration.temperature(2, router.DEFAULT)
-            if isinstance(calibration, RoutedCalibration) else calibration.temperature(2)
-        )
+        binary_temperature = temperature(2)
         want_yes = _expected(2, binary_temperature)[0]
         assert abs(answers["binary"]["noul"] - want_yes) < 1e-6
         assert body["usage"]["output_tokens"] == 0 and body["usage"]["input_tokens"] > 0

open_decisions/image_jev/vision.py

@@ -35,6 +35,12 @@ MIN_PIXELS = 65_536
 MAX_PIXELS = 16_777_216
 MAX_IMAGE_BYTES = 20 * 1024 * 1024
 MAX_IMAGE_PIXELS = 20_000_000
+# Optional native-route image-token cap (a serving option, off by default).  The cap lowers the
+# processor's own max pixels to ``cap * FACTOR**2`` so its smart resize keeps at most ``cap``
+# merged image tokens per image.  Below 256 tokens the processor's minimum-area rule and its
+# one-factor edge clamp can exceed the cap for extreme aspect ratios, so smaller caps are refused.
+MIN_IMAGE_TOKEN_CAP = 256
+MAX_IMAGE_TOKEN_CAP = MAX_PIXELS // (FACTOR * FACTOR)
 BUDGET_VERSION = "b1"
 BUDGETS = MappingProxyType({"photo": 160, "scene": 224, "document": 384, "dense": 768})
 ROUTING_VERSION = "image-jev/content-routing/v1"
@@ -128,6 +134,29 @@ def image_tokens(height: int, width: int, factor: int = FACTOR) -> int:
     return (resized_h // factor) * (resized_w // factor)
 
 
+def validate_image_token_cap(value: Any) -> int:
+    """An explicit native-route image-token cap: an integer in [256, 16384]."""
+    if isinstance(value, bool) or not isinstance(value, int):
+        raise ValueError(f"image_token_cap must be an integer, got {value!r}")
+    if not MIN_IMAGE_TOKEN_CAP <= value <= MAX_IMAGE_TOKEN_CAP:
+        raise ValueError(f"image_token_cap must be between {MIN_IMAGE_TOKEN_CAP} and "
+                         f"{MAX_IMAGE_TOKEN_CAP}, got {value}")
+    return value
+
+
+def native_max_pixels(image_token_cap: int | None = None) -> int:
+    """The processor max pixels of the native route: its default, or ``cap * FACTOR**2``."""
+    if image_token_cap is None:
+        return MAX_PIXELS
+    return validate_image_token_cap(image_token_cap) * FACTOR * FACTOR
+
+
+def native_image_tokens(height: int, width: int, image_token_cap: int | None = None) -> int:
+    """Merged image tokens of one native-route image, with the optional cap applied."""
+    resized_h, resized_w = smart_resize(height, width, max_pixels=native_max_pixels(image_token_cap))
+    return (resized_h // FACTOR) * (resized_w // FACTOR)
+
+
 def budget_pixels(content_class: str) -> int:
     """Return the explicit pixel-area budget for a named content class."""
     try:
@@ -325,12 +354,40 @@ def _output_field(output, name: str):
     return value
 
 
+def _processor_min_pixels(processor) -> int:
+    """The processor's own minimum area (kept unchanged when a token cap lowers the maximum)."""
+    image_processor = getattr(processor, "image_processor", None)
+    size = getattr(image_processor, "size", None)
+    for getter in (lambda: size["shortest_edge"], lambda: getattr(size, "shortest_edge"),
+                   lambda: getattr(image_processor, "min_pixels")):
+        try:
+            value = getter()
+        except (KeyError, TypeError, AttributeError):
+            continue
+        if isinstance(value, int) and not isinstance(value, bool) and value > 0:
+            return value
+    return MIN_PIXELS
+
+
+def _grid_rows(grid) -> list[list[int]]:
+    rows = grid.tolist() if hasattr(grid, "tolist") else [list(row) for row in grid]
+    return [[int(value) for value in row] for row in rows]
+
+
 def _prepare(example: dict, processor, *, image_route: str,
              content_class: str | None = None, loaded_images=None, cache=None,
-             timer=None, device=None) -> PreparedInput:
-    """Prepare one decision after validating an explicit registered image route."""
+             timer=None, device=None, image_token_cap: int | None = None) -> PreparedInput:
+    """Prepare one decision after validating an explicit registered image route.
+
+    ``image_token_cap`` (native route only; default None = the processor's own maximum) lowers
+    the processor's max pixels so every image keeps at most that many merged image tokens.
+    """
     if image_route not in IMAGE_ROUTES:
         raise ValueError(f"image_route must be one of {IMAGE_ROUTES}, got {image_route!r}")
+    if image_token_cap is not None:
+        if image_route != "native":
+            raise ValueError("image_token_cap applies to the native image route only")
+        image_token_cap = validate_image_token_cap(image_token_cap)
     try:
         question, options = example["question"], example["options"]
     except (KeyError, TypeError) as error:
@@ -366,6 +423,11 @@ def _prepare(example: dict, processor, *, image_route: str,
                     "new_size": ready.size,
                     "image_tokens": image_tokens(ready.height, ready.width),
                 }
+                if image_token_cap is not None:
+                    # The processor, not this function, resizes; new_size stays the decoded size.
+                    record["image_token_cap"] = image_token_cap
+                    record["image_tokens"] = native_image_tokens(ready.height, ready.width,
+                                                                 image_token_cap)
             ready_images.append(ready)
             records.append(record)
 
@@ -390,12 +452,28 @@ def _prepare(example: dict, processor, *, image_route: str,
         getattr(image_processor, "backend", None) == "torch" or
         "Fast" in type(image_processor).__name__ or
         any(base.__name__ == "BaseImageProcessorFast" for base in type(image_processor).__mro__))
-    kwargs = {"images_kwargs": {"device": device}} if fast and device is not None else {}
+    images_kwargs = {"device": device} if fast and device is not None else {}
+    if image_token_cap is not None:
+        # Every spelling the processor's kwargs accept, all consistent: min unchanged, max capped.
+        min_pixels, max_pixels = _processor_min_pixels(processor), native_max_pixels(image_token_cap)
+        images_kwargs.update(size={"shortest_edge": min_pixels, "longest_edge": max_pixels},
+                             min_pixels=min_pixels, max_pixels=max_pixels)
+    kwargs = {"images_kwargs": images_kwargs} if images_kwargs else {}
     if cache is not None:
         cache.tokenizer.timer = timer
     token_before = timer.seconds["tokenise_chat_template"] if timer.seconds is not None else 0.0
     with timer.phase("processor_preprocess"):
         output = processor(text=[prompt], images=ready_images, padding=True, return_tensors="pt", **kwargs)
+        if image_token_cap is not None:
+            # Fail closed if the processor ignored the cap (a library change would do that).
+            merge = int(getattr(getattr(processor, "image_processor", None), "merge_size", 2) or 2)
+            served = [t * h * w // (merge * merge)
+                      for t, h, w in _grid_rows(_output_field(output, "image_grid_thw"))]
+            if len(served) != len(records) or any(tokens > image_token_cap for tokens in served):
+                raise RuntimeError(f"processor returned {served} image tokens for an "
+                                   f"{image_token_cap}-token cap")
+            for record, tokens in zip(records, served):
+                record["processor_image_tokens"] = tokens
     if timer.seconds is not None:
         timer.seconds["processor_preprocess"] -= timer.seconds["tokenise_chat_template"] - token_before
     return PreparedInput(input_ids=_output_field(output, "input_ids"),
@@ -413,7 +491,10 @@ def prepare(example: dict, processor, content_class: str | None = None) -> Prepa
 
 
 def prepare_native(example: dict, processor, **kwargs) -> PreparedInput:
-    """Prepare processor-default full-resolution images without budget resizing."""
+    """Prepare processor-default full-resolution images without budget resizing.
+
+    ``image_token_cap=N`` (optional, default off) lowers only the processor's max pixels.
+    """
     return _prepare(example, processor, image_route="native", **kwargs)