feat(serialization): add VIDEO encode + VIDEO_MP4_BASE64 capability

Mirrors the AUDIO sweep: adds encode_video_input / is_video_input
helpers in serialization.py, wires them into proxy_node._encode_one,
and advertises VIDEO_MP4_BASE64 in CLIENT_CAPABILITIES so servers can
require it for VIDEO inputs.

encode_video_input re-encodes through video.save_to(buf, MP4, H264) to
normalize the container — Bria's video endpoints (the first vendor with
VIDEO inputs) only accept mp4_h264. Populates an optional duration_s
extra from get_duration() so servers can pre-validate against duration
bounds without demuxing.

Amp-Thread-ID: https://ampcode.com/threads/T-019e210b-f792-7389-b8db-8e02c81e4295
Co-authored-by: Amp <amp@ampcode.com>

client.py

@@ -28,95 +28,25 @@ from .protocol import (
 log = logging.getLogger("comfy_remote_nodes.client")
 
 
-# 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).
+# 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.
 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,
-    # 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,
-    # 3D MODEL BUNDLE envelope (multi-file: primary mesh plus N
-    # companion files — textures / .mtl / .bin / etc. — as a single
-    # coherent unit). The decoder in
-    # ``serialization.decode_model3d_envelope`` dispatches on
-    # ``encoding="bundle_inline"`` and returns a ``BundledFile3D``
-    # (a ``File3D`` subclass) whose ``get_source()`` / ``save_to()``
-    # materialise the whole bundle on disk so cross-file refs (.obj
-    # -> .mtl -> textures; .gltf -> .bin + textures) resolve via
-    # the filesystem. Servers gate bundle-emitting descriptors on
-    # this token; legacy clients surface NEGOTIATION_FAILED at
-    # descriptor-load time rather than crashing at execute.
-    Capability.MODEL_3D_BUNDLE_INLINE,
-    # Cross-node task handle envelope (vendor-native task / video /
-    # job IDs with lineage metadata for chained partner provider
-    # nodes — Tripo Texture/Refine/Rig/Retarget/Conversion, Kling
-    # VideoExtend, future partner chains). The decoder in
-    # ``serialization.decode_task_handle_envelope`` returns a
-    # ``TaskHandle`` dataclass rather than a plain ``str`` so the
-    # ``parent_chain`` lineage round-trips when the handle is passed
-    # through to a downstream provider node (re-encoded by
-    # ``serialization.encode_task_handle`` on the submit path).
-    # Servers gate handle-emitting AND handle-accepting descriptors
-    # on this token (whole-descriptor gating — partial-socket
-    # synthesis is rejected because a chain that breaks meaningfully
-    # if either end is missing is worse than no chain at all).
-    Capability.IO_TASK_HANDLE,
+    Capability.SVG_XML_BASE64,
 ]
 
 # Reported in the X-RNP-Client-Version header.

notes/run_batch_uri_followups.py

