[codex] Support raw image offload in v1 train client

tests/test_client_multimodal_types.py

@@ -1,7 +1,6 @@
 from types import SimpleNamespace
 
 import pytest
-
 from verifiers.clients.openai_chat_completions_client import OpenAIChatCompletionsClient
 from verifiers.types import (
     AssistantMessage,
@@ -214,7 +213,6 @@ async def test_anthropic_merges_consecutive_tool_results_into_single_user_messag
 async def test_anthropic_from_native_response_extracts_usage():
     anthropic = pytest.importorskip("anthropic")
     from anthropic.types import Message as AnthropicMessage
-
     from verifiers.clients.anthropic_messages_client import AnthropicMessagesClient
 
     client = AnthropicMessagesClient(object())
@@ -267,7 +265,6 @@ async def test_anthropic_tool_call_round_trips_thinking_blocks():
     pytest.importorskip("anthropic")
     from anthropic.types import Message as AnthropicMessage
     from anthropic.types import Usage as AnthropicUsage
-
     from verifiers.clients.anthropic_messages_client import AnthropicMessagesClient
 
     client = AnthropicMessagesClient(object())
@@ -296,3 +293,54 @@ async def test_anthropic_tool_call_round_trips_thinking_blocks():
         {"type": "thinking", "thinking": "hidden chain", "signature": "sig_1"},
         {"type": "tool_use", "id": "call_1", "name": "lookup", "input": {"q": "x"}},
     ]
+
+
+def test_prepare_images_inplace_offloads_every_image_part_shape(tmp_path):
+    """Ingress offload must cover the full renderer part treaty: nested
+    ``image_url`` dicts, direct-string ``image_url``, direct ``image``
+    strings, and typed pydantic parts — and reject non-file leftovers."""
+    import base64
+
+    pytest.importorskip(
+        "renderers.mm_store", reason="needs a renderers version with raw image offload"
+    )
+    from verifiers.utils.multimodal import prepare_images_inplace
+
+    raw = b"png-ish bytes"
+    data_url = "data:image/png;base64," + base64.b64encode(raw).decode("ascii")
+    wire = {
+        "messages": [
+            {
+                "role": "user",
+                "content": [
+                    {"type": "image_url", "image_url": {"url": data_url}},
+                    {"type": "image_url", "image_url": data_url},
+                    {"type": "image", "image": data_url},
+                ],
+            }
+        ]
+    }
+    typed = UserMessage(
+        content=[ImageUrlContentPart(image_url=ImageUrlSource(url=data_url))]
+    )
+
+    prepare_images_inplace(wire, image_dir=tmp_path)
+    prepare_images_inplace(typed, image_dir=tmp_path)
+
+    parts = wire["messages"][0]["content"]
+    offloaded = [
+        parts[0]["image_url"]["url"],
+        parts[1]["image_url"],
+        parts[2]["image"],
+        typed.content[0].image_url.url,
+    ]
+    assert len(set(offloaded)) == 1  # content-addressed: same bytes, same file
+    assert offloaded[0].startswith("file://")
+    written = list(tmp_path.iterdir())
+    assert len(written) == 1 and written[0].read_bytes() == raw
+
+    with pytest.raises(RuntimeError, match="file://"):
+        prepare_images_inplace(
+            {"type": "image", "image": "https://example.com/x.png"},
+            image_dir=tmp_path,
+        )

verifiers/clients/renderer_client.py

