feat(model_3d): add client-side MODEL_3D decoder + capability advertisement

Closes the loop on the just-merged comfy-rnp-server PR #37 which added
the four Tripo / Rodin 3D providers and the
``Capability.MODEL_3D_GLB_INLINE`` token. Without this client-side half
the providers were unreachable from any RNP client — descriptors would
fail negotiation because the proxy_node didn't advertise the token.

Client changes (all inside
``ComfyUI/custom_nodes/comfy_remote_nodes``):

* ``protocol.py`` — mirror the new ``Capability.MODEL_3D_GLB_INLINE``
  constant and add ``"model_3d"`` to ``HEAVY_TYPES`` so ``is_envelope``
  recognises inbound 3D-model envelopes. The vendored ``protocol``
  module is the single source of truth for the proxy_node and
  serialization layers; both already consume ``is_envelope`` /
  ``decode_envelope_data`` from here.
* ``serialization.py`` — add ``decode_model3d_envelope`` that hands
  raw GLB bytes to ``comfy_api.latest._util.File3D`` via ``BytesIO``,
  mirroring the existing ``decode_video_envelope`` shape. Only
  ``glb_inline`` is accepted today; any other encoding surfaces
  ``RnpProtocolError(code=INTERNAL)`` as defence-in-depth so a future
  encoding never silently downgrades to garbage bytes. Registered in
  the ``_DECODERS`` dispatch table and re-exported via ``__all__``.
* ``proxy_node.py`` — add ``"MODEL_3D" -> IO.File3DGLB.Output`` to
  ``_OUTPUT_CLASSES`` so descriptors with ``output=["MODEL_3D"]``
  parse to a real socket. The descriptor io_type string is the logical
  name ``"MODEL_3D"`` (consistent across vendors) — the resulting V3
  Output carries the upstream ``"FILE_3D_GLB"`` comfytype tag so
  frontend-side socket matching works against any GLB-aware downstream
  node (preview3d / SaveGLB / Load3D).
* ``client.py`` — advertise ``Capability.MODEL_3D_GLB_INLINE`` in
  ``CLIENT_CAPABILITIES`` so servers gate provider descriptors on the
  presence of the token. Legacy clients that don't advertise it still
  see NEGOTIATION_FAILED at descriptor-load time (safe degradation),
  not a parse error at execute.

Smoke test (``notes/run_model3d_decoder.py``) covers:
1. ``HEAVY_TYPES`` includes ``"model_3d"``.
2. ``is_envelope`` recognises a MODEL_3D envelope.
3. ``decode_model3d_envelope`` round-trips a minimal valid GLB
   (binary glTF 2.0, magic ``b"glTF"``).
4. ``decode_envelope`` dispatches ``type=model_3d`` to the new decoder.
5. Unsupported encoding raises ``RnpProtocolError``.
6. End-to-end against the server's ``make_model3d_envelope`` helper —
   server packs, client decodes, bytes match.
7. ``proxy_node._OUTPUT_CLASSES`` contract:
   ``"MODEL_3D" -> IO.File3DGLB.Output``.
8. ``client.CLIENT_CAPABILITIES`` advertises
   ``Capability.MODEL_3D_GLB_INLINE``.

No server-side changes. The 4 just-shipped Tripo / Rodin providers
(``TripoTextToModelApi_RNP``, ``TripoImageToModelApi_RNP``,
``TripoMultiviewToModelApi_RNP``, ``Rodin3DRegularApi_RNP``) are now
reachable end-to-end from ComfyUI.

Co-authored-by: Amp <amp@ampcode.com>
Amp-Thread-ID: https://ampcode.com/threads/T-019e3314-4336-71d2-a333-6d1e33a9986e

client.py