@@ -1,352 +0,0 @@
-"""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_dynamic_combo_parser.py

@@ -1,526 +0,0 @@
-"""Smoke test for the DYNAMIC_COMBO + AUTOGROW parser extension.
-
-Verifies the proxy_node parser now constructs real V3
-``IO.DynamicCombo.Input`` and ``IO.Autogrow.Input`` objects from the
-flattened RNP wire shapes that the server already emits, instead of
-falling through to the opaque ``IO.Custom`` bucket.
-
-Checks:
-
-1.  Bria-shaped 2-branch DYNAMIC_COMBO (the "false" branch has empty
-    ``inputs``; the "true" branch has 3 BOOLEAN sub-widgets) parses
-    into an ``IO.DynamicCombo.Input`` whose two ``Option``s carry the
-    right keys + sub-input objects. Proves empty branches are kept and
-    that the outer id is whatever the descriptor key was (here
-    ``"moderation"``, *not* ``"model"``).
-2.  Grok-V2-shaped DYNAMIC_COMBO with three branches, each containing
-    a nested AUTOGROW(IMAGE) plus a COMBO + INT (and an aspect_ratio
-    COMBO on 2 of the 3 branches). Asserts the AUTOGROW becomes a real
-    ``IO.Autogrow.Input`` with a ``TemplatePrefix`` whose underlying
-    input is ``IO.Image.Input``, and that prefix / min / max survive.
-3.  ElevenLabs-shaped DYNAMIC_COMBO whose branches mix
-    ``IO.Custom(<opaque_io_type>).Input`` (voiceN custom-IO) with
-    primitive STRING sub-widgets — proves the recursive opaque
-    fallback path inside a branch still works for partner helper
-    types.
-4.  Synthetic nested DYNAMIC_COMBO-inside-DYNAMIC_COMBO — no provider
-    uses this today, but defensive coverage so the recursive entry
-    point keeps working.
-5.  Malformed DYNAMIC_COMBO / AUTOGROW shapes return ``None`` (so the
-    caller skips the descriptor entirely) instead of silently
-    falling through to the opaque IO.Custom bucket.
-6.  Regression: ``_collect_local_validate`` output is byte-identical
-    before and after the parser change (the rules-extraction path is
-    independent of input construction and must not regress).
-7.  AUTOGROW template that is itself a dynamic input is rejected
-    (upstream Autogrow asserts this).
-8.  Source-level check: the proxy_node docstring + dispatch covers
-    DYNAMIC_COMBO and AUTOGROW (not just the existing primitive set).
-
-Run with the ComfyUI venv (needs torch / comfy_api on PYTHONPATH):
-
-    python notes/run_dynamic_combo_parser.py
-"""
-from __future__ import annotations
-
-import importlib.util
-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))
-
-
-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 all the heavy deps proxy_node + comfy_api/latest/_io.py pull in.
-#
-# We only need three things from the world:
-#
-# * the real V3 IO classes (DynamicCombo / Autogrow / Image / Combo / String /
-#   Int / Boolean / Custom) — load ``_io.py`` standalone so we don't pay the
-#   full ``comfy_api.latest`` package init (PIL, numpy, comfy.cli_args,
-#   comfy_execution, the sync-class generator etc.).
-# * the real proxy_node parser helpers — load ``proxy_node.py`` after the
-#   ``comfy_api.latest`` shim is in place.
-# * the real protocol module (lightweight; for the ErrorCode / Header / etc.
-#   imports proxy_node does up top).
-#
-# Everything else (server.PromptServer, aiohttp, comfy_api_nodes.*) is
-# stubbed because proxy_node only references those at request-time, not at
-# import-time-after-trivial-stub.
-# ---------------------------------------------------------------------------
-
-# 1) Stub ``torch`` — _io.py uses ``torch.Tensor`` only in type annotations
-#    (``Type = torch.Tensor`` class-level field), so a bare-class stub works.
-_torch_stub = types.ModuleType("torch")
-class _Tensor:  # noqa: D401
-    pass
-_torch_stub.Tensor = _Tensor
-sys.modules.setdefault("torch", _torch_stub)
-
-# 2) Stub ``comfy_execution.graph_utils.ExecutionBlocker`` — only used as a
-#    class reference inside Output construction; never instantiated here.
-_comfy_execution = types.ModuleType("comfy_execution")
-_graph_utils = types.ModuleType("comfy_execution.graph_utils")
-class _ExecutionBlocker:  # noqa: D401
-    pass
-_graph_utils.ExecutionBlocker = _ExecutionBlocker
-sys.modules.setdefault("comfy_execution", _comfy_execution)
-sys.modules.setdefault("comfy_execution.graph_utils", _graph_utils)
-
-# 3) Stub ``comfy_api.internal`` — _io.py imports a handful of helper names.
-#    The parser code paths exercised here don't actually invoke any of these;
-#    bare-function / bare-class stubs are sufficient to satisfy the import.
-_comfy_api = types.ModuleType("comfy_api")
-_comfy_api.__path__ = []
-_comfy_api_internal = types.ModuleType("comfy_api.internal")
-class _ComfyNodeInternal:  # noqa: D401
-    pass
-class _NodeOutputInternal:  # noqa: D401
-    pass
-class _classproperty:  # noqa: D401
-    def __init__(self, f):
-        self.f = f
-    def __get__(self, instance, owner):
-        return self.f(owner)
-def _identity(x, *_args, **_kw):
-    return x
-def _first_real_override(*_a, **_kw):
-    return None
-def _is_class(x):
-    return isinstance(x, type)
-def _prune_dict(d):
-    return {k: v for k, v in (d or {}).items() if v is not None}
-_comfy_api_internal._ComfyNodeInternal = _ComfyNodeInternal
-_comfy_api_internal._NodeOutputInternal = _NodeOutputInternal
-_comfy_api_internal.classproperty = _classproperty
-_comfy_api_internal.copy_class = _identity
-_comfy_api_internal.first_real_override = _first_real_override
-_comfy_api_internal.is_class = _is_class
-_comfy_api_internal.prune_dict = _prune_dict
-_comfy_api_internal.shallow_clone_class = _identity
-sys.modules.setdefault("comfy_api", _comfy_api)
-sys.modules.setdefault("comfy_api.internal", _comfy_api_internal)
-
-# 3b) Stub ``comfy_api.input`` for the lazy imports inside _io.py's
-#     Curve / RangeInput class bodies — the parser code never touches
-#     those types, but the *class body* runs at module load.
-_comfy_api_input = types.ModuleType("comfy_api.input")
-class _CurvePoint: pass  # noqa: E701,D401
-class _RangeInput: pass  # noqa: E701,D401
-_comfy_api_input.CurvePoint = _CurvePoint
-_comfy_api_input.RangeInput = _RangeInput
-sys.modules.setdefault("comfy_api.input", _comfy_api_input)
-
-# 4) Stub ``comfy_api.latest._util`` — _io.py imports MESH / VOXEL / SVG /
-#    File3D as type-token classes. We don't construct any here.
-_comfy_api_latest = types.ModuleType("comfy_api.latest")
-_comfy_api_latest.__path__ = []
-_comfy_api_latest_util = types.ModuleType("comfy_api.latest._util")
-class _MESH: pass  # noqa: E701,D401
-class _VOXEL: pass  # noqa: E701,D401
-class _SVG: pass  # noqa: E701,D401
-class _File3D: pass  # noqa: E701,D401
-_comfy_api_latest_util.MESH = _MESH
-_comfy_api_latest_util.VOXEL = _VOXEL
-_comfy_api_latest_util.SVG = _SVG
-_comfy_api_latest_util.File3D = _File3D
-sys.modules.setdefault("comfy_api.latest", _comfy_api_latest)
-sys.modules.setdefault("comfy_api.latest._util", _comfy_api_latest_util)
-
-# 5) Load the real ``_io.py`` as ``comfy_api.latest._io`` — its
-#    relative ``from ._util import ...`` requires the parent-package
-#    setup we just did.
-_io_path = os.path.join(_COMFYUI_DIR, "comfy_api", "latest", "_io.py")
-_io_mod = _load_module("comfy_api.latest._io", _io_path)
-# Expose ``IO`` exactly like the real package does
-# (``from . import _io_public as io; IO = io`` where _io_public is just
-# ``from ._io import *``).
-_comfy_api_latest.IO = _io_mod
-_comfy_api_latest.io = _io_mod
-IO = _io_mod  # local alias for the assertions below
-
-# 6) Stub proxy_node's other heavyweight imports.
-for missing in (
-    "server",
-    "aiohttp",
-    "comfy_api_nodes",
-    "comfy_api_nodes.util",
-    "comfy_api_nodes.util.client",
-    "comfy_api_nodes.util.common_exceptions",
-):
-    if missing not in sys.modules:
-        m = types.ModuleType(missing)
-        sys.modules[missing] = m
-        m.__path__ = []  # type: ignore[attr-defined]
-sys.modules["server"].PromptServer = type("_StubPS", (), {})
-sys.modules["comfy_api_nodes.util.client"].ApiEndpoint = type("_StubAE", (), {})
-sys.modules["comfy_api_nodes.util.client"].poll_op_raw = lambda *a, **k: None
-sys.modules["comfy_api_nodes.util.common_exceptions"].ProcessingInterrupted = type(
-    "_StubPI", (Exception,), {},
-)
-
-# 7) Emulate the package layout for the client so its relative imports
-#    work (``from .protocol import ...`` etc.).
-pkg = types.ModuleType("comfy_remote_nodes_test")
-pkg.__path__ = [_CLIENT_DIR]
-sys.modules["comfy_remote_nodes_test"] = pkg
-sys.modules["comfy_remote_nodes_test.client"] = types.ModuleType(
-    "comfy_remote_nodes_test.client",
-)
-sys.modules["comfy_remote_nodes_test.serialization"] = types.ModuleType(
-    "comfy_remote_nodes_test.serialization",
-)
-_load_module(
-    "comfy_remote_nodes_test.protocol",
-    os.path.join(_CLIENT_DIR, "protocol.py"),
-)
-
-proxy_node = _load_module(
-    "comfy_remote_nodes_test.proxy_node",
-    os.path.join(_CLIENT_DIR, "proxy_node.py"),
-)
-
-
-# ---------------------------------------------------------------------------
-# 1. Bria-shaped 2-branch DYNAMIC_COMBO.
-# ---------------------------------------------------------------------------
-bria_spec = [
-    "DYNAMIC_COMBO",
-    {
-        "options": [
-            {"key": "false", "inputs": []},
-            {
-                "key": "true",
-                "inputs": [
-                    ["prompt_content_moderation", "BOOLEAN", {"default": False}],
-                    ["visual_input_moderation", "BOOLEAN", {"default": False}],
-                    ["visual_output_moderation", "BOOLEAN", {"default": True}],
-                ],
-            },
-        ],
-        "tooltip": "Moderation settings",
-    },
-]
-inp = proxy_node._parse_input_spec("moderation", bria_spec, optional=False)
-assert inp is not None, "Bria DYNAMIC_COMBO returned None"
-assert isinstance(inp, IO.DynamicCombo.Input), (
-    f"Expected IO.DynamicCombo.Input, got {type(inp).__name__}"
-)
-assert inp.id == "moderation", f"outer id should track descriptor key, got {inp.id!r}"
-assert len(inp.options) == 2, f"expected 2 branches, got {len(inp.options)}"
-assert [o.key for o in inp.options] == ["false", "true"], (
-    f"branch keys wrong: {[o.key for o in inp.options]}"
-)
-# Empty branch preserved (UX needs the dropdown entry).
-assert inp.options[0].inputs == [], "empty 'false' branch dropped"
-assert len(inp.options[1].inputs) == 3, (
-    f"'true' branch should have 3 sub-inputs, got {len(inp.options[1].inputs)}"
-)
-for sub in inp.options[1].inputs:
-    assert isinstance(sub, IO.Boolean.Input), (
-        f"branch sub-input should be IO.Boolean.Input, got {type(sub).__name__}"
-    )
-assert [s.id for s in inp.options[1].inputs] == [
-    "prompt_content_moderation",
-    "visual_input_moderation",
-    "visual_output_moderation",
-], "branch sub-input names lost"
-print("PASS: Bria DYNAMIC_COMBO parses with empty + populated branches.")
-
-
-# ---------------------------------------------------------------------------
-# 2. Grok-V2-shaped DYNAMIC_COMBO with nested AUTOGROW(IMAGE) per branch.
-# ---------------------------------------------------------------------------
-def _grok_v2_subinputs(*, max_ref_images: int, with_aspect_ratio: bool) -> list:
-    inputs: list = [
-        [
-            "images",
-            "AUTOGROW",
-            {
-                "template": [
-                    "IMAGE",
-                    {"local_validate": {"image_max_batch": max_ref_images}},
-                ],
-                "prefix": "image",
-                "min": 1,
-                "max": max_ref_images,
-                "tooltip": (
-                    "Reference image to edit."
-                    if max_ref_images == 1
-                    else f"Reference image(s) to edit. Up to {max_ref_images} images."
-                ),
-            },
-        ],
-        ["resolution", "COMBO", {"options": ["768p", "1k"]}],
-        ["number_of_images", "INT", {
-            "default": 1, "min": 1, "max": 10, "step": 1,
-        }],
-    ]
-    if with_aspect_ratio:
-        inputs.append(["aspect_ratio", "COMBO", {
-            "options": ["auto", "1:1", "16:9"], "default": "auto",
-        }])
-    return inputs
-
-grok_spec = [
-    "DYNAMIC_COMBO",
-    {
-        "options": [
-            {
-                "key": "grok-imagine-image-quality",
-                "inputs": _grok_v2_subinputs(max_ref_images=3, with_aspect_ratio=True),
-            },
-            {
-                "key": "grok-imagine-image-pro",
-                "inputs": _grok_v2_subinputs(max_ref_images=1, with_aspect_ratio=False),
-            },
-            {
-                "key": "grok-imagine-image",
-                "inputs": _grok_v2_subinputs(max_ref_images=3, with_aspect_ratio=True),
-            },
-        ],
-        "tooltip": "The model to use for editing.",
-    },
-]
-grok_inp = proxy_node._parse_input_spec("model", grok_spec, optional=False)
-assert grok_inp is not None, "Grok V2 DYNAMIC_COMBO returned None"
-assert isinstance(grok_inp, IO.DynamicCombo.Input), (
-    f"Expected IO.DynamicCombo.Input, got {type(grok_inp).__name__}"
-)
-assert len(grok_inp.options) == 3
-# Pro branch (max_ref_images=1, no aspect_ratio).
-pro_branch = next(o for o in grok_inp.options if o.key == "grok-imagine-image-pro")
-assert len(pro_branch.inputs) == 3, f"pro branch should have 3 sub-inputs, got {len(pro_branch.inputs)}"
-images_inp = pro_branch.inputs[0]
-assert isinstance(images_inp, IO.Autogrow.Input), (
-    f"Expected nested AUTOGROW to parse as IO.Autogrow.Input, got "
-    f"{type(images_inp).__name__}"
-)
-assert images_inp.id == "images"
-assert isinstance(images_inp.template, IO.Autogrow.TemplatePrefix), (
-    f"Expected TemplatePrefix template, got {type(images_inp.template).__name__}"
-)
-assert images_inp.template.prefix == "image"
-assert images_inp.template.min == 1
-assert images_inp.template.max == 1, f"pro branch max should be 1, got {images_inp.template.max}"
-assert isinstance(images_inp.template.input, IO.Image.Input), (
-    f"AUTOGROW template input should be IO.Image.Input, got "
-    f"{type(images_inp.template.input).__name__}"
-)
-# Quality branch (max_ref_images=3 + aspect_ratio).
-qual_branch = next(o for o in grok_inp.options if o.key == "grok-imagine-image-quality")
-assert len(qual_branch.inputs) == 4, f"quality branch should have 4 sub-inputs, got {len(qual_branch.inputs)}"
-qual_images = qual_branch.inputs[0]
-assert isinstance(qual_images, IO.Autogrow.Input)
-assert qual_images.template.max == 3, f"quality branch max should be 3, got {qual_images.template.max}"
-# Primitive widgets after the autogrow.
-assert isinstance(qual_branch.inputs[1], IO.Combo.Input)
-assert isinstance(qual_branch.inputs[2], IO.Int.Input)
-assert isinstance(qual_branch.inputs[3], IO.Combo.Input)
-print("PASS: Grok V2 DYNAMIC_COMBO + nested AUTOGROW parses correctly.")
-
-
-# ---------------------------------------------------------------------------
-# 3. ElevenLabs-shaped branch with custom-IO sub-inputs (voiceN).
-# ---------------------------------------------------------------------------
-el_spec = [
-    "DYNAMIC_COMBO",
-    {
-        "options": [
-            {
-                "key": "2",
-                "inputs": [
-                    ["voice0", "ELEVENLABS_VOICE", {}],
-                    ["text0", "STRING", {"multiline": True, "default": ""}],
-                    ["voice1", "ELEVENLABS_VOICE", {}],
-                    ["text1", "STRING", {"multiline": True, "default": ""}],
-                ],
-            },
-        ],
-        "tooltip": "Number of dialogue turns.",
-    },
-]
-el_inp = proxy_node._parse_input_spec("inputs", el_spec, optional=False)
-assert el_inp is not None, "ElevenLabs DYNAMIC_COMBO returned None"
-assert isinstance(el_inp, IO.DynamicCombo.Input)
-assert el_inp.id == "inputs", (
-    f"outer id should be 'inputs' (not always 'model'), got {el_inp.id!r}"
-)
-branch = el_inp.options[0]
-# IO.Custom("FOO") returns a synthesized ComfyTypeIO subclass — each call
-# creates a *new* class, so isinstance-against-a-fresh-Custom doesn't
-# work. The @comfytype decorator stamps ``io_type`` onto the Input class
-# instead; check that the synthesized custom Input carries the right
-# string (this is how the frontend's connection-validity check chains
-# sockets).
-voice0 = branch.inputs[0]
-assert voice0.get_io_type() == "ELEVENLABS_VOICE", (
-    f"voice0 should be opaque ELEVENLABS_VOICE Input, got "
-    f"{voice0.get_io_type()!r} on {type(voice0).__name__}"
-)
-assert isinstance(branch.inputs[1], IO.String.Input)
-assert isinstance(branch.inputs[3], IO.String.Input)
-print("PASS: ElevenLabs DYNAMIC_COMBO branch mixes custom-IO + primitive.")
-
-
-# ---------------------------------------------------------------------------
-# 4. Synthetic nested DYNAMIC_COMBO inside DYNAMIC_COMBO.
-# ---------------------------------------------------------------------------
-nested = [
-    "DYNAMIC_COMBO",
-    {
-        "options": [
-            {
-                "key": "outer_a",
-                "inputs": [
-                    [
-                        "submodel",
-                        "DYNAMIC_COMBO",
-                        {
-                            "options": [
-                                {"key": "inner_x", "inputs": [
-                                    ["x", "INT", {"default": 1, "min": 0, "max": 10}],
-                                ]},
-                                {"key": "inner_y", "inputs": []},
-                            ],
-                            "tooltip": "Inner branch.",
-                        },
-                    ],
-                ],
-            },
-        ],
-        "tooltip": "Outer branch.",
-    },
-]
-nest_inp = proxy_node._parse_input_spec("model", nested, optional=False)
-assert isinstance(nest_inp, IO.DynamicCombo.Input)
-inner = nest_inp.options[0].inputs[0]
-assert isinstance(inner, IO.DynamicCombo.Input), (
-    f"nested DynamicCombo should be IO.DynamicCombo.Input, got {type(inner).__name__}"
-)
-assert [o.key for o in inner.options] == ["inner_x", "inner_y"]
-assert isinstance(inner.options[0].inputs[0], IO.Int.Input)
-print("PASS: nested DYNAMIC_COMBO-in-DYNAMIC_COMBO parses recursively.")
-
-
-# ---------------------------------------------------------------------------
-# 5. Malformed dynamic specs return None (no opaque fallthrough).
-# ---------------------------------------------------------------------------
-# Missing options list on DYNAMIC_COMBO.
-assert proxy_node._parse_input_spec("x", ["DYNAMIC_COMBO", {}], optional=False) is None
-# Option missing key.
-bad = ["DYNAMIC_COMBO", {"options": [{"inputs": []}]}]
-assert proxy_node._parse_input_spec("x", bad, optional=False) is None
-# Option's inputs entry not a 3-tuple.
-bad2 = ["DYNAMIC_COMBO", {"options": [{"key": "a", "inputs": [["x"]]}]}]
-assert proxy_node._parse_input_spec("x", bad2, optional=False) is None
-# AUTOGROW without template.
-assert proxy_node._parse_input_spec("x", ["AUTOGROW", {"prefix": "p"}], optional=False) is None
-# AUTOGROW without prefix/names.
-assert (
-    proxy_node._parse_input_spec("x", ["AUTOGROW", {"template": ["IMAGE", {}]}], optional=False)
-    is None
-)
-print("PASS: malformed DYNAMIC_COMBO / AUTOGROW specs return None.")
-
-
-# ---------------------------------------------------------------------------
-# 6. Regression: _collect_local_validate output unchanged for these inputs.
-# ---------------------------------------------------------------------------
-desc_grok = {
-    "input": {
-        "required": {"model": grok_spec},
-        "optional": {},
-    },
-}
-rules = proxy_node._collect_local_validate(desc_grok)
-# rules['model'] should carry __branches__ with per-branch maps that each
-# carry an __template__ entry under "images" with image_max_batch.
-assert "model" in rules, f"_collect_local_validate dropped 'model': {rules!r}"
-br = rules["model"].get("__branches__")
-assert isinstance(br, dict) and set(br) == {
-    "grok-imagine-image-quality",
-    "grok-imagine-image-pro",
-    "grok-imagine-image",
-}, f"branches mismatch: {br!r}"
-pro = br["grok-imagine-image-pro"]
-assert "images" in pro, f"pro branch should carry 'images' rule: {pro!r}"
-tmpl_rules = pro["images"].get("__template__")
-assert tmpl_rules == {"image_max_batch": 1}, (
-    f"pro branch template rules wrong: {tmpl_rules!r}"
-)
-qual = br["grok-imagine-image-quality"]
-assert qual["images"].get("__template__") == {"image_max_batch": 3}
-print("PASS: _collect_local_validate output unchanged (branches + template intact).")
-
-
-# ---------------------------------------------------------------------------
-# 7. AUTOGROW with a dynamic template input is rejected.
-# ---------------------------------------------------------------------------
-ag_with_dyn_template = [
-    "AUTOGROW",
-    {
-        "template": [
-            "DYNAMIC_COMBO",
-            {"options": [{"key": "k", "inputs": []}]},
-        ],
-        "prefix": "img",
-        "min": 1,
-        "max": 3,
-    },
-]
-assert (
-    proxy_node._parse_input_spec("bad", ag_with_dyn_template, optional=False) is None
-), "AUTOGROW with DYNAMIC_COMBO template should be rejected"
-print("PASS: AUTOGROW rejects dynamic template input (matches upstream assert).")
-
-
-# ---------------------------------------------------------------------------
-# 8. Source-level: docstring + dispatch mention DYNAMIC_COMBO + AUTOGROW.
-# ---------------------------------------------------------------------------
-proxy_src = open(
-    os.path.join(_CLIENT_DIR, "proxy_node.py"), "r", encoding="utf-8",
-).read()
-assert "DYNAMIC_COMBO" in proxy_src, "proxy_node source missing DYNAMIC_COMBO branch"
-assert "AUTOGROW" in proxy_src, "proxy_node source missing AUTOGROW branch"
-assert "_parse_named_input_spec" in proxy_src, (
-    "proxy_node source missing _parse_named_input_spec helper"
-)
-assert "_build_input" in proxy_src, "proxy_node source missing _build_input helper"
-print("PASS: proxy_node source contains the new dispatch + helpers.")
-
-print("ALL CHECKS PASSED.")

notes/run_image_max_batch.py

@@ -1,441 +0,0 @@
-"""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_bundle_decoder.py