@@ -8,6 +8,7 @@
 concurrent rollouts tokenize in parallel instead of blocking the event loop.
 """
 
+import asyncio
 import json
 import threading
 from collections.abc import Mapping
@@ -56,6 +57,7 @@
     UserMessage,
 )
 from verifiers.utils.client_utils import setup_openai_client
+from verifiers.utils.multimodal import prepare_images_inplace
 
 # Module-level bridge counters. Incremented by every RendererClient instance
 # that tries to stitch a multi-turn prompt; callers (e.g. prime-rl's
@@ -472,6 +474,7 @@ def _get_renderer_or_pool(
     async def to_native_prompt(
         self, messages: Messages
     ) -> tuple[list[RendererMessage], dict]:
+        await asyncio.to_thread(prepare_images_inplace, messages)
         return (
             _attach_tool_call_names([_to_renderer_message(m) for m in messages]),
             {},

verifiers/types.py

@@ -213,7 +213,7 @@ class ResponseTokens(CustomBaseModel):
     completion_logprobs: list[float]
     routed_experts: RoutedExpertsPayload | None = None
     # Renderer-emitted multimodal sidecar (renderers.base.MultiModalData)
-    # carrying processed pixel_values / placeholder ranges per modality.
+    # carrying raw image descriptors / placeholder ranges per modality.
     # Populated by the renderer client when the rollout went through a
     # multimodal-aware renderer; ``None`` otherwise. Stored as ``Any`` to
     # avoid a hard import dependency on ``renderers`` at this layer.
@@ -260,7 +260,7 @@ class TrajectoryStepTokens(TypedDict):
     is_truncated: bool
     routed_experts: RoutedExpertsPayload | None
     # Renderer-emitted multimodal sidecar (renderers.base.MultiModalData)
-    # carrying processed pixel_values / placeholder ranges per modality.
+    # carrying raw image descriptors / placeholder ranges per modality.
     # ``NotRequired`` because text-only rollouts (and non-renderer client
     # types) never populate it.
     multi_modal_data: NotRequired[Any]

verifiers/utils/multimodal.py

@@ -0,0 +1,104 @@
+"""Multimodal ingress helpers for renderer-backed training."""
+
+from __future__ import annotations
+
+from importlib import import_module
+from pathlib import Path
+from typing import Any
+
+
+def _offload_image_url(url: object, image_dir: Path | None) -> str | None:
+    try:
+        offload_image_to_run_assets = getattr(
+            import_module("renderers.mm_store"),
+            "offload_image_to_run_assets",
+        )
+    except (
+        ImportError,
+        AttributeError,
+    ) as exc:  # pragma: no cover - dependency-version guard
+        raise RuntimeError(
+            "Multimodal training requires a renderers version with raw image asset offload support."
+        ) from exc
+
+    return offload_image_to_run_assets(url, image_dir=image_dir)
+
+
+def _part_image_field(part_type: object) -> str | None:
+    """The field carrying a content part's image source, keyed by ``type``.
+
+    Mirrors the renderer-side part treaty (``renderers.qwen3_vl._image_source``):
+    ``image_url`` parts nest the URL under ``image_url`` (``{"url": ...}`` or a
+    direct string), ``image`` parts carry the URL string directly.
+    """
+    if part_type in ("image_url", "image"):
+        return str(part_type)
+    return None
+
+
+def _get_field(container: Any, name: str) -> object:
+    if isinstance(container, dict):
+        return container.get(name)
+    return getattr(container, name, None)
+
+
+def _set_field(container: Any, name: str, value: str) -> None:
+    if isinstance(container, dict):
+        container[name] = value
+    else:
+        setattr(container, name, value)
+
+
+def _prepare_image_part(part: Any, field: str, *, image_dir: Path | None) -> None:
+    """Offload one image part's source to run assets and require ``file://``."""
+    source = _get_field(part, field)
+    if source is None:
+        return
+    if isinstance(source, str):
+        container, key = part, field
+    elif field == "image_url":
+        container, key = source, "url"
+    else:
+        raise RuntimeError(
+            f"multimodal training requires string image sources; got {type(source).__name__} under {field!r}"
+        )
+
+    url = _get_field(container, key)
+    offloaded = _offload_image_url(url, image_dir)
+    if offloaded is not None:
+        _set_field(container, key, offloaded)
+        url = offloaded
+    if not isinstance(url, str) or not url.startswith("file://"):
+        raise RuntimeError(
+            "multimodal training requires image sources offloaded to file:// "
+            f"run image assets; got {url!r} under {field!r}"
+        )
+
+
+def prepare_images_inplace(value: Any, *, image_dir: Path | None = None) -> None:
+    """Offload image URLs reachable from ``value`` to run image assets.
+
+    Handles OpenAI wire dicts/lists and the pydantic v0/v1 message/content-part
+    models used by trajectories and traces.
+    """
+    if isinstance(value, dict):
+        field = _part_image_field(value.get("type"))
+        if field is not None:
+            _prepare_image_part(value, field, image_dir=image_dir)
+        for child in value.values():
+            prepare_images_inplace(child, image_dir=image_dir)
+        return
+
+    if isinstance(value, (list, tuple)):
+        for child in value:
+            prepare_images_inplace(child, image_dir=image_dir)
+        return
+
+    field = _part_image_field(getattr(value, "type", None))
+    if field is not None:
+        _prepare_image_part(value, field, image_dir=image_dir)
+        return
+
+    content = getattr(value, "content", None)
+    if isinstance(content, (list, tuple)):
+        prepare_images_inplace(content, image_dir=image_dir)