@@ -28,25 +28,69 @@ from .protocol import (
 log = logging.getLogger("comfy_remote_nodes.client")
 
 
-# Capability tokens advertised on every outbound request. Covers every
-# input/output encoding ``serialization.py`` knows how to round-trip,
-# plus the ``schema:v3`` marker so descriptors come back in the V3
-# ``get_v1_info()`` shape. Audio rides as ``mp3_base64`` outbound (the
-# encoder always re-encodes a ComfyUI AUDIO dict via
-# ``audio_input_to_mp3``) and decodes any of mp3_base64/mp3_inline/
-# wav_base64/wav_inline inbound — Stability returns wav_base64 by
-# default.
+# Capability tokens advertised on every outbound request. Currently
+# covers the input/output encodings the registered providers (LTXV,
+# Gemini, Ideogram, ...) need, plus the ``schema:v3`` marker so
+# descriptors come back in the V3 ``get_v1_info()`` shape. Servers use
+# this list to negotiate per-request: e.g. a provider only emits the
+# multi-frame ``image:png_base64_batch`` envelope when this list
+# contains :data:`Capability.IMAGE_PNG_BASE64_BATCH`, otherwise it
+# falls back to the single-frame ``image:png_base64`` shape (with a
+# server-side truncation warning logged when extras are dropped).
 CLIENT_CAPABILITIES = [
     Capability.SCHEMA_V3,
     Capability.VIDEO_MP4_INLINE,
     Capability.VIDEO_MP4_BASE64,
     Capability.IMAGE_PNG_BASE64,
+    Capability.IMAGE_PNG_BASE64_BATCH,
+    # Advertises that this client can emit a multi-frame IMAGE
+    # envelope with per-frame presigned-GET URIs instead of inline
+    # base64 frames when the summed batch payload would overflow the
+    # 8 MiB inline cap (see ``serialization.maybe_externalize``).
+    # Opportunistic: the client falls back to it whenever a batch
+    # envelope it produced exceeds the cap, so descriptors don't
+    # need to publish it in their ``input_serialization`` to
+    # benefit — any input that already accepts ``png_base64_batch``
+    # transparently accepts the externalised variant on the server
+    # via the shared ``resolve_image_envelope_pngs`` dispatch.
+    Capability.IMAGE_PNG_BASE64_BATCH_URI,
     Capability.MASK_PNG_BASE64,
+    Capability.AUDIO_MP3_INLINE,
     Capability.AUDIO_MP3_BASE64,
     Capability.AUDIO_WAV_BASE64,
-    Capability.AUDIO_MP3_INLINE,
     Capability.AUDIO_WAV_INLINE,
-    Capability.SVG_XML_BASE64,
+    # OGG/Opus decode path. PyAV / FFmpeg handle OGG/Opus natively in
+    # ``audio_bytes_to_audio_input`` so the proxy_node side is the
+    # same as MP3/WAV — advertising the token unlocks per-provider
+    # ``output_format`` choices that emit Opus (e.g. ElevenLabs'
+    # ``opus_48000_192``) without requiring servers to label-launder
+    # the upstream container as ``mp3_base64``.
+    Capability.AUDIO_OPUS_BASE64,
+    # Advertises that this client can drive the
+    # ``execute_async`` + poll-loop lifecycle (submit, poll, cancel,
+    # TASK_LOST replay). Servers gate ``async_polling``-mode
+    # descriptors on the presence of this token so an older client
+    # that only knows ``/execute`` doesn't silently submit a
+    # long-poll node into the sync endpoint.
+    Capability.ASYNC_EXECUTE,
+    # Generic opaque pass-through bucket for partner helper-config
+    # custom IO types (OPENAI_INPUT_FILES, OPENAI_CHAT_CONFIG,
+    # GEMINI_INPUT_FILES, RECRAFT_*, future partner helper types).
+    # Servers may gate descriptors that publish ``input_serialization``
+    # / ``output_serialization`` entries with the ``"opaque"`` marker
+    # on the presence of this token; legacy clients that don't know
+    # about IO.Custom fallthrough get filtered out before they try to
+    # parse a descriptor whose io_type strings they wouldn't recognise.
+    Capability.IO_OPAQUE,
+    # 3D MODEL envelope (binary glTF 2.0 — single-mesh GLB). The
+    # decoder in ``serialization.decode_model3d_envelope`` hands raw
+    # GLB bytes to ``comfy_api.latest._util.File3D`` so downstream
+    # nodes (preview3d / SaveGLB / Load3D) see the same value shape
+    # they would from a local Tripo / Rodin / Meshy node. Servers
+    # gate ``MODEL_3D`` provider descriptors on this token so a
+    # legacy client without the decoder surfaces NEGOTIATION_FAILED
+    # at descriptor-load time rather than crashing at execute.
+    Capability.MODEL_3D_GLB_INLINE,
 ]
 
 # Reported in the X-RNP-Client-Version header.

notes/run_batch_uri_followups.py

@@ -0,0 +1,352 @@
+"""Smoke test for the three batch-URI followups (PR-2/PR-3 cleanup):
+
+1. ``serialization.maybe_externalize`` keys batch dispatch off the
+   ``encoding`` string (``png_base64_batch``) — NOT field presence
+   (``frames``).
+2. ``serialization.maybe_externalize`` rejects an empty/missing
+   ``frames`` list on a ``png_base64_batch`` envelope as a hard
+   ``RnpProtocolError`` rather than silently no-op-ing.
+3. ``proxy_node._deserialize_output`` accepts the externalised
+   ``png_base64_batch_uri`` shape on the response path: fetches every
+   ``uri`` (in parallel via ``asyncio.gather``), rewrites to inline
+   ``frames``, and decodes into a ``[B,H,W,C]`` tensor with frame
+   ordering preserved.
+
+Run with any python that has ``torch`` and ``Pillow`` available:
+
+    python notes/run_batch_uri_followups.py
+"""
+from __future__ import annotations
+
+import asyncio
+import base64
+import http.server
+import importlib.util
+import io
+import os
+import socketserver
+import sys
+import threading
+import time
+import types
+
+import numpy as np
+import torch
+from PIL import Image
+
+# ---------------------------------------------------------------------------
+# Locate the worktree under test.
+# ---------------------------------------------------------------------------
+_HERE = os.path.dirname(os.path.abspath(__file__))
+_CLIENT_DIR = os.path.dirname(_HERE)
+
+
+def _load_module(name: str, path: str) -> types.ModuleType:
+    spec = importlib.util.spec_from_file_location(name, path)
+    mod = importlib.util.module_from_spec(spec)
+    sys.modules[name] = mod
+    spec.loader.exec_module(mod)
+    return mod
+
+
+# Stub out client.upload_asset so the dispatch tests don't need a real
+# RNP server — the externalize batch path will call this once per frame.
+_uploads: list[bytes] = []
+
+
+async def _fake_upload_asset(server_url, raw, *, content_type, auth_headers=None):  # noqa: ANN001
+    _uploads.append(bytes(raw))
+    return f"https://fake-storage/{len(_uploads) - 1}"
+
+
+def _install_fake_client() -> None:
+    fake_client = types.ModuleType("comfy_remote_nodes.client")
+    fake_client.upload_asset = _fake_upload_asset
+    sys.modules["comfy_remote_nodes.client"] = fake_client
+
+
+def _png_b64_for_color(color: tuple[int, int, int]) -> str:
+    img = Image.new("RGB", (4, 4), color=color)
+    buf = io.BytesIO()
+    img.save(buf, format="PNG")
+    return base64.b64encode(buf.getvalue()).decode("ascii")
+
+
+def _png_bytes_for_color(color: tuple[int, int, int]) -> bytes:
+    img = Image.new("RGB", (4, 4), color=color)
+    buf = io.BytesIO()
+    img.save(buf, format="PNG")
+    return buf.getvalue()
+
+
+# ---------------------------------------------------------------------------
+# Test 1+2: maybe_externalize dispatch + empty-frames
+# ---------------------------------------------------------------------------
+
+async def _test_maybe_externalize() -> list[str]:
+    failures: list[str] = []
+
+    # Re-import serialization fresh because it caches a `client` ref.
+    sys.modules.pop("comfy_remote_nodes", None)
+    sys.modules.pop("comfy_remote_nodes.serialization", None)
+    sys.modules.pop("comfy_remote_nodes.protocol", None)
+
+    # Build a minimal package shim so `from . import client` works.
+    pkg = types.ModuleType("comfy_remote_nodes")
+    pkg.__path__ = [_CLIENT_DIR]
+    sys.modules["comfy_remote_nodes"] = pkg
+    _install_fake_client()
+    protocol = _load_module(
+        "comfy_remote_nodes.protocol",
+        os.path.join(_CLIENT_DIR, "protocol.py"),
+    )
+    serialization = _load_module(
+        "comfy_remote_nodes.serialization",
+        os.path.join(_CLIENT_DIR, "serialization.py"),
+    )
+
+    # --- Case A: batch envelope with payload bigger than cap → externalises
+    _uploads.clear()
+    big_frames = [_png_b64_for_color((i * 30, 0, 0)) for i in range(4)]
+    env = {
+        "type": "image",
+        "encoding": "png_base64_batch",
+        "frames": big_frames,
+        "shape": [4, 4, 4, 3],
+        "dtype": "uint8",
+        "color_space": "srgb",
+    }
+    out = await serialization.maybe_externalize(
+        env, server_url="http://fake", max_inline_bytes=1, auth_headers={},
+    )
+    if out.get("encoding") != "png_base64_batch_uri":
+        failures.append(
+            f"A: expected encoding=png_base64_batch_uri, got {out.get('encoding')!r}"
+        )
+    if "frames" in out:
+        failures.append("A: 'frames' field still present after externalise")
+    if not isinstance(out.get("uris"), list) or len(out["uris"]) != 4:
+        failures.append(f"A: expected 4 uris, got {out.get('uris')!r}")
+    if len(_uploads) != 4:
+        failures.append(f"A: expected 4 uploads, got {len(_uploads)}")
+
+    # --- Case B: dispatch keys off encoding, NOT field presence.
+    # An envelope with a stray `frames` field but a non-batch encoding
+    # must NOT be treated as a batch (would have under the old
+    # `isinstance(frames, list) and frames` dispatch).
+    _uploads.clear()
+    rogue = {
+        "type": "image",
+        "encoding": "png_base64",  # singleton encoding
+        "data": _png_b64_for_color((10, 20, 30)) * 200,  # > cap
+        "frames": ["bogus-extra-field"],
+    }
+    out_b = await serialization.maybe_externalize(
+        rogue, server_url="http://fake", max_inline_bytes=1, auth_headers={},
+    )
+    if out_b.get("encoding") == "png_base64_batch_uri":
+        failures.append(
+            "B: dispatch took batch branch on a singleton encoding (rogue frames field)"
+        )
+    if "uri" not in out_b:
+        failures.append("B: singleton encoding should have externalised to 'uri'")
+    if len(_uploads) != 1:
+        failures.append(f"B: expected 1 upload (singleton path), got {len(_uploads)}")
+
+    # --- Case C: empty frames on a batch encoding is a hard error.
+    bad = {
+        "type": "image",
+        "encoding": "png_base64_batch",
+        "frames": [],
+        "shape": [0, 4, 4, 3],
+    }
+    try:
+        await serialization.maybe_externalize(
+            bad, server_url="http://fake", max_inline_bytes=1, auth_headers={},
+        )
+    except protocol.RnpProtocolError:
+        pass
+    else:
+        failures.append("C: empty 'frames' on batch encoding did not raise RnpProtocolError")
+
+    # --- Case D: under-cap batch: no upload, envelope unchanged.
+    _uploads.clear()
+    small = {
+        "type": "image",
+        "encoding": "png_base64_batch",
+        "frames": big_frames,
+        "shape": [4, 4, 4, 3],
+    }
+    out_d = await serialization.maybe_externalize(
+        small,
+        server_url="http://fake",
+        max_inline_bytes=10 * 1024 * 1024,
+        auth_headers={},
+    )
+    if out_d is not small and out_d != small:
+        failures.append("D: under-cap batch should be a no-op")
+    if _uploads:
+        failures.append(f"D: under-cap batch should not upload, got {len(_uploads)}")
+
+    return failures
+
+
+# ---------------------------------------------------------------------------
+# Test 3: _deserialize_output handles png_base64_batch_uri on response
+# ---------------------------------------------------------------------------
+
+DELAY_S = 0.4
+N_FRAMES = 4
+
+
+class _SlowPngHandler(http.server.BaseHTTPRequestHandler):
+    def do_GET(self) -> None:  # noqa: N802
+        time.sleep(DELAY_S)
+        try:
+            idx = int(self.path.rsplit("/", 1)[-1])
+        except ValueError:
+            self.send_response(404); self.end_headers(); return
+        # Distinct color per frame so we can verify ordering after decode.
+        body = _png_bytes_for_color((idx * 60 % 256, 80, 120))
+        self.send_response(200)
+        self.send_header("Content-Type", "image/png")
+        self.send_header("Content-Length", str(len(body)))
+        self.end_headers()
+        self.wfile.write(body)
+
+    def log_message(self, *_a, **_kw) -> None:
+        return
+
+
+async def _test_deserialize_output() -> list[str]:
+    failures: list[str] = []
+
+    server = socketserver.ThreadingTCPServer(("127.0.0.1", 0), _SlowPngHandler)
+    host, port = server.server_address
+    th = threading.Thread(target=server.serve_forever, daemon=True)
+    th.start()
+    try:
+        # Stub the proxy_node's _fetch_url_envelope so we don't have to
+        # bring up the whole comfy_api_nodes.util.client stack.
+        import urllib.request as _urlreq
+
+        async def fake_fetch(envelope, policy, *, auth_headers, node_cls):  # noqa: ANN001, ARG001
+            # Real ``_fetch_url_envelope`` is properly async (aiohttp);
+            # the smoke stub uses sync urllib so we hop to a thread to
+            # preserve the parallel-gather semantics under test.
+            uri = envelope["uri"]
+            def _do_fetch() -> bytes:
+                with _urlreq.urlopen(uri, timeout=10.0) as resp:
+                    return resp.read()
+            return await asyncio.to_thread(_do_fetch)
+
+        # Re-import proxy_node helpers via AST to avoid loading ComfyUI.
+        import ast
+
+        with open(os.path.join(_CLIENT_DIR, "proxy_node.py"), "r", encoding="utf-8") as fh:
+            src = fh.read()
+        tree = ast.parse(src)
+        wanted = {"_deserialize_output"}
+        nodes = [n for n in tree.body if isinstance(n, ast.AsyncFunctionDef) and n.name in wanted]
+        if len(nodes) != len(wanted):
+            failures.append(f"missing AST nodes; found {[n.name for n in nodes]}")
+            return failures
+
+        # Re-import the protocol + serialization modules with the package
+        # shim so `from . import` works inside serialization.
+        sys.modules.pop("comfy_remote_nodes.serialization", None)
+        sys.modules.pop("comfy_remote_nodes.protocol", None)
+        pkg = types.ModuleType("comfy_remote_nodes")
+        pkg.__path__ = [_CLIENT_DIR]
+        sys.modules["comfy_remote_nodes"] = pkg
+        _install_fake_client()
+        protocol = _load_module(
+            "comfy_remote_nodes.protocol",
+            os.path.join(_CLIENT_DIR, "protocol.py"),
+        )
+        serialization = _load_module(
+            "comfy_remote_nodes.serialization",
+            os.path.join(_CLIENT_DIR, "serialization.py"),
+        )
+
+        ns: dict = {
+            "Any": object,
+            "RnpProtocolError": protocol.RnpProtocolError,
+            "ErrorCode": protocol.ErrorCode,
+            "is_envelope": protocol.is_envelope,
+            "serialization": serialization,
+            "_fetch_url_envelope": fake_fetch,
+            "log": types.SimpleNamespace(warning=lambda *a, **kw: None),
+        }
+        mod_src = ast.Module(body=nodes, type_ignores=[])
+        exec(compile(mod_src, "<batch_uri_smoke>", "exec"), ns)
+        deserialize = ns["_deserialize_output"]
+
+        # Build the externalised batch envelope on the response path.
+        uris = [f"http://{host}:{port}/frame/{i}" for i in range(N_FRAMES)]
+        envelope = {
+            "type": "image",
+            "encoding": "png_base64_batch_uri",
+            "uris": uris,
+            "shape": [N_FRAMES, 4, 4, 3],
+            "dtype": "uint8",
+            "color_space": "srgb",
+        }
+        url_fetch_policy = {"image": {"url_kind": "presigned", "timeout_s": 10}}
+
+        t0 = time.monotonic()
+        tensor = await deserialize(
+            envelope, meta=None,
+            url_fetch_policy=url_fetch_policy,
+            server_url="http://fake",
+            auth_headers={},
+            node_cls=None,
+        )
+        elapsed = time.monotonic() - t0
+
+        if not isinstance(tensor, torch.Tensor):
+            failures.append(f"expected torch.Tensor, got {type(tensor).__name__}")
+            return failures
+        if tuple(tensor.shape) != (N_FRAMES, 4, 4, 3):
+            failures.append(f"shape {tuple(tensor.shape)} != ({N_FRAMES}, 4, 4, 3)")
+
+        # Each frame's R-channel should match the expected color.
+        for i in range(N_FRAMES):
+            expected_r = (i * 60 % 256) / 255.0
+            actual_r = float(tensor[i, 0, 0, 0])
+            if abs(actual_r - expected_r) > 1e-3:
+                failures.append(
+                    f"frame {i}: R channel {actual_r:.3f} != expected {expected_r:.3f} "
+                    f"(ordering broken or wrong fetch)"
+                )
+
+        # Wall-clock: parallel should be ~DELAY_S, sequential would be ~N*DELAY_S.
+        sequential_lower = N_FRAMES * DELAY_S * 0.7
+        if elapsed >= sequential_lower:
+            failures.append(
+                f"wall clock {elapsed:.2f}s >= sequential lower bound "
+                f"{sequential_lower:.2f}s — gather not running in parallel"
+            )
+
+        return failures
+    finally:
+        server.shutdown(); server.server_close()
+
+
+def main() -> int:
+    failures: list[str] = []
+    failures += asyncio.run(_test_maybe_externalize())
+    failures += asyncio.run(_test_deserialize_output())
+    if failures:
+        print()
+        print("FAIL:")
+        for f in failures:
+            print(" -", f)
+        return 1
+    print("PASS: maybe_externalize dispatches on encoding + rejects empty frames; "
+          "_deserialize_output handles png_base64_batch_uri in parallel")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

notes/run_image_max_batch.py

@@ -0,0 +1,441 @@
+"""Smoke test for the descriptor-driven ``local_validate.image_max_batch``
+client-side fail-fast.
+
+Exercises the proxy_node's ``_collect_local_validate`` /
+``_enforce_local_validate`` pair against a fake descriptor that mirrors
+the Flux2 wire shape. Asserts that:
+
+1. ``_collect_local_validate`` extracts ``{"images": {"image_max_batch": 2}}``
+   from the descriptor's IMAGE-input options dict.
+2. Encoding a 3-frame tensor for that input raises ``RnpProtocolError``
+   with ``code=INPUT_INVALID`` and ``user_facing=True`` *before* any
+   serialization / HTTP attempt.
+3. A 1-frame tensor passes through cleanly.
+4. A descriptor without ``local_validate`` is a no-op.
+5. ``cap=1`` surfaces the single-frame copy.
+6. A ``local_validate`` block missing the ``image_max_batch`` key is a no-op.
+
+Run with any python that has ``torch`` available:
+
+    python notes/run_image_max_batch.py
+
+The test loads the helper functions out of ``proxy_node.py`` via AST
+extraction so it doesn't pay the cost of importing ComfyUI's
+``comfy_api`` / ``comfy_api_nodes`` packages (which the helper
+functions don't touch). Standalone — no test runner / fixtures needed.
+"""
+from __future__ import annotations
+
+import ast
+import asyncio
+import importlib.util
+import os
+import sys
+import types
+
+import torch
+
+# ---------------------------------------------------------------------------
+# Locate the worktree under test (this file's parent's parent).
+# ---------------------------------------------------------------------------
+
+_HERE = os.path.dirname(os.path.abspath(__file__))
+_CLIENT_DIR = os.path.dirname(_HERE)
+
+
+def _load_protocol() -> types.ModuleType:
+    """Load ``protocol.py`` standalone (no relative imports)."""
+    path = os.path.join(_CLIENT_DIR, "protocol.py")
+    spec = importlib.util.spec_from_file_location("rnp_smoke_protocol", path)
+    mod = importlib.util.module_from_spec(spec)
+    sys.modules[spec.name] = mod
+    spec.loader.exec_module(mod)
+    return mod
+
+
+def _build_serialization_stub(protocol: types.ModuleType) -> types.ModuleType:
+    """Minimal ``serialization`` shim covering exactly the surface
+    ``_encode_one`` / ``_enforce_local_validate`` / ``_encode_inputs`` use:
+
+    * ``_is_torch_tensor(value)`` — duck-type check for torch tensors.
+    * ``is_audio_input(value)`` — always False (no audio in this test).
+    * ``encode_image_tensor(tensor, *, accepts_batch)`` — fake envelope.
+    * ``encode_mask_tensor(tensor)`` — fake envelope.
+    * ``is_envelope(value)`` — False so ``maybe_externalize`` is skipped.
+    """
+    mod = types.ModuleType("rnp_smoke_serialization")
+
+    def _is_torch_tensor(value):  # noqa: ANN001
+        try:
+            import torch as _torch
+        except ImportError:
+            return False
+        return isinstance(value, _torch.Tensor)
+
+    def is_audio_input(value):  # noqa: ANN001
+        return isinstance(value, dict) and {"waveform", "sample_rate"} <= set(value.keys())
+
+    def encode_image_tensor(tensor, *, accepts_batch=False):  # noqa: ANN001, ARG001
+        return {"type": "image", "encoding": "png_base64", "frames": int(tensor.shape[0])}
+
+    def encode_mask_tensor(tensor):  # noqa: ANN001, ARG001
+        return {"type": "mask", "encoding": "png_base64"}
+
+    def encode_audio_input(value):  # noqa: ANN001, ARG001
+        return {"type": "audio", "encoding": "mp3_base64"}
+
+    def is_envelope(value):  # noqa: ANN001
+        return False  # smoke test never externalizes
+
+    mod._is_torch_tensor = _is_torch_tensor
+    mod.is_audio_input = is_audio_input
+    mod.encode_image_tensor = encode_image_tensor
+    mod.encode_mask_tensor = encode_mask_tensor
+    mod.encode_audio_input = encode_audio_input
+    mod.is_envelope = is_envelope
+
+    async def maybe_externalize(*args, **kwargs):  # pragma: no cover — unused
+        return args[0] if args else None
+    mod.maybe_externalize = maybe_externalize
+    return mod
+
+
+def _build_proxy_node_stub(
+    protocol: types.ModuleType, serialization: types.ModuleType,
+) -> types.ModuleType:
+    """Extract the four helper functions under test from ``proxy_node.py``
+    and exec them in a minimal namespace so we don't drag the rest of
+    the module's heavy ComfyUI imports into the test process.
+    """
+    mod = types.ModuleType("rnp_smoke_proxy_node")
+    mod.__dict__.update({
+        "Any": object,
+        "log": types.SimpleNamespace(warning=lambda *a, **kw: None),
+        "serialization": serialization,
+        "RnpProtocolError": protocol.RnpProtocolError,
+        "ErrorCode": protocol.ErrorCode,
+    })
+    src_path = os.path.join(_CLIENT_DIR, "proxy_node.py")
+    with open(src_path, "r", encoding="utf-8") as fh:
+        src = fh.read()
+    tree = ast.parse(src)
+    wanted = {
+        "_collect_local_validate",
+        "_rules_from_input_spec",
+        "_enforce_local_validate",
+        "_check_image_max_batch",
+        "_encode_one",
+        "_encode_inputs",
+    }
+    nodes = [
+        n for n in tree.body
+        if isinstance(n, (ast.AsyncFunctionDef, ast.FunctionDef)) and n.name in wanted
+    ]
+    snippet = ast.Module(body=nodes, type_ignores=[])
+    code = compile(snippet, src_path, "exec")
+    exec(code, mod.__dict__)
+    missing = wanted - set(mod.__dict__)
+    if missing:
+        raise RuntimeError(f"proxy_node helpers missing from extract: {missing}")
+    return mod
+
+
+# ---------------------------------------------------------------------------
+# Test fixtures
+# ---------------------------------------------------------------------------
+
+def _flux2_like_descriptor(cap: int) -> dict:
+    """Mirrors the Flux2 ``images`` input descriptor surface."""
+    return {
+        "name": "FakeFlux2_RNP",
+        "input": {
+            "required": {
+                "prompt": ["STRING", {"multiline": True, "default": ""}],
+            },
+            "optional": {
+                "images": ["IMAGE", {
+                    "tooltip": "Up to N images to be used as references.",
+                    "local_validate": {"image_max_batch": cap},
+                }],
+            },
+            "hidden": {},
+        },
+    }
+
+
+def _no_validate_descriptor() -> dict:
+    return {
+        "name": "FakeNoValidate_RNP",
+        "input": {
+            "required": {},
+            "optional": {
+                "images": ["IMAGE", {"tooltip": "no cap"}],
+            },
+            "hidden": {},
+        },
+    }
+
+
+def _autogrow_descriptor(cap: int, prefix: str = "reference") -> dict:
+    """Mirrors GrokVideoReferenceNode_RNP's top-level Autogrow input
+    after server descriptor unwraps the nested DynamicCombo branch
+    (kept simpler here for direct AUTOGROW coverage)."""
+    return {
+        "name": "FakeAutogrow_RNP",
+        "input": {
+            "required": {
+                "reference_images": ["AUTOGROW", {
+                    "template": ["IMAGE", {
+                        "local_validate": {"image_max_batch": cap},
+                    }],
+                    "prefix": prefix,
+                    "min": 1,
+                    "max": cap,
+                }],
+            },
+            "optional": {},
+            "hidden": {},
+        },
+    }
+
+
+def _dyncombo_with_autogrow_descriptor() -> dict:
+    """Mirrors GrokImageEditNodeV2_RNP's ``model`` DynamicCombo with
+    two branches each carrying an Autogrow ``images`` template — one
+    capped at 3, the other at 1."""
+    def branch(key: str, cap: int) -> dict:
+        return {
+            "key": key,
+            "inputs": [
+                ["images", "AUTOGROW", {
+                    "template": ["IMAGE", {
+                        "local_validate": {"image_max_batch": cap},
+                    }],
+                    "prefix": "image",
+                    "min": 1,
+                    "max": cap,
+                }],
+                ["resolution", "COMBO", {"options": ["1K", "2K"]}],
+            ],
+        }
+    return {
+        "name": "FakeDynComboAutogrow_RNP",
+        "input": {
+            "required": {
+                "model": ["DYNAMIC_COMBO", {
+                    "options": [
+                        branch("grok-imagine-image-quality", 3),
+                        branch("grok-imagine-image-pro", 1),
+                    ],
+                }],
+            },
+            "optional": {},
+            "hidden": {},
+        },
+    }
+
+
+# ---------------------------------------------------------------------------
+# Tests
+# ---------------------------------------------------------------------------
+
+async def main() -> int:
+    protocol = _load_protocol()
+    serialization = _build_serialization_stub(protocol)
+    proxy_node = _build_proxy_node_stub(protocol, serialization)
+
+    # ---- 1. _collect_local_validate extracts the rule
+    desc = _flux2_like_descriptor(cap=2)
+    rules = proxy_node._collect_local_validate(desc)
+    assert rules == {"images": {"image_max_batch": 2}}, rules
+    print("ok: _collect_local_validate extracted", rules)
+
+    # ---- 2. over-cap tensor raises INPUT_INVALID before encoding
+    over_cap = torch.zeros((3, 64, 64, 3), dtype=torch.float32)
+    raised = False
+    try:
+        await proxy_node._encode_inputs(
+            {"images": over_cap},
+            input_serialization={"images": ["png_base64", "png_base64_batch"]},
+            local_validate=rules,
+            node_id="FakeFlux2_RNP",
+        )
+    except protocol.RnpProtocolError as e:
+        raised = True
+        assert e.code == protocol.ErrorCode.INPUT_INVALID, e.code
+        assert e.user_facing is True, e.user_facing
+        msg = str(e)
+        assert "FakeFlux2_RNP" in msg, msg
+        assert "3" in msg and "2" in msg, msg
+        print("ok: over-cap raised INPUT_INVALID:", msg)
+    assert raised, "over-cap tensor did not raise"
+
+    # ---- 3. at-cap tensor passes through (no exception)
+    at_cap = torch.zeros((2, 64, 64, 3), dtype=torch.float32)
+    out = await proxy_node._encode_inputs(
+        {"images": at_cap},
+        input_serialization={"images": ["png_base64", "png_base64_batch"]},
+        local_validate=rules,
+        node_id="FakeFlux2_RNP",
+    )
+    assert "images" in out
+    print("ok: at-cap (B=2) encoded without error")
+
+    # ---- 4. cap=1 / multi-frame raises with the single-frame copy
+    desc1 = _flux2_like_descriptor(cap=1)
+    rules1 = proxy_node._collect_local_validate(desc1)
+    raised = False
+    try:
+        await proxy_node._encode_inputs(
+            {"images": over_cap},
+            local_validate=rules1,
+            node_id="FakeI2V_RNP",
+        )
+    except protocol.RnpProtocolError as e:
+        raised = True
+        assert e.code == protocol.ErrorCode.INPUT_INVALID
+        assert "single image" in str(e), str(e)
+        print("ok: cap=1 single-frame copy fired:", str(e))
+    assert raised
+
+    # ---- 5. descriptor without local_validate is a no-op
+    desc_n = _no_validate_descriptor()
+    rules_n = proxy_node._collect_local_validate(desc_n)
+    assert rules_n == {}, rules_n
+    out = await proxy_node._encode_inputs(
+        {"images": over_cap},
+        input_serialization={"images": ["png_base64", "png_base64_batch"]},
+        local_validate=rules_n,
+        node_id="FakeNoValidate_RNP",
+    )
+    assert "images" in out
+    print("ok: missing local_validate is a no-op")
+
+    # ---- 6. local_validate block without image_max_batch sub-key is a no-op
+    out = await proxy_node._encode_inputs(
+        {"images": over_cap},
+        input_serialization={"images": ["png_base64", "png_base64_batch"]},
+        local_validate={"images": {"some_other_rule": 7}},
+        node_id="FakeOther_RNP",
+    )
+    assert "images" in out
+    print("ok: missing image_max_batch sub-key is a no-op")
+
+    # ---- 7. AUTOGROW: each slot value enforced against the template cap
+    desc_ag = _autogrow_descriptor(cap=2)
+    rules_ag = proxy_node._collect_local_validate(desc_ag)
+    assert "reference_images" in rules_ag, rules_ag
+    assert rules_ag["reference_images"].get("__template__") == {"image_max_batch": 2}, rules_ag
+    print("ok: AUTOGROW collected __template__ rules:", rules_ag)
+
+    one_frame = torch.zeros((1, 64, 64, 3), dtype=torch.float32)
+    raised = False
+    try:
+        # Slot 1 is over-cap (B=3 > cap=2) → should fire on that slot
+        await proxy_node._encode_inputs(
+            {"reference_images": {"reference0": one_frame, "reference1": over_cap}},
+            local_validate=rules_ag,
+            node_id="FakeAutogrow_RNP",
+        )
+    except protocol.RnpProtocolError as e:
+        raised = True
+        assert e.code == protocol.ErrorCode.INPUT_INVALID
+        msg = str(e)
+        # Slot path label should mention reference_images.<slot_key>
+        assert "reference_images" in msg, msg
+        assert "reference1" in msg, msg
+        print("ok: AUTOGROW slot over-cap fired:", msg)
+    assert raised, "AUTOGROW slot over-cap did not raise"
+
+    # All-under-cap autogrow passes (no exception)
+    out = await proxy_node._encode_inputs(
+        {"reference_images": {"reference0": one_frame, "reference1": one_frame, "reference2": None}},
+        local_validate=rules_ag,
+        node_id="FakeAutogrow_RNP",
+    )
+    assert "reference_images" in out
+    print("ok: AUTOGROW all-under-cap (None slots tolerated) passed")
+
+    # ---- 8. DYNAMIC_COMBO with nested AUTOGROW: branch-aware enforcement
+    desc_dc = _dyncombo_with_autogrow_descriptor()
+    rules_dc = proxy_node._collect_local_validate(desc_dc)
+    assert "model" in rules_dc, rules_dc
+    branches = rules_dc["model"].get("__branches__")
+    assert isinstance(branches, dict), rules_dc
+    assert "grok-imagine-image-quality" in branches
+    assert "grok-imagine-image-pro" in branches
+    assert branches["grok-imagine-image-quality"]["images"]["__template__"] == {"image_max_batch": 3}
+    assert branches["grok-imagine-image-pro"]["images"]["__template__"] == {"image_max_batch": 1}
+    print("ok: DYNAMIC_COMBO collected branch-keyed templates")
+
+    # Branch A (cap=3): slot with B=4 should raise
+    over_4 = torch.zeros((4, 64, 64, 3), dtype=torch.float32)
+    raised = False
+    try:
+        await proxy_node._encode_inputs(
+            {"model": {
+                "model": "grok-imagine-image-quality",
+                "resolution": "1K",
+                "images": {"image0": over_4},
+            }},
+            local_validate=rules_dc,
+            node_id="FakeDynComboAutogrow_RNP",
+        )
+    except protocol.RnpProtocolError as e:
+        raised = True
+        assert e.code == protocol.ErrorCode.INPUT_INVALID
+        msg = str(e)
+        assert "image0" in msg and "3" in msg and "4" in msg, msg
+        print("ok: DYNAMIC_COMBO branch-A (cap=3) fired on B=4 slot:", msg)
+    assert raised
+
+    # Branch A (cap=3): all slots within cap → passes
+    out = await proxy_node._encode_inputs(
+        {"model": {
+            "model": "grok-imagine-image-quality",
+            "resolution": "2K",
+            "images": {"image0": one_frame, "image1": one_frame, "image2": one_frame},
+        }},
+        local_validate=rules_dc,
+        node_id="FakeDynComboAutogrow_RNP",
+    )
+    assert "model" in out
+    print("ok: DYNAMIC_COMBO branch-A all-under-cap passed")
+
+    # Branch B (cap=1): a B=3 slot fires the single-frame copy
+    raised = False
+    try:
+        await proxy_node._encode_inputs(
+            {"model": {
+                "model": "grok-imagine-image-pro",
+                "resolution": "2K",
+                "images": {"image0": over_cap},
+            }},
+            local_validate=rules_dc,
+            node_id="FakeDynComboAutogrow_RNP",
+        )
+    except protocol.RnpProtocolError as e:
+        raised = True
+        assert e.code == protocol.ErrorCode.INPUT_INVALID
+        assert "single image" in str(e), str(e)
+        print("ok: DYNAMIC_COMBO branch-B (cap=1) fired single-frame copy:", str(e))
+    assert raised
+
+    # Unknown branch key → no enforcement (defensive)
+    out = await proxy_node._encode_inputs(
+        {"model": {
+            "model": "unknown-branch-key",
+            "images": {"image0": over_cap},
+        }},
+        local_validate=rules_dc,
+        node_id="FakeDynComboAutogrow_RNP",
+    )
+    assert "model" in out
+    print("ok: DYNAMIC_COMBO unknown branch key is a defensive no-op")
+
+    print("PASS: image_max_batch local_validate smoke test (incl. nested traversal)")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(asyncio.run(main()))

notes/run_model3d_decoder.py

@@ -0,0 +1,316 @@
+"""Smoke test for the client-side MODEL_3D decoder.
+
+Verifies:
+
+1. The vendored ``protocol.HEAVY_TYPES`` now recognises ``model_3d`` so
+   ``is_envelope`` returns True for an inbound MODEL_3D envelope.
+2. ``serialization.decode_model3d_envelope`` round-trips an inline
+   ``glb_inline`` envelope into a ``comfy_api.latest._util.File3D``
+   whose ``BytesIO`` payload matches the original GLB bytes.
+3. ``decode_envelope`` dispatches ``type=model_3d`` to the new decoder.
+4. The proxy_node ``_OUTPUT_CLASSES`` now maps ``"MODEL_3D"`` to
+   ``IO.File3DGLB.Output`` so a descriptor with ``output=["MODEL_3D"]``
+   parses without falling through to the opaque IO.Custom branch.
+5. ``client.CLIENT_CAPABILITIES`` advertises
+   ``model_3d:glb_inline`` on every outbound request so descriptors
+   gated on the token negotiate cleanly.
+6. End-to-end against the server's ``make_model3d_envelope`` helper
+   (imported from the comfy-rnp-server worktree) — the bytes the
+   server packs are exactly the bytes the client decodes.
+
+Run with the ComfyUI venv (needs torch / comfy_api on PYTHONPATH):
+
+    python notes/run_model3d_decoder.py
+"""
+from __future__ import annotations
+
+import base64
+import importlib.util
+import io
+import json
+import os
+import sys
+import types
+
+# ---------------------------------------------------------------------------
+# Locate the worktree under test.
+# ---------------------------------------------------------------------------
+_HERE = os.path.dirname(os.path.abspath(__file__))
+_CLIENT_DIR = os.path.dirname(_HERE)
+_COMFYUI_DIR = os.path.dirname(os.path.dirname(_CLIENT_DIR))
+_SERVER_DIR = os.path.normpath(
+    os.path.join(_COMFYUI_DIR, "..", "comfy-rnp-server")
+)
+
+
+def _load_module(name: str, path: str) -> types.ModuleType:
+    spec = importlib.util.spec_from_file_location(name, path)
+    mod = importlib.util.module_from_spec(spec)
+    sys.modules[name] = mod
+    spec.loader.exec_module(mod)
+    return mod
+
+
+# Make ComfyUI importable so ``comfy_api.latest._util.File3D`` resolves.
+if _COMFYUI_DIR not in sys.path:
+    sys.path.insert(0, _COMFYUI_DIR)
+# And the server worktree so we can import the ENCODER side end-to-end.
+if _SERVER_DIR not in sys.path:
+    sys.path.insert(0, _SERVER_DIR)
+
+# ComfyUI's full ``comfy_api`` tree pulls torch / PIL / comfy /
+# comfy_execution at import time — way too heavy for a decoder
+# round-trip check. Stub just the leaf module our decoder imports
+# (``comfy_api.latest._util``) with a faithful pure-Python ``File3D``.
+class _StubFile3D:
+    def __init__(self, source, file_format: str = "") -> None:
+        self._source = source
+        self._format = file_format
+
+_comfy_api = types.ModuleType("comfy_api")
+_comfy_api_latest = types.ModuleType("comfy_api.latest")
+_comfy_api_latest_util = types.ModuleType("comfy_api.latest._util")
+_comfy_api_latest_util.File3D = _StubFile3D
+_comfy_api.latest = _comfy_api_latest
+_comfy_api_latest._util = _comfy_api_latest_util
+sys.modules.setdefault("comfy_api", _comfy_api)
+sys.modules.setdefault("comfy_api.latest", _comfy_api_latest)
+sys.modules.setdefault("comfy_api.latest._util", _comfy_api_latest_util)
+
+# Load the client modules as if they were a package — the ``protocol``
+# module imports from ``.protocol`` relative to its package, so we
+# emulate the package layout under the name ``comfy_remote_nodes``.
+pkg = types.ModuleType("comfy_remote_nodes_test")
+pkg.__path__ = [_CLIENT_DIR]
+sys.modules["comfy_remote_nodes_test"] = pkg
+
+protocol = _load_module(
+    "comfy_remote_nodes_test.protocol",
+    os.path.join(_CLIENT_DIR, "protocol.py"),
+)
+# serialization imports ``client`` for ``rnp_client`` but we don't need
+# the HTTP client for the decoder path; provide a tiny stub so the
+# import succeeds.
+client_stub = types.ModuleType("comfy_remote_nodes_test.client")
+sys.modules["comfy_remote_nodes_test.client"] = client_stub
+serialization = _load_module(
+    "comfy_remote_nodes_test.serialization",
+    os.path.join(_CLIENT_DIR, "serialization.py"),
+)
+
+
+# ---------------------------------------------------------------------------
+# Build a minimal valid binary glTF 2.0 (GLB) payload for the round-trip.
+# Format: 12-byte header + N chunks. Header: magic(4) + version(4) + length(4).
+# Magic must be b"glTF", version=2. We include exactly one JSON chunk that
+# describes an empty asset (enough to satisfy any GLB-magic check; downstream
+# nodes do their own validation).
+# ---------------------------------------------------------------------------
+def _make_minimal_glb() -> bytes:
+    json_payload = b'{"asset":{"version":"2.0"}}'
+    # Chunks must be 4-byte aligned; pad JSON with spaces (0x20).
+    pad = (4 - (len(json_payload) % 4)) % 4
+    json_payload = json_payload + (b" " * pad)
+    json_chunk = (
+        len(json_payload).to_bytes(4, "little")  # chunk length
+        + b"JSON"                                  # chunk type
+        + json_payload                             # chunk data
+    )
+    total = 12 + len(json_chunk)
+    header = (
+        b"glTF"
+        + (2).to_bytes(4, "little")
+        + total.to_bytes(4, "little")
+    )
+    return header + json_chunk
+
+
+# ---------------------------------------------------------------------------
+# 1. HEAVY_TYPES recognises model_3d.
+# ---------------------------------------------------------------------------
+assert "model_3d" in protocol.HEAVY_TYPES, (
+    f"HEAVY_TYPES missing 'model_3d': {sorted(protocol.HEAVY_TYPES)}"
+)
+assert protocol.Capability.MODEL_3D_GLB_INLINE == "model_3d:glb_inline", (
+    f"Capability.MODEL_3D_GLB_INLINE wrong value: "
+    f"{protocol.Capability.MODEL_3D_GLB_INLINE!r}"
+)
+print("PASS: HEAVY_TYPES includes 'model_3d'; Capability constant intact.")
+
+
+# ---------------------------------------------------------------------------
+# 2. is_envelope returns True for a MODEL_3D envelope.
+# ---------------------------------------------------------------------------
+glb_bytes = _make_minimal_glb()
+envelope = {
+    "type":     "model_3d",
+    "encoding": "glb_inline",
+    "data":     base64.b64encode(glb_bytes).decode("ascii"),
+    "format":   "glb",
+    "byte_size": len(glb_bytes),
+}
+assert protocol.is_envelope(envelope), (
+    "is_envelope returned False for a well-formed MODEL_3D envelope; "
+    "HEAVY_TYPES update did not take effect."
+)
+print("PASS: is_envelope recognises model_3d envelopes.")
+
+
+# ---------------------------------------------------------------------------
+# 3. decode_model3d_envelope round-trips inline bytes.
+# ---------------------------------------------------------------------------
+file3d = serialization.decode_model3d_envelope(envelope)
+from comfy_api.latest._util import File3D as _File3D_for_check  # noqa: E402
+assert isinstance(file3d, _File3D_for_check), (
+    f"Expected File3D, got {type(file3d).__name__}"
+)
+# File3D exposes get_data() / get_source(); the public API on this
+# version returns a BytesIO when constructed with one. Drain bytes:
+src = file3d._source  # internal, but stable across this code path
+assert hasattr(src, "read"), f"File3D source not BytesIO-like: {type(src)}"
+src.seek(0)
+round_trip = src.read()
+assert round_trip == glb_bytes, (
+    f"GLB byte mismatch after decode: input={len(glb_bytes)} bytes, "
+    f"output={len(round_trip)} bytes"
+)
+assert round_trip[:4] == b"glTF", "GLB magic bytes lost in round-trip"
+print(f"PASS: decode_model3d_envelope round-trips {len(glb_bytes)} GLB bytes.")
+
+
+# ---------------------------------------------------------------------------
+# 4. decode_envelope dispatches model_3d to the new decoder.
+# ---------------------------------------------------------------------------
+file3d2 = serialization.decode_envelope(envelope)
+assert isinstance(file3d2, _File3D_for_check), (
+    f"decode_envelope dispatch failed for type=model_3d: got "
+    f"{type(file3d2).__name__}"
+)
+print("PASS: decode_envelope dispatches type='model_3d' correctly.")
+
+
+# ---------------------------------------------------------------------------
+# 5. Unsupported encoding raises a hard error (defence-in-depth so a
+#    future encoding never silently downgrades to garbage bytes).
+# ---------------------------------------------------------------------------
+bad = dict(envelope, encoding="obj_inline")
+try:
+    serialization.decode_model3d_envelope(bad)
+except protocol.RnpProtocolError as e:
+    assert "Unsupported model_3d encoding" in str(e), (
+        f"Wrong error message for unsupported encoding: {e!r}"
+    )
+    assert e.code == protocol.ErrorCode.INTERNAL
+    print("PASS: unsupported model_3d encoding raises RnpProtocolError.")
+else:
+    raise AssertionError(
+        "decode_model3d_envelope accepted an unsupported encoding"
+    )
+
+
+# ---------------------------------------------------------------------------
+# 6. End-to-end: server's make_model3d_envelope -> client decode.
+#    Skips when the server worktree isn't on disk (e.g. CI running
+#    against just the client repo).
+# ---------------------------------------------------------------------------
+server_protocol_path = os.path.join(
+    _SERVER_DIR, "comfy_rnp_protocol", "envelopes.py",
+)
+if os.path.exists(server_protocol_path):
+    # Import the server-side envelopes module standalone (it has no
+    # ComfyUI deps and uses its own sibling ``constants`` import).
+    server_pkg = types.ModuleType("comfy_rnp_protocol_e2e")
+    server_pkg.__path__ = [os.path.join(_SERVER_DIR, "comfy_rnp_protocol")]
+    sys.modules["comfy_rnp_protocol_e2e"] = server_pkg
+    server_constants = _load_module(
+        "comfy_rnp_protocol_e2e.constants",
+        os.path.join(_SERVER_DIR, "comfy_rnp_protocol", "constants.py"),
+    )
+    server_envelopes = _load_module(
+        "comfy_rnp_protocol_e2e.envelopes",
+        os.path.join(_SERVER_DIR, "comfy_rnp_protocol", "envelopes.py"),
+    )
+    # Build using the server helper exactly as a provider would.
+    server_env = server_envelopes.make_model3d_envelope(
+        encoding="glb_inline",
+        data=base64.b64encode(glb_bytes).decode("ascii"),
+        format="glb",
+        byte_size=len(glb_bytes),
+    )
+    assert server_env["type"] == "model_3d", server_env
+    assert server_env["encoding"] == "glb_inline", server_env
+    assert "model_3d" in server_envelopes.HEAVY_TYPES, (
+        "Server HEAVY_TYPES missing 'model_3d' — server PR #37 should "
+        "have added it; reinstall comfy_rnp_protocol if this fires."
+    )
+    # Pipe through the client decoder.
+    file3d3 = serialization.decode_envelope(server_env)
+    src = file3d3._source
+    src.seek(0)
+    assert src.read() == glb_bytes, "End-to-end byte mismatch"
+    print(
+        "PASS: end-to-end server make_model3d_envelope -> client "
+        "decode_envelope preserves bytes."
+    )
+else:
+    print(
+        f"SKIP end-to-end: server worktree not found at {_SERVER_DIR!r}"
+    )
+
+
+# ---------------------------------------------------------------------------
+# 7. proxy_node._OUTPUT_CLASSES maps "MODEL_3D" -> IO.File3DGLB.Output.
+#    proxy_node pulls in the full comfy_api / server / comfy_api_nodes
+#    tree at import time (too heavy to stub here), so we verify the
+#    mapping with a source-level check instead. The pattern is exact:
+#    a quoted ``"MODEL_3D":`` key whose value is the
+#    ``IO.File3DGLB.Output`` class reference inside the
+#    ``_OUTPUT_CLASSES`` dict literal. The same source check would
+#    catch a typo (e.g. ``IO.File3DAny.Output``) that a smoke test
+#    importing the live module would miss too — the dict literal is
+#    the contract.
+# ---------------------------------------------------------------------------
+proxy_src = open(
+    os.path.join(_CLIENT_DIR, "proxy_node.py"), "r", encoding="utf-8",
+).read()
+assert "_OUTPUT_CLASSES = {" in proxy_src, "proxy_node missing _OUTPUT_CLASSES dict"
+assert '"MODEL_3D":' in proxy_src and "IO.File3DGLB.Output" in proxy_src, (
+    "proxy_node._OUTPUT_CLASSES does not appear to map "
+    "'MODEL_3D' -> IO.File3DGLB.Output"
+)
+# Verify the mapping is in the same dict literal (not somewhere else).
+oc_start = proxy_src.index("_OUTPUT_CLASSES = {")
+oc_end = proxy_src.index("}", oc_start)
+oc_block = proxy_src[oc_start:oc_end]
+assert '"MODEL_3D":' in oc_block, (
+    "'MODEL_3D' key not inside the _OUTPUT_CLASSES dict literal"
+)
+assert "IO.File3DGLB.Output" in oc_block, (
+    "IO.File3DGLB.Output value not inside the _OUTPUT_CLASSES dict literal"
+)
+print(
+    "PASS: proxy_node._OUTPUT_CLASSES dict literal contains "
+    "'MODEL_3D' -> IO.File3DGLB.Output."
+)
+
+
+# ---------------------------------------------------------------------------
+# 8. CLIENT_CAPABILITIES advertises model_3d:glb_inline.
+#    client.py imports aiohttp / comfy_api_nodes too — same source-level
+#    contract check.
+# ---------------------------------------------------------------------------
+client_src = open(
+    os.path.join(_CLIENT_DIR, "client.py"), "r", encoding="utf-8",
+).read()
+assert "CLIENT_CAPABILITIES = [" in client_src, (
+    "client.py missing CLIENT_CAPABILITIES list"
+)
+cc_start = client_src.index("CLIENT_CAPABILITIES = [")
+cc_end = client_src.index("]", cc_start)
+cc_block = client_src[cc_start:cc_end]
+assert "Capability.MODEL_3D_GLB_INLINE" in cc_block, (
+    "Capability.MODEL_3D_GLB_INLINE not advertised in CLIENT_CAPABILITIES"
+)
+print("PASS: CLIENT_CAPABILITIES advertises Capability.MODEL_3D_GLB_INLINE.")
+
+print("ALL CHECKS PASSED.")

protocol.py

@@ -104,19 +104,69 @@ class Capability:
     IMAGE_PNG_BASE64 = "image:png_base64"
     IMAGE_JPEG_BASE64 = "image:jpeg_base64"
     IMAGE_RAW_CHW_UINT8 = "image:raw_b64_chw_uint8"
+    # Multi-frame IMAGE envelope: a single value carries N PNG-base64
+    # frames sharing the same H/W/C, decoded into a [B,H,W,C] tensor on
+    # this client. Mirrors the AUDIO sweep precedent (parallel encodings
+    # rather than extending ``png_base64``) so legacy server emits still
+    # decode correctly when the new token isn't advertised.
+    IMAGE_PNG_BASE64_BATCH = "image:png_base64_batch"
+    # Externalised counterpart of ``image:png_base64_batch``: same
+    # multi-frame IMAGE envelope shape, but every frame's PNG bytes
+    # are uploaded out-of-band via the presigned-PUT path (see
+    # ``serialization.maybe_externalize``) and the envelope carries
+    # a parallel ``uris`` list instead of inline ``frames``. Required
+    # because the original ``maybe_externalize`` only inspected the
+    # singleton ``data`` field, so a batch of N PNGs whose summed
+    # base64 payload crossed the 8 MiB inline cap silently rode the
+    # wire as a 30+ MiB POST body. Per-frame URIs (NOT a tar/zip
+    # blob) keep this composable with the existing per-frame fetch
+    # helpers and let the server fetch in parallel. Decoder dispatch
+    # keys off the ``encoding`` string — ``frames`` vs ``uris`` is a
+    # payload field, not a shape selector.
+    IMAGE_PNG_BASE64_BATCH_URI = "image:png_base64_batch_uri"
     VIDEO_MP4_INLINE = "video:mp4_inline"
     VIDEO_MP4_BASE64 = "video:mp4_base64"
     AUDIO_MP3_INLINE = "audio:mp3_inline"
     AUDIO_MP3_BASE64 = "audio:mp3_base64"
     AUDIO_WAV_BASE64 = "audio:wav_base64"
     AUDIO_WAV_INLINE = "audio:wav_inline"
+    # OGG/Opus container as base64 in ``data``. Mirrors the parallel-
+    # encoding negotiation pattern already used for the AUDIO sweep
+    # (mp3 vs. wav) — servers only emit ``audio:opus_base64`` when
+    # this client advertises the token, otherwise they fall back to
+    # ``mp3_base64``. PyAV/FFmpeg decodes OGG/Opus natively so
+    # ``decode_audio_envelope`` ships the bytes through
+    # ``audio_bytes_to_audio_input`` the same way it ships MP3/WAV.
+    AUDIO_OPUS_BASE64 = "audio:opus_base64"
     MASK_PNG_BASE64 = "mask:png_base64"
     SVG_XML_BASE64 = "svg:svg_xml_base64"
     SCHEMA_V3 = "schema:v3"
     ASYNC_EXECUTE = "execute:async"
+    # Pass-through custom-IO bucket: when an input/output type string
+    # is not in the standard primitive/media set, the proxy_node parser
+    # falls back to ``IO.Custom(<type>)`` so partner helper-config nodes
+    # (OpenAIInputFiles, OpenAIChatConfig, GeminiInputFiles, Recraft
+    # style/color/controls helpers, future partner helper-config types)
+    # round-trip without per-type encoder plumbing. The descriptor's
+    # original io_type string is preserved on the V3 input so node-to-
+    # node connections only chain when the strings match. Encoder
+    # dispatch passes opaque dict/list/string values through
+    # ``_encode_one`` unchanged — they're already JSON-serialisable.
+    IO_OPAQUE = "io:opaque"
+    # 3D MODEL envelope: a single-mesh GLB (binary glTF 2.0, magic
+    # b"glTF") wrapped opaquely in ``data`` (base64). Mirrors the
+    # parallel-encoding pattern already used for VIDEO_MP4_INLINE — the
+    # server emits this token only when the inbound request advertised
+    # it; legacy clients without the token will surface
+    # NEGOTIATION_FAILED at descriptor-load time. The decoder on the
+    # client side hands raw GLB bytes straight to ComfyUI's
+    # ``Types.File3D`` / ``IO.File3DGLB`` machinery (no in-server GLB
+    # parse — the magic-byte check at offset 0 is the only validity
+    # guard the server applies).
+    MODEL_3D_GLB_INLINE = "model_3d:glb_inline"
 
 
-HEAVY_TYPES = frozenset({"image", "video", "audio", "mask", "svg"})
+HEAVY_TYPES = frozenset({"image", "video", "audio", "mask", "model_3d"})
 
 
 def is_envelope(value: Any) -> bool:

proxy_node.py

@@ -3,11 +3,13 @@
 The descriptor wire format is V3 ``Schema.get_v1_info()`` verbatim, so
 per-input dicts are handed straight to V3's IO classes without
 reinterpretation. ``_parse_input_spec`` knows STRING / INT / FLOAT /
-BOOLEAN / COMBO / IMAGE / VIDEO / AUDIO / MASK natively; any other
-``io_type`` string is treated as an opaque pass-through socket (see
-``IO.Custom``) so partner-defined types like ``RECRAFT_STYLE_V3`` /
-``RECRAFT_COLOR`` round-trip between RNP nodes without the descriptor
-being rejected.
+BOOLEAN / COMBO / IMAGE / VIDEO / AUDIO / MASK; any other io_type
+string falls through to ``IO.Custom(io_type)`` (the §B opaque
+pass-through bucket — partner helper-config types like RECRAFT_*,
+OPENAI_INPUT_FILES, OPENAI_CHAT_CONFIG, GEMINI_INPUT_FILES round-trip
+as raw JSON blobs with the original io_type preserved on the V3
+socket, so node-to-node connections only chain when the strings
+match). Only a non-string / empty io_type causes a skip.
 
 Hidden inputs (``auth_token_comfy_org`` / ``api_key_comfy_org`` /
 ``unique_id`` / ``prompt`` / ``extra_pnginfo`` / ``dynprompt``) are
@@ -45,7 +47,6 @@ from .protocol import (
     DEFAULT_POLL_INTERVAL_S,
     DEFAULT_RETRY_BACKOFF_PER_POLL,
     DEFAULT_RETRY_DELAY_PER_POLL_S,
-    DEFAULT_SOFT_TIMEOUT_S,
     DEFAULT_TIMEOUT_PER_POLL_S,
     ErrorCode,
     ExecutionMode,
@@ -308,6 +309,32 @@ def build_node_class(
     schema_hash = remote.get("schema_hash")
     execute_path = execute_endpoint["path"]
     execution = remote.get("execution") or {}
+    # ``descriptor.remote.input_serialization`` advertises the wire
+    # encodings each input accepts. The wire shape is
+    # ``{input_name: <str | list[str]>}`` — single string for a
+    # single-encoding input (e.g. ``"png_base64"``), list when the
+    # input accepts multiple encodings (e.g.
+    # ``["png_base64", "png_base64_batch"]`` for a multi-frame IMAGE).
+    # Normalize to ``dict[str, list[str]]`` so the encoder can do a
+    # single ``in`` check per input without re-parsing per call.
+    raw_input_serialization = (
+        remote.get("input_serialization") if isinstance(remote.get("input_serialization"), dict) else {}
+    )
+    input_serialization: dict[str, list[str]] = {}
+    for name, val in raw_input_serialization.items():
+        if isinstance(val, str):
+            input_serialization[name] = [val]
+        elif isinstance(val, list):
+            input_serialization[name] = [v for v in val if isinstance(v, str)]
+
+    # ``descriptor.input.<section>.<name>[1].local_validate`` carries
+    # declarative client-side validation rules (e.g.
+    # ``{"image_max_batch": 9}`` for a Flux2 reference-image input).
+    # The encoder consults this map and raises ``INPUT_INVALID``
+    # locally before any HTTP round trip when an over-cap batch is
+    # wired in. Lookup is defensive: missing ``local_validate`` →
+    # no validation; missing per-rule key → no validation.
+    local_validate = _collect_local_validate(descriptor)
     timeout_s = float(
         execution.get("hard_timeout_s")
         or execution.get("soft_timeout_s")
@@ -324,16 +351,6 @@ def build_node_class(
     url_fetch_policy = (
         remote.get("url_fetch") if isinstance(remote.get("url_fetch"), dict) else {}
     )
-    # ``descriptor.remote.hidden_forwarded`` opt-in for non-auth hidden
-    # markers (``prompt`` / ``extra_pnginfo`` / ``dynprompt``). Auth
-    # tokens and ``unique_id`` always forward when declared. Anything
-    # else is declared on the schema (so the executor populates
-    # ``cls.hidden``) but dropped before building the request body so
-    # large EXTRA_PNGINFO blobs don't ship by default.
-    hidden_forwarded_raw = remote.get("hidden_forwarded") or []
-    hidden_forwarded: list[str] = [
-        h for h in hidden_forwarded_raw if isinstance(h, str)
-    ] if isinstance(hidden_forwarded_raw, (list, tuple)) else []
 
     display_name = descriptor.get("display_name") or node_id
     category = descriptor.get("category") or "remote"
@@ -392,13 +409,14 @@ def build_node_class(
                 timeout_s=timeout_s,
                 outputs_meta=schema_outputs,
                 hidden_names=hidden_names,
-                hidden_forwarded=hidden_forwarded,
                 inputs=kwargs,
                 max_inline_bytes=max_inline_bytes,
                 estimated_duration_s=estimated_duration_s,
                 execution_mode=execution_mode,
                 execution_policy=execution_policy,
                 url_fetch_policy=url_fetch_policy,
+                input_serialization=input_serialization,
+                local_validate=local_validate,
             )
 
     RemoteProxyNode.__name__ = node_id
@@ -424,6 +442,111 @@ _HIDDEN_MARKERS = {
 }
 
 
+def _collect_local_validate(descriptor: dict[str, Any]) -> dict[str, dict[str, Any]]:
+    """Pull ``local_validate`` rule dicts off every required/optional input
+    spec in the descriptor, recursing into AUTOGROW templates and
+    DYNAMIC_COMBO branches.
+
+    Returns ``{input_name: rules_dict}`` for every input that carries a
+    non-empty ``local_validate`` block (or whose nested template /
+    branches do). The returned ``rules_dict`` may carry the literal
+    ``local_validate`` keys (e.g. ``{"image_max_batch": 9}``) plus two
+    private composite keys consumed by ``_enforce_local_validate``:
+
+    * ``__template__`` — for AUTOGROW inputs, the rules harvested from
+      the template's options dict. The enforcer applies these to every
+      slot value in the runtime Autogrow dict
+      (``{"image0": <env>, "image1": <env>, ...}``).
+    * ``__branches__`` — for DYNAMIC_COMBO inputs, a
+      ``{branch_key: {sub_name: rules}}`` map. The enforcer reads the
+      selected branch from ``value[<name>]`` and walks the matching
+      branch's sub-inputs (used by GPT Image V2 / Grok ImageEditV2 /
+      Grok VideoReference, where the cap lives on a nested Autogrow
+      template inside a DynamicCombo branch).
+
+    Inputs whose entire subtree is empty are omitted so the encoder can
+    use a single ``in`` check per call.
+    """
+    out: dict[str, dict[str, Any]] = {}
+    input_def = descriptor.get("input") or {}
+    for kind in ("required", "optional"):
+        defs = input_def.get(kind) or {}
+        for name, spec in defs.items():
+            rules = _rules_from_input_spec(spec)
+            if rules:
+                out[name] = rules
+    return out
+
+
+def _rules_from_input_spec(spec: Any) -> dict[str, Any]:
+    """Extract ``local_validate`` rules from one V3 input spec.
+
+    Accepts both wire shapes:
+
+    * ``[io_type, options_dict]`` — top-level inputs (the descriptor's
+      ``input.required[name]`` value, or an AUTOGROW template).
+    * ``[name, io_type, options_dict]`` — DynamicCombo branch sub-inputs
+      (each entry of ``options.options[i].inputs``).
+
+    For ``AUTOGROW`` specs, recurses into ``options["template"]`` and
+    stashes the result under ``__template__``. For ``DYNAMIC_COMBO``,
+    walks each branch's sub-inputs and stashes the resulting
+    ``{branch_key: {sub_name: sub_rules}}`` map under ``__branches__``.
+    """
+    if not isinstance(spec, (list, tuple)) or len(spec) < 2:
+        return {}
+    if (
+        len(spec) >= 3
+        and isinstance(spec[1], str)
+        and isinstance(spec[2], dict)
+    ):
+        # [name, io_type, options]
+        io_type, options = spec[1], spec[2]
+    elif isinstance(spec[1], dict):
+        # [io_type, options]
+        io_type, options = spec[0], spec[1]
+    else:
+        return {}
+    rules: dict[str, Any] = {}
+    direct = options.get("local_validate")
+    if isinstance(direct, dict) and direct:
+        rules.update(direct)
+    if io_type == "AUTOGROW":
+        tmpl_spec = options.get("template")
+        tmpl_rules = (
+            _rules_from_input_spec(tmpl_spec)
+            if isinstance(tmpl_spec, (list, tuple))
+            else {}
+        )
+        if tmpl_rules:
+            rules["__template__"] = tmpl_rules
+    elif io_type == "DYNAMIC_COMBO":
+        branches: dict[str, dict[str, Any]] = {}
+        for opt in (options.get("options") or []):
+            if not isinstance(opt, dict):
+                continue
+            key = opt.get("key")
+            if not isinstance(key, str):
+                continue
+            sub_rules: dict[str, dict[str, Any]] = {}
+            for sub_spec in (opt.get("inputs") or []):
+                if (
+                    not isinstance(sub_spec, (list, tuple))
+                    or len(sub_spec) < 1
+                    or not isinstance(sub_spec[0], str)
+                ):
+                    continue
+                sub_name = sub_spec[0]
+                r = _rules_from_input_spec(sub_spec)
+                if r:
+                    sub_rules[sub_name] = r
+            if sub_rules:
+                branches[key] = sub_rules
+        if branches:
+            rules["__branches__"] = branches
+    return rules
+
+
 def _parse_inputs(descriptor: dict[str, Any]) -> tuple[list[Any] | None, list[str]]:
     """Return ``(input_objs, hidden_input_names)``.
 
@@ -468,17 +591,13 @@ def _parse_input_spec(name: str, spec: list[Any], optional: bool) -> Any | None:
     if isinstance(io_type, list):
         # Legacy V1-style combo: io_type *is* the options list.
         return IO.Combo.Input(name, options=list(io_type),
-                              default=options.get("default"),
-                              control_after_generate=options.get("control_after_generate"),
-                              **common)
+                              default=options.get("default"), **common)
     if io_type == "COMBO":
         opts = options.get("options")
         if not isinstance(opts, (list, tuple)):
             return None
         return IO.Combo.Input(name, options=list(opts),
-                              default=options.get("default"),
-                              control_after_generate=options.get("control_after_generate"),
-                              **common)
+                              default=options.get("default"), **common)
     if io_type == "STRING":
         return IO.String.Input(
             name,
@@ -493,7 +612,6 @@ def _parse_input_spec(name: str, spec: list[Any], optional: bool) -> Any | None:
             min=options.get("min", 0),
             max=options.get("max", 2147483647),
             step=options.get("step", 1),
-            control_after_generate=options.get("control_after_generate"),
             **common,
         )
     if io_type == "FLOAT":
@@ -517,130 +635,20 @@ def _parse_input_spec(name: str, spec: list[Any], optional: bool) -> Any | None:
         return IO.Audio.Input(name, **common)
     if io_type == "MASK":
         return IO.Mask.Input(name, **common)
-    if io_type == "SVG":
-        return IO.SVG.Input(name, **common)
-    if io_type == "AUTOGROW":
-        # Wire shape:
-        #   ["AUTOGROW", {
-        #       "template": ["IMAGE", {...}],   # nested input spec
-        #       "prefix":   "image",
-        #       "min":      1,
-        #       "max":      5,
-        #   }]
-        # The descriptor declares N "<prefix>0".."<prefix>{max-1}"
-        # virtual inputs that the V3 frontend renders as a single
-        # auto-growing block — see ``IO.Autogrow.TemplatePrefix``.
-        # The first ``min`` slots are required, the rest optional;
-        # the executor packs the connected ones into a dict keyed by
-        # virtual id which the server provider iterates.
-        template_spec = options.get("template")
-        if (
-            not isinstance(template_spec, (list, tuple))
-            or len(template_spec) < 1
-            or not isinstance(template_spec[0], str)
-        ):
-            log.warning(
-                "AUTOGROW input %s: malformed template spec %r",
-                name, template_spec,
-            )
-            return None
-        # The template is itself an ``[io_type, {...}]`` pair — reuse
-        # the parser recursively so AUTOGROW inherits every IO type the
-        # parser knows about (IMAGE, MASK, STRING, custom IO …).
-        # The framework forbids ``DynamicInput`` as an Autogrow template
-        # (``assert(not isinstance(input, DynamicInput))`` in
-        # ``_AutogrowTemplate.__init__``), so passing AUTOGROW or
-        # DYNAMIC_COMBO here will raise inside ``TemplatePrefix`` —
-        # caught below.
-        template_input = _parse_input_spec(
-            "_autogrow_template", list(template_spec), optional=False,
-        )
-        if template_input is None:
-            log.warning(
-                "AUTOGROW input %s: failed to parse template %r",
-                name, template_spec,
-            )
-            return None
-        prefix = options.get("prefix")
-        if not isinstance(prefix, str) or not prefix:
-            return None
-        min_slots = int(options.get("min", 1))
-        max_slots = int(options.get("max", 1))
-        if max_slots < 1 or min_slots < 0 or min_slots > max_slots:
-            return None
-        try:
-            template = IO.Autogrow.TemplatePrefix(
-                template_input,
-                prefix=prefix,
-                min=min_slots,
-                max=max_slots,
-            )
-        except Exception as e:  # noqa: BLE001
-            log.warning(
-                "AUTOGROW input %s: failed to build TemplatePrefix: %s",
-                name, e,
-            )
-            return None
-        return IO.Autogrow.Input(
-            name,
-            template=template,
-            display_name=options.get("display_name"),
-            optional=optional,
-            tooltip=options.get("tooltip"),
-        )
-    if io_type == "DYNAMIC_COMBO":
-        # Wire shape:
-        #   ["DYNAMIC_COMBO", {
-        #       "options": [
-        #           {"key": "recraftv4",     "inputs": [
-        #               ["size", "COMBO", {"options": [...]}],
-        #               ...
-        #           ]},
-        #           {"key": "recraftv4_pro", "inputs": [...]},
-        #       ],
-        #       "default": "recraftv4",
-        #       "tooltip": "...",
-        #   }]
-        # The selected key drives which sub-widget block the frontend
-        # renders; the executor packs ``{"model": <key>, ...sub fields}``
-        # into a dict the server consumes via
-        # ``inputs["model"]["model"]`` / ``inputs["model"]["size"]``
-        # idiom — same shape ``IO.DynamicCombo`` produces locally.
-        opts = options.get("options")
-        if not isinstance(opts, list) or not opts:
-            return None
-        dc_options: list[Any] = []
-        for entry in opts:
-            if not isinstance(entry, dict):
-                return None
-            key = entry.get("key")
-            sub_inputs_spec = entry.get("inputs")
-            if not isinstance(key, str) or not isinstance(sub_inputs_spec, list):
-                return None
-            sub_inputs: list[Any] = []
-            for sub in sub_inputs_spec:
-                if not (isinstance(sub, (list, tuple)) and len(sub) >= 2 and isinstance(sub[0], str)):
-                    return None
-                sub_name = sub[0]
-                sub_spec = list(sub[1:])
-                # ``[name, io_type, options]`` flattens to the parser's
-                # ``[io_type, options]`` shape with sub_name carried separately.
-                sub_input = _parse_input_spec(sub_name, sub_spec, optional=False)
-                if sub_input is None:
-                    return None
-                sub_inputs.append(sub_input)
-            dc_options.append(IO.DynamicCombo.Option(key, sub_inputs))
-        return IO.DynamicCombo.Input(
-            name,
-            options=dc_options,
-            display_name=options.get("display_name"),
-            optional=optional,
-            tooltip=options.get("tooltip"),
-        )
-    # Any other string io_type — partner-defined custom IO (RecraftStyle,
-    # RecraftColor, Tripo3DModel, …). Built as an opaque pass-through
-    # socket so the descriptor is accepted; the value flows between RNP
-    # nodes verbatim and the proxy_node never inspects or serialises it.
+    # Opaque custom-IO fallthrough — any unknown io_type string becomes
+    # an ``IO.Custom(io_type)`` socket so partner helper-config nodes
+    # (OpenAIInputFiles → OPENAI_INPUT_FILES, OpenAIChatConfig →
+    # OPENAI_CHAT_CONFIG, GeminiInputFiles → GEMINI_INPUT_FILES, the
+    # already-shipped Recraft RECRAFT_* helpers, future partner types)
+    # round-trip without per-type encoder plumbing. The descriptor's
+    # original ``io_type`` string is preserved on ``IO.Custom``'s
+    # ComfyType.io_type via the V3 ``@comfytype`` decorator, so the
+    # frontend's connection-validity check only chains sockets when the
+    # strings match — an OPENAI_INPUT_FILES output cannot connect to a
+    # GEMINI_INPUT_FILES input. Encoder dispatch in
+    # ``serialization`` already passes non-tensor / non-audio values
+    # through ``_encode_one`` unchanged, and the deserializer treats
+    # non-envelope values as already-native Python objects.
     if isinstance(io_type, str) and io_type:
         return IO.Custom(io_type).Input(name, **common)
     return None
@@ -727,32 +735,47 @@ def _parse_outputs(descriptor: dict[str, Any]) -> list[Any] | None:
         tip = output_tooltips[i] if i < len(output_tooltips) else None
         cls = _OUTPUT_CLASSES.get(io_type)
         if cls is None:
-            # Partner-defined custom IO output (RecraftStyle, Tripo3DModel,
-            # Hunyuan3DMesh, …). Built as an opaque pass-through socket so
-            # the value flows downstream verbatim — no envelope decoding.
-            if not isinstance(io_type, str) or not io_type:
+            # Opaque custom-IO fallthrough (mirror of
+            # ``_parse_input_spec``) — any unknown output io_type
+            # string becomes ``IO.Custom(io_type).Output`` so partner
+            # helper-config nodes can publish their own opaque
+            # outputs (RECRAFT_*, OPENAI_INPUT_FILES, GEMINI_INPUT_FILES,
+            # OPENAI_CHAT_CONFIG, …). The original io_type string
+            # rides on the resulting Output via V3's ``@comfytype``
+            # decorator, so the frontend only chains output→input
+            # sockets when the strings match.
+            if isinstance(io_type, str) and io_type:
+                cls = IO.Custom(io_type).Output
+            else:
                 log.warning(
-                    "Skipping descriptor: unsupported output type %r", io_type,
+                    "Skipping descriptor: unsupported output type %r",
+                    io_type,
                 )
                 return None
-            out.append(IO.Custom(io_type).Output(
-                display_name=name, is_output_list=is_list, tooltip=tip,
-            ))
-            continue
         out.append(cls(display_name=name, is_output_list=is_list, tooltip=tip))
     return out
 
 
 _OUTPUT_CLASSES = {
-    "VIDEO":   IO.Video.Output,
-    "IMAGE":   IO.Image.Output,
-    "AUDIO":   IO.Audio.Output,
-    "MASK":    IO.Mask.Output,
-    "SVG":     IO.SVG.Output,
-    "STRING":  IO.String.Output,
-    "INT":     IO.Int.Output,
-    "FLOAT":   IO.Float.Output,
-    "BOOLEAN": IO.Boolean.Output,
+    "VIDEO":    IO.Video.Output,
+    "IMAGE":    IO.Image.Output,
+    "AUDIO":    IO.Audio.Output,
+    "MASK":     IO.Mask.Output,
+    "STRING":   IO.String.Output,
+    "INT":      IO.Int.Output,
+    "FLOAT":    IO.Float.Output,
+    "BOOLEAN":  IO.Boolean.Output,
+    # 3D MODEL output: the Tripo / Rodin / Meshy partner endpoints all
+    # return a single GLB file; the server emits a ``model_3d:glb_inline``
+    # envelope which ``serialization.decode_model3d_envelope`` turns
+    # back into a ``File3D`` so downstream nodes (preview3d / SaveGLB /
+    # Load3D) chain off the output socket unchanged. The descriptor
+    # io_type string is the logical name ``"MODEL_3D"`` (consistent
+    # across vendors) — the resulting V3 socket carries the upstream
+    # ``"FILE_3D_GLB"`` comfytype tag from ``IO.File3DGLB`` so
+    # frontend-side input/output matching works against any GLB-aware
+    # node.
+    "MODEL_3D": IO.File3DGLB.Output,
 }
 
 
@@ -994,28 +1017,21 @@ async def _execute_remote(
     timeout_s: float,
     outputs_meta: list[Any],
     hidden_names: list[str],
-    hidden_forwarded: list[str],
     inputs: dict[str, Any],
     max_inline_bytes: int | None = None,
     estimated_duration_s: int | float | None = None,
     execution_mode: str = ExecutionMode.REQUEST_RESPONSE,
     execution_policy: dict[str, Any] | None = None,
     url_fetch_policy: dict[str, Any] | None = None,
+    input_serialization: dict[str, list[str]] | None = None,
+    local_validate: dict[str, dict[str, Any]] | None = None,
 ) -> IO.NodeOutput:
     policy = execution_policy or _extract_execution_policy({})
     url_fetch_policy = url_fetch_policy or {}
     # Strip hidden inputs from the user-supplied kwargs. Auth credentials
     # (``auth_token_comfy_org`` / ``api_key_comfy_org``) travel as standard
     # HTTP headers so the server can forward them to api.comfy.org without
-    # parsing the body; ``unique_id`` always rides in ``context`` for
-    # correlation. Other markers (``prompt`` / ``extra_pnginfo`` /
-    # ``dynprompt``) only forward when the descriptor explicitly opted
-    # in via ``remote.hidden_forwarded`` — otherwise they're declared on
-    # the schema (so the executor populates ``cls.hidden``) but stripped
-    # before we build the request body, so EXTRA_PNGINFO blobs don't
-    # leak by default.
-    _ALWAYS_FORWARD = {"auth_token_comfy_org", "api_key_comfy_org", "unique_id"}
-    forward_set = _ALWAYS_FORWARD | set(hidden_forwarded)
+    # parsing the body; everything else rides in the request ``context``.
     context: dict[str, Any] = {}
     auth_headers: dict[str, str] = {}
     hidden_holder = getattr(cls, "hidden", None)
@@ -1036,17 +1052,24 @@ async def _execute_remote(
             auth_headers["Authorization"] = f"Bearer {value}"
         elif hname == "api_key_comfy_org":
             auth_headers["X-API-KEY"] = str(value)
-        elif hname in forward_set:
+        else:
             context[hname] = value
 
     # Encode heavy-typed inputs (IMAGE/MASK/AUDIO) as RNP value
     # envelopes; scalars pass through. Oversized envelopes are
     # uploaded out-of-band and rewritten to ``{..., uri: <url>}``.
+    # ``input_serialization`` carries the per-input encoding allow-list
+    # from the descriptor — the encoder uses it to decide whether to
+    # ship a multi-frame ``png_base64_batch`` envelope or fall back
+    # to single-frame ``png_base64``.
     payload_inputs = await _encode_inputs(
         inputs,
         server_url=server_url,
         max_inline_bytes=max_inline_bytes,
         auth_headers=auth_headers,
+        input_serialization=input_serialization or {},
+        local_validate=local_validate or {},
+        node_id=node_id,
     )
 
     ticker_node_id = _node_id_for_ticker(cls)
@@ -1221,8 +1244,54 @@ async def _deserialize_output(
         # advisory here, the value already matches the native ComfyUI
         # type for STRING / INT / FLOAT / BOOLEAN.
         return value
-    if "data" not in value and "uri" in value:
-        env_type = value.get("type", "")
+    encoding = value.get("encoding")
+    env_type = value.get("type", "")
+    if encoding == "png_base64_batch_uri":
+        # Externalised IMAGE batch on the response path (mirror of the
+        # request-side ``serialization.maybe_externalize`` behaviour).
+        # No server emits this shape today, but the decoder owns the
+        # symmetry: any envelope shape the client may externalise on
+        # the way out, the client must also accept on the way in.
+        # Per-frame URIs fetched in parallel via ``asyncio.gather`` so
+        # a 9-frame batch import is one round-trip rather than N — the
+        # GET fan-out is the dominant latency for an out-of-band batch.
+        # Frame ordering is preserved (gather returns positional
+        # results) — critical because the IMAGE tensor's batch axis
+        # carries semantic order.
+        policy = url_fetch_policy.get(env_type)
+        if not isinstance(policy, dict):
+            raise RnpProtocolError(
+                f"Server returned a URL batch envelope for output type "
+                f"{env_type!r} but the descriptor declared no "
+                f"remote.url_fetch policy for that type",
+                code=ErrorCode.INTERNAL,
+            )
+        uris = value.get("uris")
+        if not isinstance(uris, list) or not uris:
+            raise RnpProtocolError(
+                "png_base64_batch_uri envelope requires a non-empty 'uris' list",
+                code=ErrorCode.INTERNAL,
+            )
+        import asyncio as _asyncio
+        import base64 as _b64
+        per_frame_envs = [
+            {"type": env_type, "encoding": "png_base64", "uri": u}
+            for u in uris
+        ]
+        raws = await _asyncio.gather(*[
+            _fetch_url_envelope(
+                env, policy,
+                auth_headers=auth_headers,
+                node_cls=node_cls,
+            )
+            for env in per_frame_envs
+        ])
+        # Re-stamp as the inline batch shape so the existing
+        # ``_decode_image_batch_envelope`` path handles it uniformly.
+        value = {k: v for k, v in value.items() if k != "uris"}
+        value["encoding"] = "png_base64_batch"
+        value["frames"] = [_b64.b64encode(r).decode("ascii") for r in raws]
+    elif "data" not in value and "uri" in value:
         policy = url_fetch_policy.get(env_type)
         if not isinstance(policy, dict):
             raise RnpProtocolError(
@@ -1304,6 +1373,9 @@ async def _encode_inputs(
     server_url: str | None = None,
     max_inline_bytes: int | None = None,
     auth_headers: dict[str, str] | None = None,
+    input_serialization: dict[str, list[str]] | None = None,
+    local_validate: dict[str, dict[str, Any]] | None = None,
+    node_id: str | None = None,
 ) -> dict[str, Any]:
     """Convert in-place: replace heavy-typed input values with the
     matching RNP value envelope. Scalars pass through unchanged.
@@ -1319,10 +1391,29 @@ async def _encode_inputs(
     When ``max_inline_bytes`` is set, any envelope whose inline payload
     exceeds the threshold is externalized via ``POST /rnp/v1/uploads``
     (see ``serialization.maybe_externalize``).
+
+    ``input_serialization`` is the descriptor's per-input encoding
+    allow-list (e.g. ``{"images": ["png_base64", "png_base64_batch"]}``).
+    The encoder consults it to decide whether a multi-frame IMAGE input
+    can ride a single batch envelope vs. falling back to single-frame.
+
+    ``local_validate`` is the per-input declarative-rule map
+    (e.g. ``{"images": {"image_max_batch": 9}}``). For each rule the
+    encoder enforces the constraint locally and raises
+    ``RnpProtocolError(code=INPUT_INVALID, user_facing=True)`` so the
+    proxy_node's friendly-message dispatch surfaces a per-provider
+    error before any paid HTTP round trip. Lookup is defensive: a
+    missing ``local_validate`` entry or missing per-rule key is a
+    no-op.
     """
+    serialization_map = input_serialization or {}
+    validate_map = local_validate or {}
     out: dict[str, Any] = {}
     for name, value in inputs.items():
-        encoded = _encode_one(name, value)
+        _enforce_local_validate(
+            name, value, validate_map.get(name) or {}, node_id=node_id,
+        )
+        encoded = _encode_one(name, value, serialization_map.get(name))
         # Only envelope-shaped values can be externalized; scalars pass
         # straight through ``maybe_externalize`` unchanged.
         if serialization.is_envelope(encoded):
@@ -1336,52 +1427,129 @@ async def _encode_inputs(
     return out
 
 
-def _dict_contains_tensor(value: Any) -> bool:
-    """True iff a dict (or nested dict) holds at least one torch tensor.
+def _enforce_local_validate(
+    name: str,
+    value: Any,
+    rules: dict[str, Any],
+    *,
+    node_id: str | None,
+) -> None:
+    """Apply each ``local_validate.*`` rule to ``value`` and raise
+    ``RnpProtocolError(INPUT_INVALID)`` on a violation.
 
-    Used to gate the recursive encode branch so we don't pointlessly
-    rebuild scalar custom-IO chain dicts (RECRAFT_V3_STYLE etc.) on
-    every encode pass, and so already-encoded envelope dicts pass
-    straight through. Nested-dict support keeps DynamicCombo→Autogrow
-    composition (and any future layered dynamic IO) round-trippable.
+    Rules currently supported:
+
+    * ``image_max_batch`` — for IMAGE inputs (rank-4 tensors), assert
+      ``tensor.shape[0] <= cap``. Mirrors the per-call cap raised on
+      the server (e.g. Flux2 raises "The current maximum number of
+      supported images is 9.") so a workflow with N>cap reference
+      frames fails fast on the client before any paid HTTP round trip.
+
+    Two private composite keys are walked transparently:
+
+    * ``__template__`` (AUTOGROW) — ``value`` is expected to be a
+      runtime Autogrow dict (``{"image0": <env>, "image1": <env>, ...}``).
+      The template's rules are applied to each non-``None`` slot value.
+    * ``__branches__`` (DYNAMIC_COMBO) — ``value`` is expected to be the
+      DynamicCombo runtime dict whose ``value[<name>]`` carries the
+      selected branch key. The matching branch's per-sub-input rules are
+      applied recursively. This is what makes GPT Image V2 / Grok
+      ImageEditV2 / Grok VideoReference's nested per-slot caps fire
+      client-side before any HTTP round trip.
     """
-    if isinstance(value, dict):
-        return any(_dict_contains_tensor(v) for v in value.values())
-    return serialization._is_torch_tensor(value)
+    if not rules:
+        return
+    _check_image_max_batch(name, value, rules.get("image_max_batch"), node_id=node_id)
+    tmpl = rules.get("__template__")
+    if isinstance(tmpl, dict) and isinstance(value, dict):
+        for slot_key, slot_value in value.items():
+            if slot_value is None:
+                continue
+            _enforce_local_validate(
+                f"{name}.{slot_key}", slot_value, tmpl, node_id=node_id,
+            )
+    branches = rules.get("__branches__")
+    if isinstance(branches, dict) and isinstance(value, dict):
+        branch_key = value.get(name)
+        if isinstance(branch_key, str):
+            sub_rules_map = branches.get(branch_key)
+            if isinstance(sub_rules_map, dict):
+                for sub_name, sub_rules in sub_rules_map.items():
+                    if sub_name in value:
+                        _enforce_local_validate(
+                            f"{name}.{sub_name}", value[sub_name], sub_rules,
+                            node_id=node_id,
+                        )
 
 
-def _encode_one(name: str, value: Any) -> Any:
+def _check_image_max_batch(
+    name: str,
+    value: Any,
+    cap: Any,
+    *,
+    node_id: str | None,
+) -> None:
+    """Apply the ``image_max_batch`` rule. No-op if cap is missing /
+    non-int / value is not a recognisable image tensor."""
+    if cap is None:
+        return
+    try:
+        cap_int = int(cap)
+    except (TypeError, ValueError):
+        return
+    if not serialization._is_torch_tensor(value):
+        return
+    rank = value.dim()
+    last_dim = int(value.shape[-1]) if rank >= 1 else 0
+    if not (rank == 4 or (rank == 3 and last_dim in (3, 4))):
+        return
+    batch = int(value.shape[0]) if rank == 4 else 1
+    if batch <= cap_int:
+        return
+    label = f" ({node_id})" if node_id else ""
+    if cap_int == 1:
+        msg = (
+            f"Remote node{label}: input {name!r} expects a single image "
+            f"frame, got a batch of {batch}. Wire a single-frame IMAGE "
+            f"into this input."
+        )
+    else:
+        msg = (
+            f"Remote node{label}: input {name!r} accepts at most "
+            f"{cap_int} image frames, got a batch of {batch}. The "
+            f"current maximum number of supported images is "
+            f"{cap_int}."
+        )
+    raise RnpProtocolError(
+        msg,
+        code=ErrorCode.INPUT_INVALID,
+        user_facing=True,
+    )
+
+
+def _encode_one(
+    name: str,
+    value: Any,
+    accepted_encodings: list[str] | None = None,
+) -> Any:
     if serialization.is_audio_input(value):
         return serialization.encode_audio_input(value)
-    if serialization.is_video_input(value):
-        return serialization.encode_video_input(value)
-    if (
-        isinstance(value, dict)
-        and not serialization.is_envelope(value)
-        and not serialization.is_audio_input(value)
-        and _dict_contains_tensor(value)
-    ):
-        # IO.Autogrow inputs arrive grouped as a dict (e.g. ``{"image0":
-        # tensor, "image1": tensor}``) keyed by the per-slot virtual id.
-        # Nested dynamic types (DynamicCombo whose option inputs include
-        # an Autogrow, etc.) can produce arbitrarily nested dict-of-
-        # tensor shapes — recurse so every heavy-typed leaf becomes an
-        # envelope. Skip ordinary scalar dicts (custom-IO chain values
-        # like RECRAFT_V3_STYLE) and already-encoded envelopes; both
-        # pass through unchanged via the ``return value`` tail.
-        encoded_dict: dict[str, Any] = {}
-        for sub_name, sub_value in value.items():
-            encoded_dict[sub_name] = _encode_one(f"{name}.{sub_name}", sub_value)
-        return encoded_dict
     if not serialization._is_torch_tensor(value):
         return value
     rank = value.dim()
     last_dim = int(value.shape[-1]) if rank >= 1 else 0
     if rank == 4 or (rank == 3 and last_dim in (3, 4)):
-        # Per-image tensors: a length-1 batch is the common case; longer
-        # batches send only the first frame (the framework hasn't asked
-        # for batch-of-envelopes semantics yet).
-        return serialization.encode_image_tensor(value)
+        # Per-image tensors: a length-1 batch is the common case.
+        # When the descriptor advertises ``"png_base64_batch"`` for
+        # this input AND the tensor has > 1 frame, ship the multi-
+        # frame envelope; otherwise fall back to single-frame
+        # (``tensor[0]`` for batches >1, full tensor for length-1).
+        accepts_batch = bool(
+            accepted_encodings and "png_base64_batch" in accepted_encodings
+        )
+        return serialization.encode_image_tensor(
+            value, accepts_batch=accepts_batch,
+        )
     if rank in (2, 3):
         return serialization.encode_mask_tensor(value)
     log.warning(

serialization.py

@@ -17,6 +17,7 @@ tensor — no parallel dialect.
 """
 from __future__ import annotations
 
+import asyncio
 import base64
 import logging
 from io import BytesIO
@@ -56,14 +57,64 @@ async def maybe_externalize(
     max_inline_bytes: int | None,
     auth_headers: dict[str, str] | None = None,
 ) -> dict[str, Any]:
-    """If the envelope's inline data exceeds ``max_inline_bytes``,
-    upload the bytes and replace ``data`` with ``uri``.
+    """If the envelope's inline payload exceeds ``max_inline_bytes``,
+    upload it via the presigned-PUT path and rewrite the envelope to
+    reference the uploaded asset(s) instead of inlining base64.
+
+    Two payload shapes are recognised:
+
+    * Singleton ``data`` envelopes (the common single-frame IMAGE /
+      MASK / VIDEO / AUDIO case). Oversize payloads upload as one
+      asset and the envelope's ``data`` field is swapped for ``uri``.
+    * Multi-frame ``frames=[…]`` IMAGE envelopes (encoding
+      ``png_base64_batch``). The original ``maybe_externalize`` only
+      inspected ``data`` and so silently let a 9× 2048² Flux2 batch
+      ride the wire as a 200 MB POST body — defeating the entire
+      reason the cap exists. The new path computes the summed raw
+      byte budget (``sum(len(b64)) * 3 // 4`` — base64 expands 4:3 so
+      this is the exact payload the server would receive) and, if it
+      crosses the cap, uploads each frame as an individual presigned
+      PUT in parallel via :func:`asyncio.gather` then rewrites the
+      envelope to ``encoding=png_base64_batch_uri`` with a parallel
+      ``uris`` list. ``shape`` / ``dtype`` / ``color_space`` ride
+      through unchanged so the server-side decoder still sees the
+      batch dimension. Per-frame URIs (NOT a tar/zip blob) compose
+      with the existing per-frame fetch helpers and let the server
+      fetch in parallel.
 
     No-op when either argument is falsy, the envelope already has a
-    ``uri``, or the inline payload fits under the cap.
+    ``uri`` / ``uris``, or the inline payload fits under the cap.
+
+    Dispatch keys off the ``encoding`` string — NOT field presence
+    (``frames`` vs. ``data``). An encoder that emitted ``frames`` for
+    a non-batch encoding by accident, or shipped both fields together,
+    used to silently take the batch branch and externalise garbage;
+    keying off the canonical encoding catches that as a hard ``frames``
+    check inside the batch branch itself.
     """
     if not server_url or not max_inline_bytes:
         return envelope
+
+    encoding = envelope.get("encoding")
+    if encoding == "png_base64_batch":
+        frames = envelope.get("frames")
+        if not isinstance(frames, list) or not frames:
+            # Mirror the server-side make_image_batch_envelope contract:
+            # an empty/missing frames list is a hard wire bug, NOT a
+            # silent no-op (which used to ship through under the old
+            # ``isinstance(frames, list) and frames`` dispatch and then
+            # blow up on the server's resolve_image_envelope_pngs check).
+            raise RnpProtocolError(
+                "png_base64_batch envelope requires a non-empty 'frames' list",
+                code=ErrorCode.INTERNAL,
+            )
+        return await _maybe_externalize_batch(
+            envelope, frames,
+            server_url=server_url,
+            max_inline_bytes=max_inline_bytes,
+            auth_headers=auth_headers,
+        )
+
     data = envelope.get("data")
     if not isinstance(data, str):
         return envelope
@@ -84,22 +135,122 @@ async def maybe_externalize(
     return out
 
 
+async def _maybe_externalize_batch(
+    envelope: dict[str, Any],
+    frames: list[Any],
+    *,
+    server_url: str,
+    max_inline_bytes: int,
+    auth_headers: dict[str, str] | None,
+) -> dict[str, Any]:
+    """Externalise a multi-frame IMAGE envelope when its summed inline
+    payload overflows ``max_inline_bytes``.
+
+    Per-frame upload (NOT a tar/zip blob): each frame becomes its own
+    presigned-PUT asset so the server can fetch in parallel and the
+    existing per-frame ``resolve_image_envelope_pngs`` helper picks
+    up the URI shape transparently. ``asyncio.gather`` issues every
+    PUT in parallel — the dominant latency for a 9-frame Flux2 batch
+    is the upload, not the server hop, so serialising would multiply
+    end-to-end submit time by N. Frame ordering is preserved (gather
+    returns positional results) — critical because a Flux2 provider
+    maps ``uris[k]`` → ``input_image_{k+1}`` 1:1.
+    """
+    # Cap on raw byte count (base64 expands 4:3, so multiplying by
+    # 3/4 recovers the actual payload the server would receive).
+    total_raw = 0
+    for f in frames:
+        if isinstance(f, str):
+            total_raw += len(f) * 3 // 4
+    if total_raw <= max_inline_bytes:
+        return envelope
+
+    env_type = envelope.get("type", "")
+    content_type = _CONTENT_TYPE_FOR_TYPE.get(env_type, "application/octet-stream")
+
+    async def _upload_one(idx: int, frame_b64: Any) -> str:
+        if not isinstance(frame_b64, str):
+            raise RnpProtocolError(
+                f"image batch frame {idx} is not a base64 string "
+                f"(got {type(frame_b64).__name__})",
+                code=ErrorCode.INTERNAL,
+            )
+        raw = base64.b64decode(frame_b64)
+        return await rnp_client.upload_asset(
+            server_url, raw,
+            content_type=content_type,
+            auth_headers=auth_headers,
+        )
+
+    uris = await asyncio.gather(
+        *(_upload_one(i, f) for i, f in enumerate(frames))
+    )
+    log.info(
+        "RNP: externalized %s batch envelope (%d frames, %d raw bytes) "
+        "-> %d presigned URIs",
+        env_type, len(frames), total_raw, len(uris),
+    )
+
+    # Strip the inline ``frames`` field; preserve ``shape`` / ``dtype``
+    # / ``color_space`` so the server-side decoder still sees the
+    # batch dimension and per-frame metadata. Encoding flips to the
+    # URI variant so dispatch keys cleanly off the string (NOT field
+    # presence — frames vs. uris is a payload field, not a shape
+    # selector).
+    out: dict[str, Any] = {
+        k: v for k, v in envelope.items() if k != "frames"
+    }
+    out["encoding"] = "png_base64_batch_uri"
+    out["uris"] = list(uris)
+    return out
+
+
 # ---------------------------------------------------------------------------
 # Image encode / decode
 # ---------------------------------------------------------------------------
 
-def encode_image_tensor(tensor: Any) -> dict[str, Any]:
-    """Encode a ComfyUI IMAGE tensor (B,H,W,C float32 in 0..1) as a
-    PNG-base64 image envelope.
+def encode_image_tensor(tensor: Any, *, accepts_batch: bool = False) -> dict[str, Any]:
+    """Encode a ComfyUI IMAGE tensor (B,H,W,C float32 in 0..1) as an
+    image envelope.
 
-    Multi-image batches send only the first frame for now; the
-    framework hasn't grown batch-of-envelopes semantics yet.
+    ``accepts_batch`` reflects whether the descriptor's
+    ``input_serialization`` for this input lists ``"png_base64_batch"``.
+    When True AND ``tensor.shape[0] > 1``, emit a multi-frame
+    ``png_base64_batch`` envelope (``frames=[<png_b64>, …]``) so the
+    server decodes a real ``[B, H, W, C]`` source. Otherwise emit a
+    single-frame ``png_base64`` envelope for ``tensor[0]`` — preserves
+    backwards compat with descriptors that haven't migrated yet, and
+    matches the wire shape of the existing ``_decode_image_batch_envelope``
+    on the response path so both directions round-trip symmetrically.
     """
     import numpy as np
     from PIL import Image
 
-    # Strip the batch dim if present.
-    if hasattr(tensor, "dim") and tensor.dim() == 4:
+    rank = getattr(tensor, "dim", lambda: 0)()
+    if rank == 4 and accepts_batch and int(tensor.shape[0]) > 1:
+        frames_arr = (tensor.cpu().clamp(0.0, 1.0).numpy() * 255).astype(np.uint8)
+        b, height, width = frames_arr.shape[:3]
+        channels = frames_arr.shape[3] if frames_arr.ndim == 4 else 1
+        pil_mode = {1: "L", 3: "RGB", 4: "RGBA"}.get(channels, "RGB")
+        frames_b64: list[str] = []
+        for i in range(b):
+            img = Image.fromarray(frames_arr[i], mode=pil_mode)
+            buf = BytesIO()
+            img.save(buf, format="PNG")
+            frames_b64.append(base64.b64encode(buf.getvalue()).decode("ascii"))
+        return {
+            "type":        "image",
+            "encoding":    "png_base64_batch",
+            "frames":      frames_b64,
+            "shape":       [b, height, width, channels],
+            "dtype":       "uint8",
+            "color_space": "srgb",
+        }
+
+    # Single-frame fallback: matches the old ``encode_image_tensor``
+    # contract (sends ``tensor[0]`` for a multi-frame batch when the
+    # descriptor doesn't advertise the batch encoding).
+    if rank == 4:
         frame = tensor[0]
     else:
         frame = tensor
@@ -121,8 +272,18 @@ def encode_image_tensor(tensor: Any) -> dict[str, Any]:
 
 
 def decode_image_envelope(envelope: dict[str, Any]) -> Any:
-    """Decode an image envelope into a ComfyUI IMAGE tensor (B,H,W,C float32)."""
+    """Decode an image envelope into a ComfyUI IMAGE tensor.
+
+    Single-frame encodings (``png_base64``, ``jpeg_base64``,
+    ``webp_base64``) decode into ``[1, H, W, C]``. The multi-frame
+    ``png_base64_batch`` encoding (a list of base64-PNG ``frames``)
+    stacks each frame into ``[B, H, W, C]`` — the native ComfyUI
+    image-batch shape — so providers whose upstream returns ``n > 1``
+    surface as a real batched tensor in the workflow runtime.
+    """
     encoding = envelope.get("encoding")
+    if encoding == "png_base64_batch":
+        return _decode_image_batch_envelope(envelope)
     if encoding not in ("png_base64", "jpeg_base64", "webp_base64"):
         raise RnpProtocolError(
             f"Unsupported image encoding: {encoding!r}",
@@ -141,6 +302,69 @@ def decode_image_envelope(envelope: dict[str, Any]) -> Any:
     return torch.from_numpy(arr).unsqueeze(0)
 
 
+def _decode_image_batch_envelope(envelope: dict[str, Any]) -> Any:
+    """Decode a ``png_base64_batch`` envelope into ``[B, H, W, C]``.
+
+    The wire shape is ``{"type":"image", "encoding":"png_base64_batch",
+    "frames":[<png_b64>, ...], "shape":[B,H,W,C], ...}``. Each frame is
+    a PNG-encoded image; all frames must share the same H/W after the
+    RGB conversion (the server-side encoder enforces this). A frame
+    count mismatch against the ``shape[0]`` field surfaces as a hard
+    parse error rather than silently dropping frames — matches the SVG
+    decoder's ``count`` integrity check.
+    """
+    import base64 as _b64
+    import numpy as np
+    import torch
+    from PIL import Image
+
+    frames = envelope.get("frames")
+    if not isinstance(frames, list) or not frames:
+        raise RnpProtocolError(
+            "image batch envelope missing or empty 'frames' list",
+            code=ErrorCode.INTERNAL,
+        )
+    declared_shape = envelope.get("shape")
+    if isinstance(declared_shape, list) and len(declared_shape) >= 1:
+        declared_b = int(declared_shape[0])
+        if declared_b != len(frames):
+            raise RnpProtocolError(
+                f"image batch envelope frame count {len(frames)} does not "
+                f"match declared shape[0]={declared_b}",
+                code=ErrorCode.INTERNAL,
+            )
+
+    arrs: list[Any] = []
+    base_hw: tuple[int, int] | None = None
+    for idx, frame_b64 in enumerate(frames):
+        if not isinstance(frame_b64, str):
+            raise RnpProtocolError(
+                f"image batch frame {idx} is not a base64 string",
+                code=ErrorCode.INTERNAL,
+            )
+        try:
+            raw = _b64.b64decode(frame_b64)
+            img = Image.open(BytesIO(raw)).convert("RGB")
+        except Exception as e:
+            raise RnpProtocolError(
+                f"image batch frame {idx} failed to decode: {e}",
+                code=ErrorCode.INTERNAL,
+            ) from e
+        arr = np.array(img).astype(np.float32) / 255.0
+        if base_hw is None:
+            base_hw = (arr.shape[0], arr.shape[1])
+        elif (arr.shape[0], arr.shape[1]) != base_hw:
+            raise RnpProtocolError(
+                f"image batch frame {idx} size {arr.shape[:2]} does not "
+                f"match frame 0 size {base_hw} (server should have rejected "
+                f"this; likely a wire bug)",
+                code=ErrorCode.INTERNAL,
+            )
+        arrs.append(arr)
+    stacked = np.stack(arrs, axis=0)
+    return torch.from_numpy(stacked)
+
+
 # ---------------------------------------------------------------------------
 # Mask encode / decode
 # ---------------------------------------------------------------------------
@@ -227,7 +451,11 @@ def encode_audio_input(audio: dict[str, Any]) -> dict[str, Any]:
 def decode_audio_envelope(envelope: dict[str, Any]) -> dict[str, Any]:
     """Decode an audio envelope back into a ComfyUI AUDIO dict."""
     encoding = envelope.get("encoding")
-    if encoding not in ("mp3_base64", "mp3_inline", "wav_base64", "wav_inline"):
+    if encoding not in (
+        "mp3_base64", "mp3_inline",
+        "wav_base64", "wav_inline",
+        "opus_base64",
+    ):
         raise RnpProtocolError(
             f"Unsupported audio encoding: {encoding!r}",
             code=ErrorCode.INTERNAL,
@@ -237,47 +465,9 @@ def decode_audio_envelope(envelope: dict[str, Any]) -> dict[str, Any]:
 
 
 # ---------------------------------------------------------------------------
-# Video encode / decode
+# Video decode (encode lands when a remote node accepts VIDEO inputs)
 # ---------------------------------------------------------------------------
 
-def encode_video_input(video: Any) -> dict[str, Any]:
-    """Encode a ComfyUI ``Input.Video`` object as an mp4-base64 video
-    envelope.
-
-    Mirrors the local ``upload_video_to_comfyapi`` two-step (allocate
-    presigned slot + PUT) but bundles the bytes inline so the server
-    sees them in one POST. Re-encodes the source through
-    ``video.save_to(buf, format=MP4, codec=H264)`` to normalize the
-    container — Bria's ``BriaRemoveVideoBackground`` (and any other
-    ``output_container_and_codec="mp4_h264"`` vendor) only accepts
-    MP4/H264.
-    """
-    from comfy_api.latest import Types
-
-    buf = BytesIO()
-    video.save_to(
-        buf,
-        format=Types.VideoContainer.MP4,
-        codec=Types.VideoCodec.H264,
-    )
-    body = buf.getvalue()
-    extra: dict[str, Any] = {}
-    try:
-        duration_s = float(video.get_duration())
-        extra["duration_s"] = duration_s
-    except Exception:  # noqa: BLE001
-        # ``get_duration`` is best-effort on the Video protocol —
-        # missing duration just means the server can't pre-validate
-        # against a duration bound.
-        pass
-    return {
-        "type":     "video",
-        "encoding": "mp4_base64",
-        "data":     base64.b64encode(body).decode("ascii"),
-        **extra,
-    }
-
-
 def decode_video_envelope(envelope: dict[str, Any]) -> Any:
     """Decode a video envelope into a ComfyUI Video object (mp4 inline)."""
     encoding = envelope.get("encoding")
@@ -290,83 +480,31 @@ def decode_video_envelope(envelope: dict[str, Any]) -> Any:
     return InputImpl.VideoFromFile(BytesIO(decode_envelope_data(envelope)))
 
 
-def is_video_input(value: Any) -> bool:
-    """Best-effort check for a ComfyUI Video object (``Input.Video``).
-
-    Detected by duck-typing on ``save_to`` + ``get_duration`` — both
-    are required by the ``VideoInput`` protocol in
-    ``comfy_api.latest._input.video_types``. Avoids importing
-    ``comfy_api`` at decode-dispatch time so this module stays cheap
-    to import in a non-ComfyUI process.
-    """
-    return (
-        not isinstance(value, dict)
-        and hasattr(value, "save_to")
-        and hasattr(value, "get_duration")
-    )
-
-
 # ---------------------------------------------------------------------------
-# SVG decode (encode lands when a remote node accepts SVG inputs; today
-# Recraft's V3/V4 vector endpoints only emit SVG, so the client is
-# decode-only).
+# 3D model decode (encode lands when a remote node accepts MODEL_3D inputs)
 # ---------------------------------------------------------------------------
 
-def decode_svg_envelope(envelope: dict[str, Any]) -> Any:
-    """Decode an SVG envelope into a ``comfy_extras.nodes_images.SVG`` instance.
+def decode_model3d_envelope(envelope: dict[str, Any]) -> Any:
+    """Decode a 3D-model envelope into a ComfyUI ``File3D`` object.
 
-    The wire payload is one or more concatenated ``<svg>…</svg>`` XML
-    documents; we split on the closing tag so a multi-frame Recraft
-    response (n>1) hydrates a single ``SVG`` carrying a list of
-    BytesIO frames — the same shape the local node produces via
-    ``SVG(svg_data)``.
+    Mirrors :func:`decode_video_envelope`: the only currently supported
+    encoding is ``glb_inline`` (binary glTF 2.0 — magic ``b"glTF"`` at
+    offset 0). The bytes are handed to ``File3D`` as a ``BytesIO`` so
+    downstream nodes (preview3d / SaveGLB / Load3D) see exactly the
+    same value shape they would from a local Tripo / Rodin / Meshy
+    node. The envelope's optional ``format`` extra is forwarded as the
+    second constructor arg so the materialiser picks the right file
+    extension when persisting.
     """
     encoding = envelope.get("encoding")
-    if encoding not in ("svg_xml_base64", "svg_xml_inline"):
+    if encoding != "glb_inline":
         raise RnpProtocolError(
-            f"Unsupported svg encoding: {encoding!r}",
+            f"Unsupported model_3d encoding: {encoding!r}",
             code=ErrorCode.INTERNAL,
         )
-    from comfy_extras.nodes_images import SVG
-
-    raw = decode_envelope_data(envelope)
-    text = raw.decode("utf-8", "replace")
-    # The server joins per-frame SVGs with ``\n`` (see
-    # ``_emit_svg_from_urls``); split on the closing tag followed by a
-    # newline so an SVG document that legitimately contains ``</svg>``
-    # in CDATA / a foreignObject / a comment isn't shredded mid-frame.
-    # Fall back to splitting on the bare closing tag for single-frame
-    # payloads or older server builds that didn't add the newline.
-    delimiter = "</svg>\n"
-    if delimiter not in text:
-        delimiter = "</svg>"
-    parts: list[BytesIO] = []
-    cursor = 0
-    closing = "</svg>"
-    while True:
-        idx = text.find(delimiter, cursor)
-        if idx < 0:
-            tail = text[cursor:].strip()
-            if tail:
-                parts.append(BytesIO(tail.encode("utf-8")))
-            break
-        end = idx + len(closing)  # keep the closing tag on this frame
-        chunk = text[cursor:end].strip()
-        if chunk:
-            parts.append(BytesIO(chunk.encode("utf-8")))
-        cursor = idx + len(delimiter)
-    declared = envelope.get("count")
-    if isinstance(declared, int) and declared > 0 and declared != len(parts):
-        log.warning(
-            "SVG envelope count=%d but parsed %d documents — server payload may be malformed",
-            declared, len(parts),
-        )
-    if not parts:
-        raise RnpProtocolError(
-            "SVG envelope payload contained no <svg> documents",
-            code=ErrorCode.INTERNAL,
-        )
-    return SVG(parts)
+    from comfy_api.latest._util import File3D
+    file_format = envelope.get("format") or "glb"
+    return File3D(BytesIO(decode_envelope_data(envelope)), str(file_format))
 
 
 # ---------------------------------------------------------------------------
@@ -374,11 +512,11 @@ def decode_svg_envelope(envelope: dict[str, Any]) -> Any:
 # ---------------------------------------------------------------------------
 
 _DECODERS = {
-    "image": decode_image_envelope,
-    "mask":  decode_mask_envelope,
-    "audio": decode_audio_envelope,
-    "video": decode_video_envelope,
-    "svg":   decode_svg_envelope,
+    "image":    decode_image_envelope,
+    "mask":     decode_mask_envelope,
+    "audio":    decode_audio_envelope,
+    "video":    decode_video_envelope,
+    "model_3d": decode_model3d_envelope,
 }
 
 
@@ -427,13 +565,11 @@ __all__ = [
     "decode_mask_envelope",
     "encode_audio_input",
     "decode_audio_envelope",
-    "encode_video_input",
     "decode_video_envelope",
-    "decode_svg_envelope",
+    "decode_model3d_envelope",
     "decode_envelope",
     "is_envelope",
     "is_image_tensor",
     "is_mask_tensor",
     "is_audio_input",
-    "is_video_input",
 ]