@@ -1,461 +0,0 @@
-"""Smoke test for the client-side multi-file MODEL_3D bundle decoder.
-
-Verifies:
-
-1. ``Capability.MODEL_3D_BUNDLE_INLINE`` is intact at the vendored
-   protocol layer and exposed as the well-formed
-   ``model_3d:bundle_inline`` token.
-2. ``is_envelope`` recognises a ``bundle_inline`` envelope (still
-   ``type=model_3d`` — HEAVY_TYPES unchanged).
-3. ``decode_model3d_envelope`` dispatches ``encoding=bundle_inline``
-   to the new ``BundledFile3D`` class, returns a real ``File3D``
-   subclass, and round-trips the primary mesh bytes via
-   ``get_data()`` / ``get_bytes()`` without materialising anything
-   on disk.
-4. ``get_source()`` materialises the whole bundle into a temp
-   directory using the producer-supplied relative paths, so a .obj
-   that references its .mtl by filename (and a texture in
-   ``textures/``) finds them on disk via the filesystem.
-5. ``save_to(dest)`` copies the whole bundle into the destination
-   directory — the primary file under the caller-specified name,
-   sidecars under their ORIGINAL relative paths — so cross-file refs
-   inside the .obj (which point at ``model.mtl``) still resolve when
-   the user saves the model somewhere stable.
-6. Bundle-specific accessors (``primary_path``, ``file_paths``,
-   ``file_bytes``, ``file_role``, ``paths_with_role``) return the
-   expected values; unknown roles round-trip without error.
-7. Malformed bundle envelopes (missing primary_path, missing
-   files, missing data, primary_path not in files) raise
-   ``RnpProtocolError`` with ``code=INTERNAL`` — no opaque
-   fallthrough.
-8. End-to-end against the server's ``make_model3d_bundle_envelope``
-   helper (imported from the comfy-rnp-server worktree) — the bytes
-   the server packs into each file are exactly the bytes the
-   client unpacks, and the path sanitization happens server-side so
-   the client never sees a malicious path.
-9. ``CLIENT_CAPABILITIES`` advertises ``model_3d:bundle_inline``
-   so server-side capability-gated descriptors negotiate cleanly.
-
-Run with the ComfyUI venv (needs torch / comfy_api on PYTHONPATH):
-
-    python notes/run_model3d_bundle_decoder.py
-"""
-from __future__ import annotations
-
-import base64
-import importlib.util
-import os
-import shutil
-import sys
-import tempfile
-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.geometry_types.File3D``
-# resolves at first BundledFile3D construction.
-if _COMFYUI_DIR not in sys.path:
-    sys.path.insert(0, _COMFYUI_DIR)
-if _SERVER_DIR not in sys.path:
-    sys.path.insert(0, _SERVER_DIR)
-
-# ComfyUI's full ``comfy_api`` tree pulls torch / PIL / comfy_execution at
-# import time. Stub just what BundledFile3D needs — the leaf
-# ``comfy_api.latest._util.geometry_types.File3D`` class with a faithful
-# subclass-friendly implementation.
-class _StubFile3D:
-    """Faithful subset of upstream ``comfy_api.latest._util.geometry_types.File3D``.
-
-    Implements the same parent-class shape (``_source`` / ``_format``
-    attrs + ``get_source`` / ``get_data`` / ``get_bytes`` / ``save_to``
-    methods) so subclassing it produces an object indistinguishable
-    from the real thing for the decoder's purposes. The smoke test
-    deliberately stubs torch + the heavy ``comfy_api.latest._util``
-    package init to keep the test runnable in the rnp venv.
-    """
-    def __init__(self, source, file_format: str = "") -> None:
-        self._source = source
-        self._format = (file_format or "").lstrip(".").lower()
-
-    @property
-    def format(self) -> str:
-        return self._format
-
-    @format.setter
-    def format(self, value: str) -> None:
-        self._format = (value or "").lstrip(".").lower()
-
-
-# Stub ``comfy_api.latest._util`` and the nested ``geometry_types`` module
-# so ``from comfy_api.latest._util.geometry_types import File3D`` works
-# without pulling torch / PIL / etc.
-_comfy_api = types.ModuleType("comfy_api")
-_comfy_api.__path__ = []  # type: ignore[attr-defined]
-_comfy_api_latest = types.ModuleType("comfy_api.latest")
-_comfy_api_latest.__path__ = []  # type: ignore[attr-defined]
-_comfy_api_latest_util = types.ModuleType("comfy_api.latest._util")
-_comfy_api_latest_util.__path__ = []  # type: ignore[attr-defined]
-_comfy_api_latest_util.File3D = _StubFile3D
-_comfy_api_latest_util_geom = types.ModuleType("comfy_api.latest._util.geometry_types")
-_comfy_api_latest_util_geom.File3D = _StubFile3D
-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)
-sys.modules.setdefault("comfy_api.latest._util.geometry_types", _comfy_api_latest_util_geom)
-
-# Emulate the client package layout for the relative imports.
-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``; the bundle decoder
-# doesn't use it, so a tiny stub suffices.
-sys.modules["comfy_remote_nodes_test.client"] = types.ModuleType(
-    "comfy_remote_nodes_test.client",
-)
-serialization = _load_module(
-    "comfy_remote_nodes_test.serialization",
-    os.path.join(_CLIENT_DIR, "serialization.py"),
-)
-
-
-# ---------------------------------------------------------------------------
-# Build a minimal Wavefront-OBJ + MTL + PNG-stub bundle for the round-trip.
-# These bytes don't need to be loadable by an actual OBJ parser — the
-# decoder is bytes-in / bytes-out — but using realistic content makes
-# debugging easier if a future regression breaks the cross-file refs.
-# ---------------------------------------------------------------------------
-def _make_minimal_obj() -> bytes:
-    return (
-        b"# Minimal OBJ for RNP bundle smoke test.\n"
-        b"mtllib model.mtl\n"
-        b"v 0.0 0.0 0.0\nv 1.0 0.0 0.0\nv 0.0 1.0 0.0\n"
-        b"usemtl test_material\n"
-        b"f 1 2 3\n"
-    )
-
-
-def _make_minimal_mtl() -> bytes:
-    return (
-        b"# Minimal MTL for RNP bundle smoke test.\n"
-        b"newmtl test_material\n"
-        b"Ka 1.0 1.0 1.0\n"
-        b"Kd 1.0 1.0 1.0\n"
-        b"map_Kd textures/albedo.png\n"
-        b"map_bump textures/normal.png\n"
-    )
-
-
-def _make_stub_png(tag: bytes) -> bytes:
-    # Not a real PNG — the decoder doesn't parse it. Tag distinguishes
-    # each "texture" so the round-trip assertions can pin which bytes
-    # went where.
-    return b"\x89PNG\r\n\x1a\n" + tag
-
-
-obj_bytes = _make_minimal_obj()
-mtl_bytes = _make_minimal_mtl()
-albedo_bytes = _make_stub_png(b"ALBEDO_STUB")
-normal_bytes = _make_stub_png(b"NORMAL_STUB")
-roughness_bytes = _make_stub_png(b"ROUGHNESS_STUB")
-
-
-# ---------------------------------------------------------------------------
-# 1. Capability constant intact at the vendored protocol layer.
-# ---------------------------------------------------------------------------
-assert protocol.Capability.MODEL_3D_BUNDLE_INLINE == "model_3d:bundle_inline", (
-    f"Capability.MODEL_3D_BUNDLE_INLINE wrong value: "
-    f"{protocol.Capability.MODEL_3D_BUNDLE_INLINE!r}"
-)
-print("PASS: Capability.MODEL_3D_BUNDLE_INLINE = 'model_3d:bundle_inline'.")
-
-
-# ---------------------------------------------------------------------------
-# Build a hand-crafted bundle envelope for the next several checks.
-# ---------------------------------------------------------------------------
-hand_envelope = {
-    "type":         "model_3d",
-    "encoding":     "bundle_inline",
-    "format":       "obj",
-    "primary_path": "model.obj",
-    "files": [
-        {"path": "model.obj",            "format": "obj", "role": "mesh",
-         "data": base64.b64encode(obj_bytes).decode("ascii"),
-         "byte_size": len(obj_bytes)},
-        {"path": "model.mtl",            "format": "mtl", "role": "material",
-         "data": base64.b64encode(mtl_bytes).decode("ascii")},
-        {"path": "textures/albedo.png",  "format": "png", "role": "texture_diffuse",
-         "data": base64.b64encode(albedo_bytes).decode("ascii")},
-        {"path": "textures/normal.png",  "format": "png", "role": "texture_normal",
-         "data": base64.b64encode(normal_bytes).decode("ascii")},
-        {"path": "textures/roughness.png","format": "png","role": "texture_roughness",
-         "data": base64.b64encode(roughness_bytes).decode("ascii")},
-        # Unknown role — must round-trip without error.
-        {"path": "extras/notes.txt",     "role": "producer_notes",
-         "data": base64.b64encode(b"hello").decode("ascii")},
-    ],
-}
-
-
-# ---------------------------------------------------------------------------
-# 2. is_envelope recognises a bundle envelope.
-# ---------------------------------------------------------------------------
-assert protocol.is_envelope(hand_envelope), (
-    "is_envelope returned False for a well-formed bundle_inline envelope; "
-    "HEAVY_TYPES update somehow regressed."
-)
-print("PASS: is_envelope recognises bundle_inline envelopes.")
-
-
-# ---------------------------------------------------------------------------
-# 3. decode_model3d_envelope dispatches + round-trips primary bytes.
-# ---------------------------------------------------------------------------
-bundle = serialization.decode_model3d_envelope(hand_envelope)
-bundle_cls = serialization._bundled_file3d_class()
-assert isinstance(bundle, bundle_cls), (
-    f"decode_model3d_envelope should return BundledFile3D, got "
-    f"{type(bundle).__name__}"
-)
-# It must ALSO be a File3D (parent class) — that's the whole point of
-# the subclass approach.
-assert isinstance(bundle, _StubFile3D), (
-    f"BundledFile3D must subclass File3D — got {type(bundle).__mro__}"
-)
-# Primary file bytes round-trip via get_data / get_bytes (no disk I/O).
-assert bundle.get_bytes() == obj_bytes, (
-    f"primary OBJ bytes lost: {len(bundle.get_bytes())} vs {len(obj_bytes)}"
-)
-src = bundle.get_data()
-src.seek(0)
-assert src.read() == obj_bytes, "primary OBJ get_data() didn't round-trip"
-assert bundle.format == "obj", f"primary format wrong: {bundle.format!r}"
-print("PASS: decode_model3d_envelope returns BundledFile3D + primary bytes round-trip.")
-
-
-# ---------------------------------------------------------------------------
-# 4. get_source() materialises the whole bundle on disk.
-# ---------------------------------------------------------------------------
-disk_path = bundle.get_source()
-assert os.path.isfile(disk_path), f"get_source() should return existing path: {disk_path}"
-assert disk_path.endswith("model.obj"), f"primary path wrong: {disk_path}"
-with open(disk_path, "rb") as f:
-    assert f.read() == obj_bytes, "disk-materialised primary OBJ bytes mismatched"
-# Sidecars must be at the SAME relative paths so the .obj's mtllib /
-# map_Kd refs resolve.
-temp_root = bundle.materialize_to_temp_dir()
-assert os.path.isfile(os.path.join(temp_root, "model.mtl")), (
-    "model.mtl missing from materialised bundle"
-)
-assert os.path.isfile(os.path.join(temp_root, "textures", "albedo.png")), (
-    "textures/albedo.png missing from materialised bundle"
-)
-assert os.path.isfile(os.path.join(temp_root, "textures", "normal.png"))
-assert os.path.isfile(os.path.join(temp_root, "textures", "roughness.png"))
-assert os.path.isfile(os.path.join(temp_root, "extras", "notes.txt"))
-# Sidecar bytes match.
-with open(os.path.join(temp_root, "model.mtl"), "rb") as f:
-    assert f.read() == mtl_bytes
-with open(os.path.join(temp_root, "textures", "albedo.png"), "rb") as f:
-    assert f.read() == albedo_bytes
-print("PASS: get_source() materialises whole bundle on disk at producer-supplied paths.")
-
-
-# ---------------------------------------------------------------------------
-# 5. save_to(dest) copies the whole bundle into the destination directory.
-# ---------------------------------------------------------------------------
-save_dir = tempfile.mkdtemp(prefix="rnp_save_to_test_")
-try:
-    out_path = bundle.save_to(os.path.join(save_dir, "renamed_scene.obj"))
-    assert os.path.isfile(out_path)
-    with open(out_path, "rb") as f:
-        assert f.read() == obj_bytes, "primary OBJ bytes wrong after save_to"
-    # Sidecars keep ORIGINAL filenames so cross-file refs inside the
-    # OBJ (which point at "model.mtl") still resolve relative to the
-    # renamed primary file's directory.
-    assert os.path.isfile(os.path.join(save_dir, "model.mtl")), (
-        "save_to should copy model.mtl alongside the renamed primary"
-    )
-    assert os.path.isfile(os.path.join(save_dir, "textures", "albedo.png"))
-    assert os.path.isfile(os.path.join(save_dir, "textures", "normal.png"))
-    with open(os.path.join(save_dir, "textures", "normal.png"), "rb") as f:
-        assert f.read() == normal_bytes
-    print("PASS: save_to copies whole bundle, primary renamed, sidecars preserve filenames.")
-finally:
-    shutil.rmtree(save_dir, ignore_errors=True)
-
-
-# ---------------------------------------------------------------------------
-# 6. Bundle-specific accessors.
-# ---------------------------------------------------------------------------
-assert bundle.primary_path == "model.obj"
-assert set(bundle.file_paths) == {
-    "model.obj", "model.mtl",
-    "textures/albedo.png", "textures/normal.png", "textures/roughness.png",
-    "extras/notes.txt",
-}
-assert bundle.file_bytes("textures/albedo.png") == albedo_bytes
-# Optional roles.
-assert bundle.file_role("model.obj") == "mesh"
-assert bundle.file_role("textures/normal.png") == "texture_normal"
-# Unknown roles round-trip — no enum check.
-assert bundle.file_role("extras/notes.txt") == "producer_notes"
-# Missing role (none supplied) returns None.
-assert bundle.file_role("nonexistent.png") is None
-# Convenience lookup.
-assert bundle.paths_with_role("texture_normal") == ["textures/normal.png"]
-assert bundle.paths_with_role("nonexistent_role") == []
-print("PASS: bundle accessors (primary_path / file_paths / file_role / paths_with_role).")
-
-
-# ---------------------------------------------------------------------------
-# 7. Malformed bundle envelopes raise RnpProtocolError with code=INTERNAL.
-# ---------------------------------------------------------------------------
-def _expect_protocol_error(bad_env: dict, needle: str) -> None:
-    try:
-        serialization.decode_model3d_envelope(bad_env)
-    except protocol.RnpProtocolError as e:
-        assert e.code == protocol.ErrorCode.INTERNAL, f"wrong error code: {e.code}"
-        assert needle in str(e), f"error msg doesn't mention {needle!r}: {e}"
-    else:
-        raise AssertionError(f"expected RnpProtocolError for {needle!r}")
-
-_expect_protocol_error(
-    {"type": "model_3d", "encoding": "bundle_inline", "files": [
-        {"path": "x.obj", "data": "AA=="},
-    ]},
-    "primary_path",
-)
-_expect_protocol_error(
-    {"type": "model_3d", "encoding": "bundle_inline", "primary_path": "x.obj"},
-    "files",
-)
-_expect_protocol_error(
-    {"type": "model_3d", "encoding": "bundle_inline", "primary_path": "x.obj",
-     "files": [{"path": "y.obj", "data": "AA=="}]},
-    "primary_path 'x.obj' not in",
-)
-_expect_protocol_error(
-    {"type": "model_3d", "encoding": "bundle_inline", "primary_path": "x.obj",
-     "files": [{"path": "x.obj"}]},  # missing data
-    "missing inline 'data'",
-)
-_expect_protocol_error(
-    {"type": "model_3d", "encoding": "wat_inline", "data": "AA=="},
-    "Unsupported model_3d encoding",
-)
-print("PASS: malformed bundle envelopes raise RnpProtocolError(INTERNAL).")
-
-
-# ---------------------------------------------------------------------------
-# 8. End-to-end: server make_model3d_bundle_envelope -> client decode.
-#    Skips when the server worktree isn't on disk.
-# ---------------------------------------------------------------------------
-server_protocol_path = os.path.join(
-    _SERVER_DIR, "comfy_rnp_protocol", "envelopes.py",
-)
-if os.path.exists(server_protocol_path):
-    server_pkg = types.ModuleType("comfy_rnp_protocol_e2e_bundle")
-    server_pkg.__path__ = [os.path.join(_SERVER_DIR, "comfy_rnp_protocol")]
-    sys.modules["comfy_rnp_protocol_e2e_bundle"] = server_pkg
-    _load_module(
-        "comfy_rnp_protocol_e2e_bundle.constants",
-        os.path.join(_SERVER_DIR, "comfy_rnp_protocol", "constants.py"),
-    )
-    server_envelopes = _load_module(
-        "comfy_rnp_protocol_e2e_bundle.envelopes",
-        os.path.join(_SERVER_DIR, "comfy_rnp_protocol", "envelopes.py"),
-    )
-    assert hasattr(server_envelopes, "make_model3d_bundle_envelope"), (
-        "Server envelopes module missing make_model3d_bundle_envelope — "
-        "server-side PR #38 should have added it."
-    )
-    server_env = server_envelopes.make_model3d_bundle_envelope(
-        files=[
-            {"path": "model.obj",            "format": "obj", "role": "mesh",
-             "data": base64.b64encode(obj_bytes).decode("ascii"),
-             "byte_size": len(obj_bytes)},
-            {"path": "model.mtl",            "format": "mtl", "role": "material",
-             "data": base64.b64encode(mtl_bytes).decode("ascii")},
-            {"path": "textures/albedo.png",  "format": "png", "role": "texture_diffuse",
-             "data": base64.b64encode(albedo_bytes).decode("ascii")},
-        ],
-        primary_path="model.obj",
-        format="obj",
-    )
-    assert server_env["type"] == "model_3d", server_env
-    assert server_env["encoding"] == "bundle_inline", server_env
-    assert server_env["primary_path"] == "model.obj"
-    assert len(server_env["files"]) == 3
-    # Path sanitization happens server-side. Confirm a malicious path
-    # would be rejected at envelope build time (defence-in-depth — the
-    # client decoder doesn't see this).
-    try:
-        server_envelopes.make_model3d_bundle_envelope(
-            files=[{"path": "../etc/passwd", "data": "AA=="}],
-            primary_path="../etc/passwd",
-        )
-    except ValueError as e:
-        assert "parent-dir" in str(e) or "absolute" in str(e), (
-            f"path sanitization mistake: {e!r}"
-        )
-    else:
-        raise AssertionError("server should reject '..' path at envelope build time")
-    # Pipe through the client decoder.
-    server_bundle = serialization.decode_model3d_envelope(server_env)
-    assert isinstance(server_bundle, bundle_cls), (
-        f"server-built envelope decoded to wrong type: {type(server_bundle)}"
-    )
-    assert server_bundle.get_bytes() == obj_bytes
-    assert server_bundle.file_bytes("textures/albedo.png") == albedo_bytes
-    assert server_bundle.file_role("model.obj") == "mesh"
-    print(
-        "PASS: end-to-end server make_model3d_bundle_envelope -> client "
-        "decode_model3d_envelope preserves all files + roles + bytes."
-    )
-else:
-    print(
-        f"SKIP end-to-end: server worktree not found at {_SERVER_DIR!r}"
-    )
-
-
-# ---------------------------------------------------------------------------
-# 9. CLIENT_CAPABILITIES advertises model_3d:bundle_inline.
-#    client.py imports aiohttp / comfy_api_nodes — same source-level
-#    contract check we used for the GLB cap in PR #1.
-# ---------------------------------------------------------------------------
-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_BUNDLE_INLINE" in cc_block, (
-    "Capability.MODEL_3D_BUNDLE_INLINE not advertised in CLIENT_CAPABILITIES"
-)
-print("PASS: CLIENT_CAPABILITIES advertises Capability.MODEL_3D_BUNDLE_INLINE.")
-
-print("ALL CHECKS PASSED.")