verifiers/v1/ARCHITECTURE.md

@@ -70,12 +70,16 @@ end to end: each surviving context window is just another root→leaf path.
 
 `Trace.to_record()` (`trace.py`) is the JSON record dump (`model_dump(mode="json")`) for
 `results.jsonl` / W&B tables, minus the per-node training tensors (`MessageNode.multi_modal_data`,
-`routed_experts`, via `_NODE_DUMP_EXCLUDE`): those hold raw numpy bytes that can't round-trip JSON
-(the dump raises `UnicodeDecodeError` on real expert ids) and bloat every line. Computed views
+`routed_experts`, via `_NODE_DUMP_EXCLUDE`): routed-expert tensors hold raw numpy bytes that can't
+round-trip JSON (the dump raises `UnicodeDecodeError` on real expert ids), and multimodal
+descriptors are trainer sidecars rather than rollout records. Computed views
 (`reward`, `branches`, `num_turns`, per-span `duration`) are pydantic properties, so they're never
 serialized and recompute on load; `state` is excluded. The tensors still reach the trainer over the
 env-server *wire*, which uses msgpack `model_dump(mode="python")` and carries them as raw `bin` bytes
-(not base64) via the field serializers on `MessageNode` (`graph.py`); only the JSON record strips them.
+(not base64) via the field serializers on `MessageNode` (`graph.py`); only the JSON record strips
+them. Multimodal training uses raw run-image assets: the train client rewrites base64 image parts to
+`file://` refs before tracing, and `MessageNode.multi_modal_data` carries lightweight renderer
+descriptors (hashes, placeholder ranges, image metadata/refs) rather than image processor outputs.
 
 ### Branching: message-level vs renderer-level, and the token invariant
 
@@ -111,9 +115,10 @@ The renderer client avoids the break entirely when it can: instead of re-renderi
 each turn, the train client (`clients/train.py`) calls `renderer.bridge_to_next_turn(...)`, which
 keeps the prior `prompt_ids + completion_ids` **verbatim** and only renders the new tail. Verbatim
 prior ⇒ the stored prefix matches token-for-token ⇒ no fork, one linear branch, invariant intact.
-The token-identity check in `commit` is the backstop for when the bridge can't apply (the renderer
-returns `None`, multimodal, the eval relay): the break still surfaces as honest branches rather than
-silent corruption.
+For multimodal renderers, the train client also passes the reusable prefix's `multi_modal_data` so
+prior image placeholders and descriptors remain aligned. The token-identity check in `commit` is the
+backstop for when the bridge can't apply (the renderer returns `None`, the eval relay): the break
+still surfaces as honest branches rather than silent corruption.
 
 ## Model access — interception, dialects, clients
 

verifiers/v1/clients/client.py

@@ -33,6 +33,19 @@ class RelayReply:
 
 
 class Client(ABC):