notes/run_model3d_decoder.py

@@ -1,316 +0,0 @@
-"""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.")

notes/run_task_handle_decoder.py

@@ -1,392 +0,0 @@
-"""Smoke test for the client-side cross-node task_handle decoder.
-
-Verifies:
-
-1. ``Capability.IO_TASK_HANDLE`` is intact at the vendored protocol
-   layer and exposed as the well-formed ``io:task_handle`` token.
-2. ``"task_handle"`` is in ``HEAVY_TYPES`` and ``is_envelope`` returns
-   True for a well-formed task_handle envelope.
-3. ``decode_task_handle_envelope`` returns a ``TaskHandle`` dataclass
-   with every field round-tripped verbatim (including a non-empty
-   ``parent_chain``).
-4. ``str(handle)`` surfaces the native_id (so UI previews / log lines
-   stay human-meaningful) without leaking the lineage chain.
-5. ``encode_task_handle(decode_task_handle_envelope(env)) == env`` —
-   the round-trip is lossless so a downstream provider node can re-
-   emit the handle unchanged.
-6. The top-level ``decode_envelope`` dispatcher routes task_handle
-   envelopes to the new decoder (so proxy_node's generic
-   ``_deserialize_output`` picks it up without provider-specific
-   plumbing).
-7. Malformed task_handle envelopes (missing fields, wrong types,
-   comma-union ``kind``, non-list ``parent_chain``, bad parent_chain
-   entries, wrong encoding) raise ``RnpProtocolError`` with
-   ``code=INTERNAL`` — no opaque fallthrough.
-8. End-to-end against the server's ``make_task_handle_envelope``
-   helper (imported from the comfy-rnp-server worktree) — the
-   envelope the server packs is exactly what the client decoder
-   unpacks, and ``encode_task_handle`` of the decoded result
-   re-produces the same wire shape.
-9. ``CLIENT_CAPABILITIES`` advertises ``io:task_handle`` so server-
-   side capability-gated descriptors negotiate cleanly.
-
-Run with the ComfyUI venv (needs torch / comfy_api on PYTHONPATH):
-
-    python notes/run_task_handle_decoder.py
-"""
-from __future__ import annotations
-
-import importlib.util
-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
-
-
-# Emulate the client package layout so the relative ``from . import
-# client as rnp_client`` at the top of serialization.py resolves
-# without pulling aiohttp / comfy_api_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``; the task_handle
-# decoder doesn't use it, so a tiny stub suffices.
-sys.modules["comfy_remote_nodes_test.client"] = types.ModuleType(
-    "comfy_remote_nodes_test.client",
-)
-serialization = _load_module(
-    "comfy_remote_nodes_test.serialization",
-    os.path.join(_CLIENT_DIR, "serialization.py"),
-)
-
-
-# ---------------------------------------------------------------------------
-# 1. Capability constant intact + well-formed token.
-# ---------------------------------------------------------------------------
-assert protocol.Capability.IO_TASK_HANDLE == "io:task_handle", (
-    f"Capability.IO_TASK_HANDLE wrong value: "
-    f"{protocol.Capability.IO_TASK_HANDLE!r}"
-)
-print("PASS: Capability.IO_TASK_HANDLE = 'io:task_handle'.")
-
-
-# ---------------------------------------------------------------------------
-# 2. HEAVY_TYPES + is_envelope.
-# ---------------------------------------------------------------------------
-assert "task_handle" in protocol.HEAVY_TYPES, (
-    f"'task_handle' missing from HEAVY_TYPES: {sorted(protocol.HEAVY_TYPES)}"
-)
-
-hand_envelope = {
-    "type":           "task_handle",
-    "encoding":       "vendor_inline",
-    "vendor":         "tripo",
-    "kind":           "MODEL_TASK_ID",
-    "native_id":      "abc123-tripo-uuid",
-    "origin_node_id": "42",
-    "parent_chain": [
-        {"vendor": "tripo", "kind": "MODEL_TASK_ID",
-         "native_id": "root-uuid", "origin_node_id": "9"},
-        {"vendor": "tripo", "kind": "MODEL_TASK_ID",
-         "native_id": "prev-uuid", "origin_node_id": "17"},
-    ],
-}
-
-assert protocol.is_envelope(hand_envelope), (
-    "is_envelope returned False for a well-formed task_handle envelope; "
-    "HEAVY_TYPES update somehow regressed."
-)
-print("PASS: 'task_handle' in HEAVY_TYPES and is_envelope recognises it.")
-
-
-# ---------------------------------------------------------------------------
-# 3. decode_task_handle_envelope returns a populated TaskHandle.
-# ---------------------------------------------------------------------------
-handle = serialization.decode_task_handle_envelope(hand_envelope)
-assert isinstance(handle, serialization.TaskHandle), (
-    f"decode_task_handle_envelope should return TaskHandle, got "
-    f"{type(handle).__name__}"
-)
-assert handle.vendor == "tripo"
-assert handle.kind == "MODEL_TASK_ID"
-assert handle.native_id == "abc123-tripo-uuid"
-assert handle.origin_node_id == "42"
-assert len(handle.parent_chain) == 2
-assert handle.parent_chain[0] == {
-    "vendor": "tripo", "kind": "MODEL_TASK_ID",
-    "native_id": "root-uuid", "origin_node_id": "9",
-}
-assert handle.parent_chain[1] == {
-    "vendor": "tripo", "kind": "MODEL_TASK_ID",
-    "native_id": "prev-uuid", "origin_node_id": "17",
-}
-print("PASS: decode_task_handle_envelope returns TaskHandle with full lineage.")
-
-
-# ---------------------------------------------------------------------------
-# 4. str(handle) surfaces the native_id.
-# ---------------------------------------------------------------------------
-assert str(handle) == "abc123-tripo-uuid", (
-    f"str(TaskHandle) should be the native_id; got {str(handle)!r}"
-)
-print("PASS: str(TaskHandle) surfaces the native_id.")
-
-
-# ---------------------------------------------------------------------------
-# 5. encode(decode(env)) == env — lossless round-trip.
-# ---------------------------------------------------------------------------
-re_encoded = serialization.encode_task_handle(handle)
-assert re_encoded == hand_envelope, (
-    f"round-trip mismatch:\n  original: {hand_envelope}\n  "
-    f"re-encoded: {re_encoded}"
-)
-# Empty chain default is preserved.
-solo = serialization.decode_task_handle_envelope({
-    "type":           "task_handle",
-    "encoding":       "vendor_inline",
-    "vendor":         "kling",
-    "kind":           "KLING_VIDEO_ID",
-    "native_id":      "video-xyz",
-    "origin_node_id": "5",
-    "parent_chain":   [],
-})
-assert solo.parent_chain == [], "empty parent_chain should round-trip empty"
-assert serialization.encode_task_handle(solo)["parent_chain"] == []
-print("PASS: encode_task_handle round-trips lossless (including empty chain).")
-
-
-# ---------------------------------------------------------------------------
-# 6. decode_envelope dispatcher routes to the new decoder.
-# ---------------------------------------------------------------------------
-dispatched = serialization.decode_envelope(hand_envelope)
-assert isinstance(dispatched, serialization.TaskHandle), (
-    f"decode_envelope didn't dispatch task_handle to the right decoder; "
-    f"got {type(dispatched).__name__}"
-)
-assert dispatched.native_id == "abc123-tripo-uuid"
-print("PASS: decode_envelope dispatches 'task_handle' -> decode_task_handle_envelope.")
-
-
-# ---------------------------------------------------------------------------
-# 7. Malformed envelopes raise RnpProtocolError(INTERNAL).
-# ---------------------------------------------------------------------------
-def _expect_protocol_error(bad_env: dict, needle: str) -> None:
-    try:
-        serialization.decode_task_handle_envelope(bad_env)
-    except protocol.RnpProtocolError as e:
-        assert e.code == protocol.ErrorCode.INTERNAL, f"wrong error code: {e.code}"
-        assert needle in str(e), f"error msg doesn't mention {needle!r}: {e}"
-    else:
-        raise AssertionError(f"expected RnpProtocolError for {needle!r}")
-
-
-# Wrong encoding.
-_expect_protocol_error(
-    {"type": "task_handle", "encoding": "replay_ticket_inline",
-     "vendor": "tripo", "kind": "MODEL_TASK_ID",
-     "native_id": "x", "origin_node_id": "1"},
-    "Unsupported task_handle encoding",
-)
-# Missing top-level fields.
-for missing in ("vendor", "kind", "native_id", "origin_node_id"):
-    bad = {
-        "type": "task_handle", "encoding": "vendor_inline",
-        "vendor": "tripo", "kind": "MODEL_TASK_ID",
-        "native_id": "x", "origin_node_id": "1",
-    }
-    del bad[missing]
-    _expect_protocol_error(bad, repr(missing))
-# Empty string fields.
-_expect_protocol_error(
-    {"type": "task_handle", "encoding": "vendor_inline",
-     "vendor": "", "kind": "MODEL_TASK_ID",
-     "native_id": "x", "origin_node_id": "1"},
-    "'vendor'",
-)
-# Comma-union kind rejected (input-acceptance string, not emit kind).
-_expect_protocol_error(
-    {"type": "task_handle", "encoding": "vendor_inline",
-     "vendor": "tripo",
-     "kind": "MODEL_TASK_ID,RIG_TASK_ID,RETARGET_TASK_ID",
-     "native_id": "x", "origin_node_id": "1"},
-    "singular socket kind",
-)
-# parent_chain not a list.
-_expect_protocol_error(
-    {"type": "task_handle", "encoding": "vendor_inline",
-     "vendor": "tripo", "kind": "MODEL_TASK_ID",
-     "native_id": "x", "origin_node_id": "1",
-     "parent_chain": "not-a-list"},
-    "'parent_chain'",
-)
-# Bad parent_chain entry — missing field.
-_expect_protocol_error(
-    {"type": "task_handle", "encoding": "vendor_inline",
-     "vendor": "tripo", "kind": "MODEL_TASK_ID",
-     "native_id": "x", "origin_node_id": "1",
-     "parent_chain": [{"vendor": "tripo", "kind": "MODEL_TASK_ID",
-                       "native_id": "y"}]},  # missing origin_node_id
-    "parent_chain[0]",
-)
-# Bad parent_chain entry — comma-union kind in lineage.
-_expect_protocol_error(
-    {"type": "task_handle", "encoding": "vendor_inline",
-     "vendor": "tripo", "kind": "MODEL_TASK_ID",
-     "native_id": "x", "origin_node_id": "1",
-     "parent_chain": [{"vendor": "tripo",
-                       "kind": "MODEL_TASK_ID,RIG_TASK_ID",
-                       "native_id": "y", "origin_node_id": "3"}]},
-    "parent_chain[0]",
-)
-# Non-dict parent_chain entry.
-_expect_protocol_error(
-    {"type": "task_handle", "encoding": "vendor_inline",
-     "vendor": "tripo", "kind": "MODEL_TASK_ID",
-     "native_id": "x", "origin_node_id": "1",
-     "parent_chain": ["not-a-dict"]},
-    "parent_chain[0]",
-)
-print("PASS: malformed task_handle envelopes raise RnpProtocolError(INTERNAL).")
-
-
-# ---------------------------------------------------------------------------
-# 8. End-to-end: server make_task_handle_envelope -> client decode.
-#    Skips when the server worktree isn't on disk.
-# ---------------------------------------------------------------------------
-server_envelopes_path = os.path.join(
-    _SERVER_DIR, "comfy_rnp_protocol", "envelopes.py",
-)
-if os.path.exists(server_envelopes_path):
-    server_pkg = types.ModuleType("comfy_rnp_protocol_e2e_handle")
-    server_pkg.__path__ = [os.path.join(_SERVER_DIR, "comfy_rnp_protocol")]
-    sys.modules["comfy_rnp_protocol_e2e_handle"] = server_pkg
-    _load_module(
-        "comfy_rnp_protocol_e2e_handle.constants",
-        os.path.join(_SERVER_DIR, "comfy_rnp_protocol", "constants.py"),
-    )
-    server_envelopes = _load_module(
-        "comfy_rnp_protocol_e2e_handle.envelopes",
-        os.path.join(_SERVER_DIR, "comfy_rnp_protocol", "envelopes.py"),
-    )
-    assert hasattr(server_envelopes, "make_task_handle_envelope"), (
-        "Server envelopes module missing make_task_handle_envelope — "
-        "server-side PR #39 should have added it."
-    )
-    server_env = server_envelopes.make_task_handle_envelope(
-        vendor="tripo",
-        kind="MODEL_TASK_ID",
-        native_id="server-built-uuid",
-        origin_node_id="100",
-        parent_chain=[
-            {"vendor": "tripo", "kind": "MODEL_TASK_ID",
-             "native_id": "ancestor-uuid", "origin_node_id": "50"},
-        ],
-    )
-    assert server_env["type"] == "task_handle", server_env
-    assert server_env["encoding"] == "vendor_inline", server_env
-    # Client decode of the server-packed bytes.
-    server_decoded = serialization.decode_task_handle_envelope(server_env)
-    assert isinstance(server_decoded, serialization.TaskHandle)
-    assert server_decoded.native_id == "server-built-uuid"
-    assert server_decoded.origin_node_id == "100"
-    assert len(server_decoded.parent_chain) == 1
-    assert server_decoded.parent_chain[0]["native_id"] == "ancestor-uuid"
-    # Round-trip back through encode_task_handle reproduces the server's wire shape.
-    re_encoded_server = serialization.encode_task_handle(server_decoded)
-    assert re_encoded_server == server_env, (
-        f"client encode -> server wire mismatch:\n"
-        f"  server: {server_env}\n  client: {re_encoded_server}"
-    )
-    # Server-side rejection paths fire as expected (defence in depth —
-    # the client decoder mirrors the same rejections, but a malicious /
-    # buggy producer can't sneak a bad envelope past the server-side
-    # builder either).
-    try:
-        server_envelopes.make_task_handle_envelope(
-            vendor="tripo",
-            kind="MODEL_TASK_ID,RIG_TASK_ID",  # comma-union rejected
-            native_id="x",
-            origin_node_id="1",
-        )
-    except ValueError as e:
-        assert "singular socket kind" in str(e), (
-            f"server-side comma-union rejection wrong msg: {e!r}"
-        )
-    else:
-        raise AssertionError(
-            "server should reject comma-union 'kind' at envelope build time"
-        )
-    # Server enforces parent_chain depth cap (32).
-    try:
-        server_envelopes.make_task_handle_envelope(
-            vendor="tripo",
-            kind="MODEL_TASK_ID",
-            native_id="x",
-            origin_node_id="1",
-            parent_chain=[
-                {"vendor": "tripo", "kind": "MODEL_TASK_ID",
-                 "native_id": f"u{i}", "origin_node_id": str(i)}
-                for i in range(33)
-            ],
-        )
-    except ValueError as e:
-        assert "depth" in str(e), (
-            f"server-side depth-cap rejection wrong msg: {e!r}"
-        )
-    else:
-        raise AssertionError(
-            "server should reject parent_chain over MAX_CHAIN_DEPTH"
-        )
-    print(
-        "PASS: end-to-end server make_task_handle_envelope -> client "
-        "decode_task_handle_envelope -> encode_task_handle round-trips "
-        "losslessly, and server-side rejections fire."
-    )
-else:
-    print(
-        f"SKIP end-to-end: server worktree not found at {_SERVER_DIR!r}"
-    )
-
-
-# ---------------------------------------------------------------------------
-# 9. CLIENT_CAPABILITIES advertises io:task_handle.
-#    client.py imports aiohttp / comfy_api_nodes — same source-level
-#    contract check we used for the GLB cap in PR #1.
-# ---------------------------------------------------------------------------
-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.IO_TASK_HANDLE" in cc_block, (
-    "Capability.IO_TASK_HANDLE not advertised in CLIENT_CAPABILITIES"
-)
-print("PASS: CLIENT_CAPABILITIES advertises Capability.IO_TASK_HANDLE.")
-
-print("ALL CHECKS PASSED.")

protocol.py

@@ -104,182 +104,19 @@ 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"
-    # 3D MODEL BUNDLE envelope: a primary mesh file plus N companion
-    # files (textures / .mtl / .bin / other PBR maps) carried inline as
-    # a single coherent unit. Wire shape:
-    #
-    #   {
-    #     "type":         "model_3d",
-    #     "encoding":     "bundle_inline",
-    #     "format":       "obj",
-    #     "primary_path": "model.obj",
-    #     "files": [
-    #       {"path": "model.obj",            "format": "obj", "role": "mesh",
-    #        "data": "<b64>", "byte_size": 12345},
-    #       {"path": "model.mtl",            "format": "mtl", "role": "material",
-    #        "data": "<b64>"},
-    #       {"path": "textures/albedo.png",  "format": "png", "role": "texture_diffuse",
-    #        "data": "<b64>"}
-    #     ]
-    #   }
-    #
-    # ``path`` is the authoritative cross-file reference key — a POSIX-
-    # relative path (no ``..``, no absolute prefix, no Windows drive
-    # letter, no backslashes, no case-insensitive collisions). The
-    # client decoder materialises files into a temp directory using
-    # exactly these paths so a ``.obj`` that references its ``.mtl``
-    # via filename (or a ``.gltf`` that references its ``.bin`` /
-    # textures via relative URI) resolves on disk.
-    #
-    # ``role`` is an OPTIONAL producer hint (NOT a closed enum) —
-    # well-known values include ``mesh``, ``material``, ``buffer``,
-    # ``texture_diffuse``, ``texture_metallic``, ``texture_normal``,
-    # ``texture_roughness``, ``texture_ao``, ``texture_emissive``,
-    # ``texture_height``. Unknown roles round-trip without error;
-    # consumers may use ``role`` for convenience but ``path`` is the
-    # primary identifier.
-    #
-    # ``primary_path`` MUST appear in ``files`` exactly once and pins
-    # the entry-point file (the one passed to ComfyUI's
-    # ``Types.File3D`` constructor by the client decoder). The
-    # top-level ``format`` (if set) MUST match that primary file's
-    # ``format``.
-    #
-    # The client decoder returns a ``BundledFile3D`` (a thin
-    # ``File3D``-compatible wrapper) — existing downstream loaders
-    # that call ``get_source()`` / ``get_data()`` / ``save_to()`` get
-    # the bundle's primary file with all sidecars materialised on
-    # disk first so cross-file refs resolve correctly. Override of
-    # ``save_to()`` copies the *whole bundle* into the destination
-    # directory, not just the primary file.
-    #
-    # Negotiation: the server emits this encoding ONLY when the
-    # inbound request advertised this token; legacy clients without
-    # the bundle decoder will surface NEGOTIATION_FAILED at
-    # descriptor-load time. Whole-descriptor gating — if a node has
-    # both bundle and non-bundle outputs and the cap is absent, the
-    # whole descriptor is skipped rather than synthesising a
-    # partial-socket variant.
-    #
-    # A future sibling ``Capability.MODEL_3D_BUNDLE_URI`` would carry
-    # ``uri`` per file for URL-fetch-out-of-band semantics; not in
-    # this PR.
-    MODEL_3D_BUNDLE_INLINE = "model_3d:bundle_inline"
-    # Cross-node task handle envelope: wraps a vendor-native task /
-    # video / job ID (Tripo MODEL_TASK_ID, Kling video_id, future
-    # partner handles) with enough lineage metadata to chain provider
-    # nodes (Tripo Texture/Refine/Rig/Retarget/Conversion, Kling
-    # VideoExtend, ...) across a single workflow graph. Wire shape:
-    #
-    #   {
-    #     "type":           "task_handle",
-    #     "encoding":       "vendor_inline",
-    #     "vendor":         "tripo",
-    #     "kind":           "MODEL_TASK_ID",
-    #     "native_id":      "abc123-tripo-uuid",
-    #     "origin_node_id": "42",
-    #     "parent_chain": [
-    #       {"vendor": "tripo", "kind": "MODEL_TASK_ID",
-    #        "native_id": "prev-uuid", "origin_node_id": "17"}
-    #     ]
-    #   }
-    #
-    # ``vendor`` is a lowercase provider slug. ``kind`` is the
-    # singular emitted socket kind string (``MODEL_TASK_ID`` /
-    # ``RIG_TASK_ID`` / ``RETARGET_TASK_ID`` / ``KLING_VIDEO_ID``) —
-    # NEVER the comma-union string that provider nodes use for input
-    # type-acceptance (e.g. ``TripoConversionNode`` accepts
-    # ``"MODEL_TASK_ID,RIG_TASK_ID,RETARGET_TASK_ID"``; the wire kind
-    # is whatever the upstream actually emitted). ``native_id`` is
-    # the vendor-native opaque string. ``origin_node_id`` is the
-    # upstream workflow node's unique id (log correlation + future
-    # replay).
-    #
-    # ``parent_chain`` is a FLAT list (not recursive nested
-    # envelopes) of ``{vendor, kind, native_id, origin_node_id}``
-    # refs in root-to-immediate-parent order. Each downstream node
-    # appends its parent's ref before the parent's chain is copied
-    # through; the chain captures the full lineage. Hard-capped at
-    # depth 32 server-side to prevent runaway growth.
-    #
-    # Negotiation: server emits ``task_handle`` envelopes only when
-    # the inbound request advertises this capability; legacy clients
-    # surface NEGOTIATION_FAILED at descriptor-load time on any
-    # provider that publishes a handle-typed output OR accepts one
-    # as input. Whole-descriptor gating — partial-socket synthesis
-    # is rejected because the chain breaks meaningfully if either
-    # end is missing.
-    #
-    # The client decoder returns a ``TaskHandle`` dataclass (NOT a
-    # plain string) so lineage metadata round-trips when the value
-    # is passed through to the next provider node — re-encoding the
-    # same dataclass on a downstream submit preserves ``parent_chain``
-    # without the proxy_node needing to know anything provider-
-    # specific.
-    IO_TASK_HANDLE = "io:task_handle"
 
 
-HEAVY_TYPES = frozenset({"image", "video", "audio", "mask", "model_3d", "task_handle"})
+HEAVY_TYPES = frozenset({"image", "video", "audio", "mask", "svg"})
 
 
 def is_envelope(value: Any) -> bool:

proxy_node.py

@@ -3,17 +3,11 @@
 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 plus the two V3
-dynamic input types DYNAMIC_COMBO (branch-selector dropdown with
-per-branch sub-widgets) and AUTOGROW (variable-arity slot template);
-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. Malformed
-DYNAMIC_COMBO / AUTOGROW specs skip the descriptor (return ``None``)
-rather than degrading to a single opaque socket.
+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.
 
 Hidden inputs (``auth_token_comfy_org`` / ``api_key_comfy_org`` /
 ``unique_id`` / ``prompt`` / ``extra_pnginfo`` / ``dynprompt``) are
@@ -51,6 +45,7 @@ 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,
@@ -313,32 +308,6 @@ 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")
@@ -355,6 +324,16 @@ 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"
@@ -413,14 +392,13 @@ 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
@@ -446,111 +424,6 @@ _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)``.
 
@@ -580,90 +453,38 @@ def _parse_inputs(descriptor: dict[str, Any]) -> tuple[list[Any] | None, list[st
 
 
 def _parse_input_spec(name: str, spec: list[Any], optional: bool) -> Any | None:
-    """Map one V3-shaped ``[io_type, options_dict]`` entry to a V3 IO Input.
-
-    This is the top-level entry used by ``_parse_inputs`` (where the
-    name comes from the descriptor's ``input.required[name]`` /
-    ``input.optional[name]`` key) and recursively as the AUTOGROW
-    template entry (which also lacks a leading name field — the slot
-    name is provided by the surrounding AUTOGROW). Branch sub-inputs
-    inside a DYNAMIC_COMBO option use the 3-tuple ``[name, io_type,
-    options]`` form instead — see :func:`_parse_named_input_spec`.
-    """
+    """Map one V3-shaped ``[io_type, options_dict]`` entry to a V3 IO Input."""
     if not isinstance(spec, (list, tuple)) or len(spec) < 1:
         return None
     io_type = spec[0]
     options: dict[str, Any] = spec[1] if len(spec) > 1 and isinstance(spec[1], dict) else {}
-    return _build_input(name, io_type, options, optional)
-
-
-def _parse_named_input_spec(spec: list[Any], optional: bool) -> Any | None:
-    """Map one V3-shaped ``[name, io_type, options_dict]`` entry to a V3 IO Input.
-
-    This is the shape used inside ``DYNAMIC_COMBO`` options' ``inputs``
-    list (each branch sub-input carries its own name as ``spec[0]``).
-    Top-level descriptor inputs use the 2-tuple form handled by
-    :func:`_parse_input_spec`.
-    """
-    if (
-        not isinstance(spec, (list, tuple))
-        or len(spec) < 2
-        or not isinstance(spec[0], str)
-    ):
-        return None
-    name = spec[0]
-    io_type = spec[1]
-    options: dict[str, Any] = spec[2] if len(spec) > 2 and isinstance(spec[2], dict) else {}
-    return _build_input(name, io_type, options, optional)
-
-
-def _build_input(name: str, io_type: Any, options: dict[str, Any], optional: bool) -> Any | None:
-    """Shared dispatch: construct one V3 IO Input from a parsed
-    ``(name, io_type, options)`` triple.
-
-    Recurses for the two V3 dynamic input types:
-
-    * ``DYNAMIC_COMBO`` — selector dropdown whose per-branch sub-inputs
-      live in ``options.options[i].inputs`` (each entry is the 3-tuple
-      ``[name, io_type, options]`` form).
-    * ``AUTOGROW`` — variable-arity slot template. The RNP wire dialect
-      flattens upstream's ``Autogrow.Input.as_dict`` shape: ``template``
-      is a 2-tuple ``[io_type, options]`` entry directly, with
-      ``prefix`` / ``min`` / ``max`` (or ``names`` / ``min``) as siblings
-      of ``template`` rather than nested under ``template.template``.
-
-    Malformed dynamic specs return ``None`` (caller logs + skips the
-    whole descriptor) rather than falling through to the opaque
-    ``IO.Custom`` bucket — that would mask a protocol/schema bug as
-    "looks fine but renders as opaque socket".
-    """
     common = {
         "tooltip": options.get("tooltip"),
         "optional": optional,
     }
-    # ``advanced`` is only accepted by WidgetInput subclasses — the
-    # dynamic constructors (DynamicCombo.Input / Autogrow.Input) do
-    # not take it. Build a separate kwargs bag for primitives.
-    primitive_common = dict(common)
     if options.get("advanced") is not None:
-        primitive_common["advanced"] = bool(options.get("advanced"))
+        common["advanced"] = bool(options.get("advanced"))
 
     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"), **primitive_common)
+                              default=options.get("default"),
+                              control_after_generate=options.get("control_after_generate"),
+                              **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"), **primitive_common)
+                              default=options.get("default"),
+                              control_after_generate=options.get("control_after_generate"),
+                              **common)
     if io_type == "STRING":
         return IO.String.Input(
             name,
             default=options.get("default", ""),
             multiline=bool(options.get("multiline", False)),
-            **primitive_common,
+            **common,
         )
     if io_type == "INT":
         return IO.Int.Input(
@@ -672,7 +493,8 @@ def _build_input(name: str, io_type: Any, options: dict[str, Any], optional: boo
             min=options.get("min", 0),
             max=options.get("max", 2147483647),
             step=options.get("step", 1),
-            **primitive_common,
+            control_after_generate=options.get("control_after_generate"),
+            **common,
         )
     if io_type == "FLOAT":
         return IO.Float.Input(
@@ -681,134 +503,144 @@ def _build_input(name: str, io_type: Any, options: dict[str, Any], optional: boo
             min=options.get("min", 0.0),
             max=options.get("max", 1.0),
             step=options.get("step", 0.01),
-            **primitive_common,
+            **common,
         )
     if io_type == "BOOLEAN":
         return IO.Boolean.Input(
-            name, default=bool(options.get("default", False)), **primitive_common,
+            name, default=bool(options.get("default", False)), **common,
         )
     if io_type == "IMAGE":
-        return IO.Image.Input(name, **primitive_common)
+        return IO.Image.Input(name, **common)
     if io_type == "VIDEO":
-        return IO.Video.Input(name, **primitive_common)
+        return IO.Video.Input(name, **common)
     if io_type == "AUDIO":
-        return IO.Audio.Input(name, **primitive_common)
+        return IO.Audio.Input(name, **common)
     if io_type == "MASK":
-        return IO.Mask.Input(name, **primitive_common)
-    if io_type == "DYNAMIC_COMBO":
-        # Wire shape: ``["DYNAMIC_COMBO", {"options": [{"key": str,
-        # "inputs": [[name, io_type, opts], ...]}, ...], "tooltip": ...}]``.
-        # Empty branch input lists are valid (e.g. Bria moderation's
-        # "false" branch with ``"inputs": []``); the branch dropdown
-        # still needs to render. Malformed shape -> None (no opaque
-        # fallthrough — would mask a schema bug).
-        opts = options.get("options")
-        if not isinstance(opts, (list, tuple)):
-            return None
-        parsed_options: list[Any] = []
-        for opt in opts:
-            if not isinstance(opt, dict):
-                return None
-            key = opt.get("key")
-            if not isinstance(key, str):
-                return None
-            sub_specs = opt.get("inputs") or []
-            if not isinstance(sub_specs, (list, tuple)):
-                return None
-            sub_inputs: list[Any] = []
-            for sub_spec in sub_specs:
-                # Branch sub-inputs are always treated as required at
-                # schema-construction time — the wire dialect doesn't
-                # express per-branch optionality. Server-side runtime
-                # validation enforces presence via ``__branches__``.
-                sub = _parse_named_input_spec(sub_spec, optional=False)
-                if sub is None:
-                    return None
-                sub_inputs.append(sub)
-            parsed_options.append(IO.DynamicCombo.Option(key=key, inputs=sub_inputs))
-        return IO.DynamicCombo.Input(
-            name,
-            options=parsed_options,
-            tooltip=options.get("tooltip"),
-            optional=optional,
-        )
+        return IO.Mask.Input(name, **common)
+    if io_type == "SVG":
+        return IO.SVG.Input(name, **common)
     if io_type == "AUTOGROW":
-        # Wire shape (RNP-flattened):
-        # ``["AUTOGROW", {"template": [io_type, options],
-        #                  "prefix": str, "min": int, "max": int,
-        #                  "tooltip": ...}]``
-        # or with ``names`` instead of ``prefix`` / ``max`` for the
-        # TemplateNames variant.
-        tmpl_spec = options.get("template")
-        if not isinstance(tmpl_spec, (list, tuple)) or len(tmpl_spec) < 1:
-            return None
-        # Reuse top-level parser for the template — the slot name is
-        # cosmetic since TemplatePrefix / TemplateNames rename each
-        # cached copy per-slot.
-        template_input = _parse_input_spec(name, tmpl_spec, optional=False)
-        if template_input is None:
-            return None
-        # Upstream Autogrow asserts the template input is NOT itself a
-        # DynamicInput (DynamicCombo.Input / Autogrow.Input). Mirror
-        # that check at parse time so a malformed descriptor surfaces
-        # as "skipped" rather than blowing up later in the IO machinery.
-        if isinstance(
-            template_input,
-            (IO.DynamicCombo.Input, IO.Autogrow.Input),
+        # 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
-        if "names" in options:
-            names = options.get("names")
-            if not isinstance(names, list) or not all(isinstance(n, str) for n in names):
-                return None
-            try:
-                template = IO.Autogrow.TemplateNames(
-                    template_input,
-                    names=names,
-                    min=int(options.get("min", 1)),
-                )
-            except AssertionError:
-                return None
-        elif "prefix" in options:
-            prefix = options.get("prefix")
-            if not isinstance(prefix, str):
-                return None
-            try:
-                template = IO.Autogrow.TemplatePrefix(
-                    template_input,
-                    prefix=prefix,
-                    min=int(options.get("min", 1)),
-                    max=int(options.get("max", 10)),
-                )
-            except AssertionError:
-                return None
-        else:
+        # 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,
-            tooltip=options.get("tooltip"),
+            display_name=options.get("display_name"),
             optional=optional,
+            tooltip=options.get("tooltip"),
         )
-    # 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.
-    #
-    # NOTE: ``DYNAMIC_COMBO`` and ``AUTOGROW`` are handled above and
-    # never reach this fallthrough — malformed dynamic specs return
-    # ``None`` instead, so the caller logs + skips the descriptor
-    # rather than silently degrading to a single opaque socket.
+    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.
     if isinstance(io_type, str) and io_type:
         return IO.Custom(io_type).Input(name, **common)
     return None
@@ -895,47 +727,32 @@ 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:
-            # 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:
+            # 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:
                 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,
-    "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,
+    "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,
 }
 
 
@@ -1177,21 +994,28 @@ 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; everything else rides in the request ``context``.
+    # 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)
     context: dict[str, Any] = {}
     auth_headers: dict[str, str] = {}
     hidden_holder = getattr(cls, "hidden", None)
@@ -1212,24 +1036,17 @@ async def _execute_remote(
             auth_headers["Authorization"] = f"Bearer {value}"
         elif hname == "api_key_comfy_org":
             auth_headers["X-API-KEY"] = str(value)
-        else:
+        elif hname in forward_set:
             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)
@@ -1404,54 +1221,8 @@ async def _deserialize_output(
         # advisory here, the value already matches the native ComfyUI
         # type for STRING / INT / FLOAT / BOOLEAN.
         return value
-    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:
+    if "data" not in value and "uri" in value:
+        env_type = value.get("type", "")
         policy = url_fetch_policy.get(env_type)
         if not isinstance(policy, dict):
             raise RnpProtocolError(
@@ -1533,9 +1304,6 @@ 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.
@@ -1551,29 +1319,10 @@ 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():
-        _enforce_local_validate(
-            name, value, validate_map.get(name) or {}, node_id=node_id,
-        )
-        encoded = _encode_one(name, value, serialization_map.get(name))
+        encoded = _encode_one(name, value)
         # Only envelope-shaped values can be externalized; scalars pass
         # straight through ``maybe_externalize`` unchanged.
         if serialization.is_envelope(encoded):
@@ -1587,129 +1336,52 @@ async def _encode_inputs(
     return out
 
 
-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.
+def _dict_contains_tensor(value: Any) -> bool:
+    """True iff a dict (or nested dict) holds at least one torch tensor.
 
-    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.
+    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.
     """
-    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,
-                        )
+    if isinstance(value, dict):
+        return any(_dict_contains_tensor(v) for v in value.values())
+    return serialization._is_torch_tensor(value)
 
 
-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:
+def _encode_one(name: str, value: Any) -> 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.
-        # 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,
-        )
+        # 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)
     if rank in (2, 3):
         return serialization.encode_mask_tensor(value)
     log.warning(

serialization.py

@@ -17,7 +17,6 @@ tensor — no parallel dialect.
 """
 from __future__ import annotations
 
-import asyncio
 import base64
 import logging
 from io import BytesIO
@@ -57,64 +56,14 @@ async def maybe_externalize(
     max_inline_bytes: int | None,
     auth_headers: dict[str, str] | None = None,
 ) -> dict[str, Any]:
-    """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.
+    """If the envelope's inline data exceeds ``max_inline_bytes``,
+    upload the bytes and replace ``data`` with ``uri``.
 
     No-op when either argument is falsy, the envelope already has a
-    ``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.
+    ``uri``, or the inline payload fits under the cap.
     """
     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
@@ -135,122 +84,22 @@ 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, *, accepts_batch: bool = False) -> dict[str, Any]:
-    """Encode a ComfyUI IMAGE tensor (B,H,W,C float32 in 0..1) as an
-    image envelope.
+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.
 