+    async def prepare_request_body(self, dialect: Dialect, body: dict) -> dict:
+        """Normalize a provider request before the interception server parses/traces it.
+
+        Relay clients keep the request verbatim. Training clients may rewrite heavy
+        in-process payloads (for example base64 images) into stable run-asset refs so the
+        trace, renderer, and trainer all see the same cheap message content.
+        """
+        return body
+
+    async def prepare_messages(self, dialect: Dialect, messages: list) -> list:
+        """Normalize typed simulator messages before adding them to the wire body/trace."""
+        return messages
+
     @abstractmethod
     async def get_response(
         self,

verifiers/v1/clients/train.py

@@ -8,6 +8,7 @@
 needs a running vLLM engine.
 """
 
+import asyncio
 import json
 from collections.abc import Mapping
 from typing import Any
@@ -16,6 +17,7 @@
 from renderers import RenderedTokens
 from renderers import OverlongPromptError as RendererOverlongPromptError
 from renderers import RendererConfig
+from renderers.base import is_multimodal
 
 from verifiers.v1.clients.client import SESSION_ID_HEADER, Client
 from verifiers.v1.dialects import FINISH_REASONS, ChatDialect, Dialect, parse_tools
@@ -32,6 +34,7 @@
     TurnTokens,
     Usage,
 )
+from verifiers.utils.multimodal import prepare_images_inplace
 
 
 def tool_to_wire(tool: Tool) -> dict:
@@ -167,16 +170,6 @@ def _is_valid_incremental_tail(messages: list[dict[str, Any]]) -> bool:
     return all(role == "tool" for role in roles)
 
 
-def _has_multimodal_content(messages) -> bool:
-    for message in messages:
-        content = getattr(message, "content", None)
-        if not isinstance(content, list):
-            continue
-        if any(getattr(part, "type", None) == "image_url" for part in content):
-            return True
-    return False
-
-
 class TrainClient(Client):
     """Renders prompts to token ids and calls a vLLM `/inference/v1/generate` engine."""
 
@@ -213,6 +206,16 @@ def _renderer_pool(
             )
         return self._pool
 
+    async def prepare_request_body(self, dialect: Dialect, body: dict) -> dict:
+        if isinstance(dialect, ChatDialect):
+            await asyncio.to_thread(prepare_images_inplace, body)
+        return body
+
+    async def prepare_messages(self, dialect: Dialect, messages: list) -> list:
+        if isinstance(dialect, ChatDialect):
+            await asyncio.to_thread(prepare_images_inplace, messages)
+        return messages
+
     async def get_response(
         self,
         dialect: Dialect,
@@ -263,23 +266,24 @@ async def get_response(
         )
         bridged_turn: PendingTurn | None = None
 
-        # Only build the (O(context)) previous-turn token ids once the cheap guards pass — a
-        # multimodal prompt or a tail that isn't a clean `[tool*, user?]` extension can't bridge.
-        can_bridge = (
-            turn is not None
-            and not _has_multimodal_content(prompt)
-            and _is_valid_incremental_tail(wire_messages)
-        )
+        # Only build the (O(context)) previous-turn token ids once the cheap guards pass: a
+        # tail that isn't a clean `[tool*, user?]` extension can't bridge.
+        can_bridge = turn is not None and _is_valid_incremental_tail(wire_messages)
         previous_ids = turn.previous_token_ids() if can_bridge else None
         if previous_ids is not None:
             previous_prompt_ids, previous_completion_ids = previous_ids
 
             def bridge():
+                kwargs: dict[str, Any] = {"tools": wire_tools}
+                if is_multimodal(renderer):
+                    kwargs["previous_multi_modal_data"] = (
+                        turn.previous_multi_modal_data()
+                    )
                 return renderer.bridge_to_next_turn(
                     previous_prompt_ids,
                     previous_completion_ids,
                     wire_messages,
-                    tools=wire_tools,
+                    **kwargs,
                 )
 
             bridged = await _maybe_offload(renderer, bridge)