-    ``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.
+    Multi-image batches send only the first frame for now; the
+    framework hasn't grown batch-of-envelopes semantics yet.
     """
     import numpy as np
     from PIL import Image
 
-    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:
+    # Strip the batch dim if present.
+    if hasattr(tensor, "dim") and tensor.dim() == 4:
         frame = tensor[0]
     else:
         frame = tensor
@@ -272,18 +121,8 @@ def encode_image_tensor(tensor: Any, *, accepts_batch: bool = False) -> dict[str
 
 
 def decode_image_envelope(envelope: dict[str, Any]) -> Any:
-    """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.
-    """
+    """Decode an image envelope into a ComfyUI IMAGE tensor (B,H,W,C float32)."""
     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}",
@@ -302,69 +141,6 @@ 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
 # ---------------------------------------------------------------------------
@@ -451,11 +227,7 @@ 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",
-        "opus_base64",
-    ):
+    if encoding not in ("mp3_base64", "mp3_inline", "wav_base64", "wav_inline"):
         raise RnpProtocolError(
             f"Unsupported audio encoding: {encoding!r}",
             code=ErrorCode.INTERNAL,
@@ -465,9 +237,47 @@ def decode_audio_envelope(envelope: dict[str, Any]) -> dict[str, Any]:
 
 
 # ---------------------------------------------------------------------------
-# Video decode (encode lands when a remote node accepts VIDEO inputs)
+# Video encode / decode
 # ---------------------------------------------------------------------------
 
+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")
@@ -480,467 +290,83 @@ def decode_video_envelope(envelope: dict[str, Any]) -> Any:
     return InputImpl.VideoFromFile(BytesIO(decode_envelope_data(envelope)))
 
 
-# ---------------------------------------------------------------------------
-# 3D model decode (encode lands when a remote node accepts MODEL_3D inputs)
-# ---------------------------------------------------------------------------
+def is_video_input(value: Any) -> bool:
+    """Best-effort check for a ComfyUI Video object (``Input.Video``).
 
-# ---------------------------------------------------------------------------
-# Multi-file 3D bundle: a ``File3D``-compatible class that carries the
-# primary mesh plus N companion files (textures / .mtl / .bin / etc.).
-#
-# Built lazily as a ``File3D`` subclass on first use — importing
-# ``File3D`` at module load pulls torch (via ``comfy_api.latest._util.
-# geometry_types``), which is fine inside ComfyUI but breaks the
-# smoke-test pattern that exercises serialization without the heavy
-# comfy_api tree. The factory below caches the constructed class so
-# every ``BundledFile3D`` instance shares the same dynamic class and
-# ``isinstance(x, File3D)`` always succeeds.
-# ---------------------------------------------------------------------------
-
-_BUNDLED_FILE3D_CLS: Any = None
-
-
-def _bundled_file3d_class() -> type:
-    """Return the cached ``BundledFile3D`` class (a ``File3D`` subclass).
-
-    Constructs it on first call so the ``comfy_api.latest._util``
-    import (which pulls torch) is paid for only when a bundle
-    envelope is actually decoded — keeps the smoke-test stubbing
-    pattern intact for callers that don't exercise this path.
+    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.
     """
-    global _BUNDLED_FILE3D_CLS
-    if _BUNDLED_FILE3D_CLS is not None:
-        return _BUNDLED_FILE3D_CLS
-    from comfy_api.latest._util.geometry_types import File3D as _File3D
-
-    class BundledFile3D(_File3D):
-        """``File3D``-compatible wrapper for a multi-file 3D bundle.
-
-        Stores the primary mesh file plus N companion files (textures
-        / .mtl / .bin / etc.) in memory as a ``path -> bytes`` dict;
-        lazily materialises the whole bundle into a temp directory the
-        first time anything cross-file-resolution-sensitive is asked
-        of it (``get_source()`` for an OBJ/GLTF that needs its
-        .mtl/.bin/textures on disk; ``save_to(dest)`` which copies the
-        *whole bundle* into the destination directory).
-
-        Cheap consumers that just want the primary file's bytes
-        (``get_data()`` / ``get_bytes()``) skip the temp-dir
-        materialise entirely.
-
-        The temp directory's lifetime is tied to this object: it is
-        cleaned up on garbage-collect via ``__del__`` (best-effort).
-        For workflows that need stable on-disk paths beyond this
-        object's lifetime, call ``save_to(dest)`` to copy the bundle
-        into a caller-controlled location.
-        """
-
-        def __init__(
-            self,
-            files: dict[str, bytes],
-            primary_path: str,
-            roles: dict[str, str] | None = None,
-            formats: dict[str, str] | None = None,
-        ) -> None:
-            if primary_path not in files:
-                raise ValueError(
-                    f"primary_path {primary_path!r} not in bundle files: "
-                    f"{sorted(files)!r}"
-                )
-            self._files: dict[str, bytes] = dict(files)
-            self._primary_path = primary_path
-            self._roles: dict[str, str] = dict(roles or {})
-            self._formats: dict[str, str] = dict(formats or {})
-            self._temp_dir: str | None = None
-            # Populate parent ``_source`` / ``_format`` so reads of
-            # those plain attributes on downstream consumers that
-            # haven't been updated still work. ``_source`` is a
-            # BytesIO of the primary file; ``_format`` is the primary
-            # file's extension.
-            primary_fmt = (
-                self._formats.get(primary_path)
-                or (
-                    primary_path.rsplit(".", 1)[-1].lower()
-                    if "." in primary_path
-                    else ""
-                )
-            )
-            super().__init__(BytesIO(self._files[primary_path]), primary_fmt)
-
-        # --------------------------------------------------------------
-        # File3D-compatible accessors (overrides)
-        # --------------------------------------------------------------
-
-        def get_source(self) -> str:
-            """Return the on-disk path to the primary file.
-
-            Materialises the whole bundle into a temp directory on
-            first call so cross-file refs (.obj -> .mtl -> texture;
-            .gltf -> .bin + textures) resolve via the filesystem.
-            """
-            self._ensure_materialized()
-            from os.path import join as _join
-            return _join(self._temp_dir, *self._primary_path.split("/"))
-
-        def get_data(self) -> BytesIO:
-            """Return the primary file's bytes (no disk I/O)."""
-            return BytesIO(self._files[self._primary_path])
-
-        def get_bytes(self) -> bytes:
-            """Return the primary file's raw bytes (no disk I/O)."""
-            return self._files[self._primary_path]
-
-        def save_to(self, path: str) -> str:
-            """Copy the WHOLE bundle into the destination directory.
-
-            ``path`` is the destination for the *primary file*;
-            sidecars are copied alongside it preserving their relative
-            paths from the original bundle layout. E.g. for
-
-                primary_path = "model.obj"
-                files = {
-                    "model.obj":            <bytes>,
-                    "model.mtl":            <bytes>,
-                    "textures/albedo.png":  <bytes>,
-                }
-
-            ``save_to("/tmp/out/scene.obj")`` writes:
-
-                /tmp/out/scene.obj
-                /tmp/out/model.mtl
-                /tmp/out/textures/albedo.png
-
-            Sidecars keep their ORIGINAL filenames (so cross-file
-            refs inside the .obj that point at ``model.mtl`` still
-            resolve); only the primary file is renamed to match
-            ``path``. The destination directory is created if needed.
-            """
-            from pathlib import Path
-            dest = Path(path)
-            dest.parent.mkdir(parents=True, exist_ok=True)
-            dest.write_bytes(self._files[self._primary_path])
-            dest_parent_resolved = dest.parent.resolve()
-            for rel_path, raw in self._files.items():
-                if rel_path == self._primary_path:
-                    continue
-                sidecar_dest = dest.parent.joinpath(*rel_path.split("/"))
-                # Path safety — rel_path was sanitized at envelope
-                # build time, but defence-in-depth check on the
-                # resolved path keeps us safe against pathological
-                # inputs that may have slipped through.
-                resolved = sidecar_dest.resolve()
-                if dest_parent_resolved not in resolved.parents and resolved != dest_parent_resolved:
-                    continue
-                sidecar_dest.parent.mkdir(parents=True, exist_ok=True)
-                sidecar_dest.write_bytes(raw)
-            return str(dest)
-
-        # --------------------------------------------------------------
-        # Bundle-specific accessors (additive)
-        # --------------------------------------------------------------
-
-        @property
-        def primary_path(self) -> str:
-            return self._primary_path
-
-        @property
-        def file_paths(self) -> list[str]:
-            return sorted(self._files)
-
-        def file_bytes(self, path: str) -> bytes:
-            """Raw bytes for one file in the bundle by relative path."""
-            return self._files[path]
-
-        def file_role(self, path: str) -> str | None:
-            """Optional producer-supplied role hint (may be ``None``)."""
-            return self._roles.get(path)
-
-        def paths_with_role(self, role: str) -> list[str]:
-            """All file paths whose role hint equals ``role``."""
-            return [p for p, r in self._roles.items() if r == role]
-
-        def materialize_to_temp_dir(self) -> str:
-            """Force-materialise the whole bundle into a temp directory.
-
-            Returns the temp-directory path. Cleanup happens on GC;
-            callers that need stable lifetimes should ``save_to()`` to
-            a caller-controlled location instead.
-            """
-            self._ensure_materialized()
-            return self._temp_dir
-
-        # --------------------------------------------------------------
-        # Internal
-        # --------------------------------------------------------------
-
-        def _ensure_materialized(self) -> None:
-            if self._temp_dir is not None:
-                return
-            import tempfile
-            from pathlib import Path
-            self._temp_dir = tempfile.mkdtemp(prefix="rnp_model3d_bundle_")
-            root = Path(self._temp_dir)
-            for rel_path, raw in self._files.items():
-                # rel_path was sanitized at envelope build time —
-                # split-on-'/' and re-join via Path is safe.
-                dest = root.joinpath(*rel_path.split("/"))
-                dest.parent.mkdir(parents=True, exist_ok=True)
-                dest.write_bytes(raw)
-
-        def __del__(self) -> None:
-            td = getattr(self, "_temp_dir", None)
-            if td is not None:
-                try:
-                    import shutil
-                    shutil.rmtree(td, ignore_errors=True)
-                except Exception:  # noqa: BLE001
-                    pass
-
-        def __repr__(self) -> str:
-            return (
-                f"BundledFile3D(primary={self._primary_path!r}, "
-                f"files={len(self._files)}, format={self._format!r})"
-            )
-
-    _BUNDLED_FILE3D_CLS = BundledFile3D
-    return _BUNDLED_FILE3D_CLS
+    return (
+        not isinstance(value, dict)
+        and hasattr(value, "save_to")
+        and hasattr(value, "get_duration")
+    )
 
 
-def decode_model3d_envelope(envelope: dict[str, Any]) -> Any:
-    """Decode a 3D-model envelope into a ComfyUI ``File3D`` object.
+# ---------------------------------------------------------------------------
+# 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).
+# ---------------------------------------------------------------------------
 
-    Dispatches on the envelope's ``encoding``:
+def decode_svg_envelope(envelope: dict[str, Any]) -> Any:
+    """Decode an SVG envelope into a ``comfy_extras.nodes_images.SVG`` instance.
 
-    * ``glb_inline`` — single-file 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.
-
-    * ``bundle_inline`` — multi-file bundle. The primary mesh file
-      plus N companion files (textures / .mtl / .bin / etc.) are
-      decoded into a :class:`BundledFile3D` (``File3D``-compatible)
-      whose ``get_source()`` / ``save_to()`` materialise the whole
-      bundle on disk so cross-file refs resolve via the filesystem.
-      Path safety (no ``..`` / no absolute prefix / no Windows drive
-      letter / no case-insensitive collisions) was enforced at
-      envelope build time on the server; the client re-asserts the
-      ``primary_path``-present invariant defensively and rejects any
-      file entry that lacks ``data``.
+    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)``.
     """
     encoding = envelope.get("encoding")
-    if encoding == "glb_inline":
-        from comfy_api.latest._util import File3D
-        file_format = envelope.get("format") or "glb"
-        return File3D(BytesIO(decode_envelope_data(envelope)), str(file_format))
-    if encoding == "bundle_inline":
-        bundle_cls = _bundled_file3d_class()
-        primary_path = envelope.get("primary_path")
-        if not isinstance(primary_path, str) or not primary_path:
-            raise RnpProtocolError(
-                "bundle_inline envelope missing 'primary_path'",
-                code=ErrorCode.INTERNAL,
-            )
-        files_spec = envelope.get("files")
-        if not isinstance(files_spec, list) or not files_spec:
-            raise RnpProtocolError(
-                "bundle_inline envelope missing non-empty 'files' list",
-                code=ErrorCode.INTERNAL,
-            )
-        files: dict[str, bytes] = {}
-        roles: dict[str, str] = {}
-        formats: dict[str, str] = {}
-        for entry in files_spec:
-            if not isinstance(entry, dict):
-                raise RnpProtocolError(
-                    f"bundle_inline file entry not a dict: {entry!r}",
-                    code=ErrorCode.INTERNAL,
-                )
-            path = entry.get("path")
-            if not isinstance(path, str) or not path:
-                raise RnpProtocolError(
-                    f"bundle_inline file entry missing 'path': {entry!r}",
-                    code=ErrorCode.INTERNAL,
-                )
-            data = entry.get("data")
-            if not isinstance(data, str):
-                raise RnpProtocolError(
-                    f"bundle_inline file {path!r} missing inline 'data'",
-                    code=ErrorCode.INTERNAL,
-                )
-            files[path] = base64.b64decode(data)
-            role = entry.get("role")
-            if isinstance(role, str):
-                roles[path] = role
-            fmt = entry.get("format")
-            if isinstance(fmt, str):
-                formats[path] = fmt
-        if primary_path not in files:
-            raise RnpProtocolError(
-                f"bundle_inline primary_path {primary_path!r} not in "
-                f"files list (have: {sorted(files)!r})",
-                code=ErrorCode.INTERNAL,
-            )
-        return bundle_cls(
-            files=files,
-            primary_path=primary_path,
-            roles=roles,
-            formats=formats,
-        )
-    raise RnpProtocolError(
-        f"Unsupported model_3d encoding: {encoding!r}",
-        code=ErrorCode.INTERNAL,
-    )
-
-
-# ---------------------------------------------------------------------------
-# Cross-node task handle
-#
-# A ``task_handle`` envelope wraps a vendor-native task / video / job
-# ID (Tripo MODEL_TASK_ID, Kling video_id, ...) with lineage metadata
-# so chained provider nodes can pass the handle through the workflow
-# without losing the upstream chain. The decoder returns a
-# :class:`TaskHandle` dataclass — re-encoding the same dataclass on a
-# downstream submit (via :func:`encode_task_handle`) preserves the
-# full ``parent_chain`` without the proxy_node having to know
-# anything provider-specific. Plain ``str`` round-trip would silently
-# drop lineage, so the decoder deliberately surfaces the rich type
-# even at the cost of slightly more work on the consumer side.
-# ---------------------------------------------------------------------------
-
-from dataclasses import dataclass, field
-
-
-@dataclass
-class TaskHandle:
-    """Cross-node handle to a vendor-native async task / video / job.
-
-    See :data:`comfy_remote_nodes.protocol.Capability.IO_TASK_HANDLE`
-    for the wire-shape contract. The dataclass mirrors the envelope
-    fields verbatim so :func:`encode_task_handle` is a trivial
-    round-trip. ``parent_chain`` is a flat list of
-    ``{vendor, kind, native_id, origin_node_id}`` dicts in
-    root-to-immediate-parent order; treat it as opaque on the client
-    — the server is the only component that ever consumes it.
-    """
-
-    vendor: str
-    kind: str
-    native_id: str
-    origin_node_id: str
-    parent_chain: list[dict[str, str]] = field(default_factory=list)
-
-    def __str__(self) -> str:
-        # Workflow runtime sometimes coerces socket values to ``str``
-        # (UI previews, log lines). Surfacing the native id keeps that
-        # output human-meaningful without leaking the lineage chain
-        # into UI strings.
-        return self.native_id
-
-
-def _validate_task_handle_ref(ref: Any, *, context: str) -> dict[str, str]:
-    """Validate one ``{vendor, kind, native_id, origin_node_id}`` ref.
-
-    Mirror of the server-side ``_validate_task_handle_ref`` —
-    rejects malformed parent_chain entries with the same field-by-
-    field error messages so a bad envelope surfaces the same way on
-    both sides.
-    """
-    if not isinstance(ref, dict):
+    if encoding not in ("svg_xml_base64", "svg_xml_inline"):
         raise RnpProtocolError(
-            f"{context}: expected dict, got {type(ref).__name__}",
+            f"Unsupported svg encoding: {encoding!r}",
             code=ErrorCode.INTERNAL,
         )
-    out: dict[str, str] = {}
-    for fld in ("vendor", "kind", "native_id", "origin_node_id"):
-        val = ref.get(fld)
-        if not isinstance(val, str) or not val:
-            raise RnpProtocolError(
-                f"{context}: field {fld!r} must be a non-empty string "
-                f"(got {val!r})",
-                code=ErrorCode.INTERNAL,
-            )
-        out[fld] = val
-    if "," in out["kind"]:
+    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(
-            f"{context}: 'kind' must be a singular socket kind, not a "
-            f"comma-union string (got {out['kind']!r})",
+            "SVG envelope payload contained no <svg> documents",
             code=ErrorCode.INTERNAL,
         )
-    return out
-
-
-def decode_task_handle_envelope(envelope: dict[str, Any]) -> TaskHandle:
-    """Decode a ``task_handle`` envelope into a :class:`TaskHandle`.
-
-    Only the ``vendor_inline`` encoding is recognised today (a future
-    sibling ``replay_ticket_inline`` would carry a signed server-
-    local replay token for cross-server portability — not in this
-    PR). All four top-level fields and every ``parent_chain`` entry
-    are validated; rejections raise ``RnpProtocolError(INTERNAL)``
-    with the same field-by-field messages the server emits at
-    envelope-build time.
-    """
-    encoding = envelope.get("encoding")
-    if encoding != "vendor_inline":
-        raise RnpProtocolError(
-            f"Unsupported task_handle encoding: {encoding!r}",
-            code=ErrorCode.INTERNAL,
-        )
-    top = _validate_task_handle_ref(
-        {
-            "vendor":         envelope.get("vendor"),
-            "kind":           envelope.get("kind"),
-            "native_id":      envelope.get("native_id"),
-            "origin_node_id": envelope.get("origin_node_id"),
-        },
-        context="task_handle",
-    )
-    chain_in = envelope.get("parent_chain", [])
-    if not isinstance(chain_in, list):
-        raise RnpProtocolError(
-            f"task_handle 'parent_chain' must be a list (got "
-            f"{type(chain_in).__name__})",
-            code=ErrorCode.INTERNAL,
-        )
-    chain_out = [
-        _validate_task_handle_ref(entry, context=f"parent_chain[{i}]")
-        for i, entry in enumerate(chain_in)
-    ]
-    return TaskHandle(
-        vendor=top["vendor"],
-        kind=top["kind"],
-        native_id=top["native_id"],
-        origin_node_id=top["origin_node_id"],
-        parent_chain=chain_out,
-    )
-
-
-def encode_task_handle(handle: TaskHandle) -> dict[str, Any]:
-    """Encode a :class:`TaskHandle` back into a wire-format envelope.
-
-    Symmetric counterpart of :func:`decode_task_handle_envelope` so a
-    handle decoded from one provider's output and fed into another
-    provider's input round-trips losslessly — the proxy_node calls
-    this whenever it sees a ``TaskHandle`` in the input values for a
-    descriptor that accepts the ``task_handle`` envelope type.
-
-    The returned dict is the same shape the server's
-    ``make_task_handle_envelope`` produces, so the server-side
-    decoder accepts it without any branching on "is this a fresh
-    handle or a re-encoded one" — the wire is the source of truth.
-    """
-    return {
-        "type":           "task_handle",
-        "encoding":       "vendor_inline",
-        "vendor":         handle.vendor,
-        "kind":           handle.kind,
-        "native_id":      handle.native_id,
-        "origin_node_id": handle.origin_node_id,
-        "parent_chain":   [dict(entry) for entry in handle.parent_chain],
-    }
+    return SVG(parts)
 
 
 # ---------------------------------------------------------------------------
@@ -948,12 +374,11 @@ def encode_task_handle(handle: TaskHandle) -> dict[str, Any]:
 # ---------------------------------------------------------------------------
 
 _DECODERS = {
-    "image":       decode_image_envelope,
-    "mask":        decode_mask_envelope,
-    "audio":       decode_audio_envelope,
-    "video":       decode_video_envelope,
-    "model_3d":    decode_model3d_envelope,
-    "task_handle": decode_task_handle_envelope,
+    "image": decode_image_envelope,
+    "mask":  decode_mask_envelope,
+    "audio": decode_audio_envelope,
+    "video": decode_video_envelope,
+    "svg":   decode_svg_envelope,
 }
 
 
@@ -1002,15 +427,13 @@ __all__ = [
     "decode_mask_envelope",
     "encode_audio_input",
     "decode_audio_envelope",
+    "encode_video_input",
     "decode_video_envelope",
-    "decode_model3d_envelope",
-    "_bundled_file3d_class",
-    "TaskHandle",
-    "decode_task_handle_envelope",
-    "encode_task_handle",
+    "decode_svg_envelope",
     "decode_envelope",
     "is_envelope",
     "is_image_tensor",
     "is_mask_tensor",
     "is_audio_input",
+    "is_video_input",
 ]