Compare commits

...
Author SHA1 Message Date
Jedrzej Kosinski 0946b83b21 Merge pull request #22 from Comfy-Org/feat/encode-nested-autogrow-dicts
fix(encoder): recurse into nested AUTOGROW / DYNAMIC_COMBO runtime dicts
2026-05-17 08:22:55 -07:00
Jedrzej KosinskiandAmp 866b7e32a3 fix(encoder): recurse into nested AUTOGROW / DYNAMIC_COMBO runtime dicts
ComfyUI's V3 ``build_nested_inputs`` (comfy_api/latest/_io.py L1722) rebuilds
AUTOGROW / DYNAMIC_COMBO runtime kwargs into nested dicts before they reach
the proxy_node's ``execute()``. For an AUTOGROW input named
``reference_images``, the value arrives as ``{"image1": <tensor>,
"image2": <tensor>, ...}``; for a DYNAMIC_COMBO branch with a nested
AUTOGROW (e.g. Wan2ReferenceVideoApi's ``model`` DC with nested
``video1..video3`` + ``image1..image5``), the value arrives as
``{"<dc-name>": branch_key, "video_refs": {"video1": <VideoInput>},
"reference_images": {"image1": <tensor>}, ...}``.

Before this change ``_encode_one`` only encoded values where the **top-
level** value was itself a torch.Tensor / VideoInput / audio dict /
model3d input. A non-audio dict (the AG / DC runtime shape) failed all
four duck-type checks and was returned unchanged, so per-slot tensors /
VideoInputs survived past encoding and either crashed at
``json.dumps`` time (no custom JSONEncoder for torch.Tensor) or
violated the server's contract (comfy_rnp_server's
``_ordered_wan_videoedit_image_envelopes`` and the WAN2 / HappyHorse /
Grok / Bria / Luma AG providers shipped over the last 5 PRs all
require each slot value to already be an RNP envelope before they
reach the upload step).

The same gap also defeated the externalization cap: ``_encode_inputs``
only ran ``maybe_externalize`` when the *top-level* encoded value
was itself an envelope, so per-slot envelopes inside an AG dict would
ship inline regardless of ``max_inline_bytes``.

Fix is structural — match the runtime data model:

1. ``_encode_one`` recurses into plain (non-envelope, non-audio) dict
   values, calling itself for each entry. Envelope dicts pass through
   so we don't double-encode; audio dicts still route to
   ``encode_audio_input`` because their dict shape (waveform +
   sample_rate) is a heavy leaf, not a container. The recursion
   passes the parent's ``accepted_encodings`` downward — single-image
   AG slots and VIDEO / AUDIO / MODEL_3D leaves don't depend on it,
   and gating per-child policy on descriptor metadata adds plumbing
   for no concrete provider need today.

2. New ``_externalize_nested`` helper walks the encoded tree the same
   way: envelope leaves go through ``serialization.maybe_externalize``;
   plain dicts recurse; everything else passes through. ``_encode_inputs``
   replaces the old "if is_envelope: externalize" guard with one call
   to the recursive walker, so per-slot envelopes inside AG / DC dicts
   get the same presigned-PUT treatment top-level envelopes get.

Lists are intentionally NOT traversed — AG / DC runtime values always
materialize as dicts, never lists, and recursing into arbitrary lists
would touch opaque payloads. Recursion depth is bounded by the
upstream Autogrow max-slot cap (100) and the maximum DC branch
nesting depth observed in any provider today (1), so an in-place
``async def`` walk is well-bounded.

Tests:

* New ``notes/run_encode_nested_dicts.py`` standalone smoke test
  exercises 9 scenarios: top-level AG-of-IMAGE, top-level AG-of-VIDEO,
  DC-wrapping-AG with branch key untouched, AUDIO dict still routes to
  encode_audio_input, pre-encoded envelope passes through unchanged,
  scalar config dict passes through unchanged, ``_externalize_nested``
  walks per-slot envelopes (cap=3 forces every slot to externalize),
  scalar siblings preserved alongside externalized envelopes, full
  encoded payload is JSON-serializable (regression guard for the
  exact bug fixed), and single-tensor / single-video / single-audio
  top-level inputs still encode unchanged (no leaf-path regression).
* Loads ``_encode_one`` / ``_encode_inputs`` / ``_externalize_nested``
  out of proxy_node.py via AST extraction so it runs without
  importing the full comfy_api / comfy_api_nodes packages, same
  pattern as ``run_image_max_batch.py``.

Unblocks the next batch of comfy-rnp-server providers that need
AG-of-VIDEO or DC-wrapping-AG-of-VIDEO support (WanReferenceVideoApi
``character1..character3`` and Wan2ReferenceVideoApi
``video1..video3`` + ``image1..image5``).

Co-authored-by: Amp <amp@ampcode.com>
Amp-Thread-ID: https://ampcode.com/threads/T-019e367d-c5b5-758d-9387-e44477ef4cb5
2026-05-17 08:22:05 -07:00
Jedrzej Kosinski 1d70ef223b Merge pull request #21 from Comfy-Org/feat/model3d-input-multitype-mapping
Map MODEL_3D input type to MultiType socket (mirror upstream union)
2026-05-17 06:57:47 -07:00
Jedrzej KosinskiandAmp 197e7ba406 Map "MODEL_3D" input type to MultiType socket (mirror upstream union)
Follow-up to PR #20: the `MODEL_3D` input wire type now resolves to
an `IO.MultiType.Input(types=[IO.File3DGLB, IO.File3DOBJ,
IO.File3DFBX, IO.File3DAny])` socket — symmetric counterpart of the
existing `"MODEL_3D"` entry in `_OUTPUT_CLASSES` that maps to
`IO.File3DGLB.Output`.

Previously the descriptor wire type `"MODEL_3D"` fell through to the
opaque `IO.Custom("MODEL_3D").Input(...)` socket, which is a
single-comfytype socket the frontend only allows connections from
identically-tagged outputs. Real upstream 3D sources expose
different comfytypes:

- `Load3D` outputs `IO.File3DAny` (comfytype `FILE_3D`).
- Meshy / Hunyuan3D / Tripo / Rodin partner nodes output
  `IO.File3DGLB` (comfytype `FILE_3D_GLB`) via the existing
  `_OUTPUT_CLASSES["MODEL_3D"]` mapping.
- Meshy's dual `FILE_3D_FBX` output socket emits `FILE_3D_FBX`.

None of these connect to `IO.Custom("MODEL_3D")`. The upstream
Tencent Hunyuan3D MODEL_3D-input nodes (`TencentModelTo3DUVNode` /
`Tencent3DTextureEditNode` / `Tencent3DPartNode` /
`TencentSmartTopologyNode`) all use `IO.MultiType.Input(types=[...])`
to accept the union; mirroring that union here keeps the proxy
socket compatible with every existing 3D-emitting node.

`get_io_type()` on `IO.MultiType.Input` returns the comma-joined
inner type IDs (e.g. `"FILE_3D_GLB,FILE_3D_OBJ,FILE_3D_FBX,FILE_3D"`)
— matches the convention SxS workflows already use for their
`SaveGLB` connector inputs.

Per-node format validation (e.g. 3DTextureEdit's FBX-only
enforcement, 3DPart's FBX-only enforcement) is done server-side in
each provider's `execute()` against the inline envelope's `format`
extra; this socket-level filter is purely about frontend connection
validity.

Unblocks the paired comfy-rnp-server PR (`TencentModelTo3DUVNode_RNP`
+ `Tencent3DPartNode_RNP`) which uses the `["MODEL_3D", {}]` wire
shape.

Co-authored-by: Amp <amp@ampcode.com>
Amp-Thread-ID: https://ampcode.com/threads/T-019e362f-db40-7338-8ca4-99f617d89812
2026-05-17 06:57:14 -07:00
Jedrzej Kosinski 95a3073d79 Merge pull request #20 from Comfy-Org/feat/model3d-input-encoder
Add MODEL_3D input encoder to close the encode-side gap
2026-05-17 06:53:57 -07:00
Jedrzej KosinskiandAmp 01f71bd806 Add MODEL_3D input encoder to close the encode-side gap
Adds the symmetric client-side encoder for `File3D` values so a remote
node descriptor can declare a `MODEL_3D` input and consume a 3D mesh
from an upstream local node (e.g. `Load3D`, `TencentTextToModelNode`,
`MeshyTextToModelNode`).

Mirrors the existing AUDIO encoder (`encode_audio_input`,
`is_audio_input`) and VIDEO encoder (`encode_video_input`,
`is_video_input` — PR #19) added at
`comfy_remote_nodes/serialization.py`:

- `encode_model3d_input(file3d)` reads the primary mesh bytes via
  `File3D.get_data()` (returns a `BytesIO`, per
  `comfy_api.latest._util.geometry_types.File3D`) and emits a
  `{type: "model_3d", encoding: "glb_inline", data: b64, format: ...}`
  envelope. `encoding="glb_inline"` is the generic single-file inline
  encoding the server-side decoder (`decode_model3d_envelope`)
  already dispatches on; the decoder hands the bytes to
  `File3D(BytesIO(...), format)` regardless of whether the actual
  format is glb / obj / fbx / etc., so the encoding name is a
  logical "inline single-file 3D" tag and the `format` extra picks
  the file extension at decode time.
- `is_model3d_input(value)` duck-types on the `format` string
  attribute + `get_data` callable. Both are present on the upstream
  `File3D` class and the `BundledFile3D` subclass built lazily in
  `_bundled_file3d_class()`. Avoids importing `File3D` at module load
  (which would pull torch via the `comfy_api` tree).
- `_encode_one` in `proxy_node.py` dispatches AUDIO / VIDEO before
  MODEL_3D as defense-in-depth (the `VideoInput` shape — `save_to`
  + `get_duration` — does not expose a `format` string property, so
  VIDEO and MODEL_3D values don't collide either way).

Multi-file bundles (`BundledFile3D`-style values where every
companion file matters — e.g. an OBJ with sibling .mtl + texture
PNGs) are intentionally not supported on the encode path in this
PR: the first server consumer is the Tencent Hunyuan3D
MODEL_3D-input node family (`TencentModelTo3DUVNode` /
`Tencent3DPartNode` / `Tencent3DTextureEditNode` /
`TencentSmartTopologyNode`), and the upstream upload helper
`upload_3d_model_to_comfyapi(cls, model_3d, file_format)` only
uploads a single file (`model_3d.get_data()`). Bundle encode
(`bundle_inline`) lands when a partner node actually needs the
companion bytes.

This unblocks 4 Tencent Hunyuan3D nodes (out of 6 in the module)
that were deferred to a follow-up PR — see matrix coverage TODO #6.
First server consumers ship in a paired comfy-rnp-server PR
(TencentModelTo3DUVNode_RNP + Tencent3DPartNode_RNP).

Co-authored-by: Amp <amp@ampcode.com>
Amp-Thread-ID: https://ampcode.com/threads/T-019e362f-db40-7338-8ca4-99f617d89812
2026-05-17 06:53:14 -07:00
Jedrzej Kosinski 569212f1e7 Merge pull request #19 from Comfy-Org/feat/video-input-encoder
Add VIDEO input encoder to close the encode-side gap
2026-05-17 05:08:38 -07:00
Jedrzej KosinskiandAmp 04ca70a853 Add VIDEO input encoder to close the encode-side gap
The proxy_node has accepted VIDEO outputs since PR #13 (server-side
`make_video_envelope` + client-side `decode_video_envelope`), but
the inbound encoding direction has been a no-op: `_encode_one` in
`proxy_node.py` only dispatched AUDIO + IMAGE + MASK tensors, so any
remote node declaring an `IO.Video.Input` would receive the raw
`VideoInput` object as a non-JSON-serialisable Python value and the
upstream serialisation would either drop it or fail.

`serialization.py:468` made the gap explicit with a placeholder
comment ("encode lands when a remote node accepts VIDEO inputs"),
and several server-side providers shipped against the
`VIDEO_MP4_BASE64` capability + `input_serialization={"video":
"mp4_base64"}` declarations on the assumption that the client
encoder would land soon — most notably `GrokVideoEditProvider`
and `GrokVideoExtendProvider`, whose end-to-end VIDEO upload path
was broken until now.

This change wires the encode direction:

* `serialization.py` gains `encode_video_input(video)` — mirrors
  the AUDIO encoder above. Re-encodes the `VideoInput` to an
  in-memory mp4/H.264 byte buffer via the upstream
  `comfy_api_nodes/util/conversions.video_to_base64_string` (same
  call shape: `video.save_to(buf, format=MP4, codec=H264)`) and
  base64-encodes the result. Populates `duration_s` from
  `video.get_duration()` so server-side providers (e.g.
  `_validate_video_duration_envelope` in `grok.py`) can range-check
  without demuxing the MP4.

* `serialization.py` gains `is_video_input(value)` — duck-types on
  the `save_to` + `get_duration` method pair (both defined on
  `comfy_api.latest._input.VideoInput` and present on every
  concrete subclass). Avoids importing `VideoInput` at module
  load (which would pull torch via the `comfy_api` tree). Both
  helpers added to `__all__`.

* `proxy_node.py:_encode_one` gains a VIDEO branch right after the
  AUDIO branch (and before the tensor-rank dispatch), mirroring the
  AUDIO pattern at the same call site.

* `proxy_node.py` `_inputs_to_envelopes` docstring updates the
  duck-typing-rules list to mention the new VIDEO branch.

Symmetric to the (still-open) MODEL_3D INPUT gap noted at
`serialization.py:484`. AUDIO INPUT was already wired (ElevenLabs
nodes exercise it end-to-end); IMAGE / MASK have always been
wired; VIDEO INPUT closes today's last input-direction envelope
gap for the existing capability vocabulary.

Verified by importing `serialization` with a stubbed
`comfy_remote_nodes` package: `is_video_input` returns False for
None / dict / audio-shaped dict and True for an object exposing
both `save_to` + `get_duration`. `encode_video_input` + the
proxy_node integration are exercised end-to-end by the server-side
`Wan2VideoContinuationApi_RNP` provider landing in the matching
comfy-rnp-server PR (it consumes `first_clip: VIDEO` and uploads
the resolved bytes to Comfy storage), and also unblocks the
shipped-but-broken Grok VideoEdit/VideoExtend providers.

Co-authored-by: Amp <amp@ampcode.com>
Amp-Thread-ID: https://ampcode.com/threads/T-019e35d1-3c68-74ce-9918-fe9cacd74276
2026-05-17 05:07:57 -07:00
Jedrzej Kosinski ed5823e8e7 Merge pull request #18 from Comfy-Org/revert/task-handle-decoder
Revert "Add client-side cross-node task_handle decoder" (PR #17)
2026-05-16 22:08:11 -07:00
Jedrzej KosinskiandAmp 22e0742d47 Revert "Add client-side cross-node task_handle decoder" (PR #17)
The task_handle envelope was overcooked for what the RNP/1 prototype
actually needs. The goal of the prototype is to move the existing
ComfyUI partner nodes to execute server-side without changing what
they do — and the upstream Tripo / Kling task chaining already works
fine over a plain ``str``:

* Tripo uses ``IO.Custom("MODEL_TASK_ID")`` / ``IO.Custom("RIG_TASK_ID")``
  / ``IO.Custom("RETARGET_TASK_ID")`` sockets, which is purely a
  client-side graph-validation mechanism. The value on the noodle is
  just a string — already covered by ``Capability.IO_OPAQUE``, which
  preserves the type string for connection validity and round-trips
  the value untouched.
* Kling uses plain ``IO.String`` for ``video_id`` — covered trivially
  by the existing string handling.

The ``TaskHandle`` dataclass + ``vendor`` / ``parent_chain`` /
``origin_node_id`` machinery were forward-looking infrastructure for
future cascading-replay semantics that the upstream partner nodes
don't have. Out of scope for "move execution server-side, change
nothing else".

Reverts merge commit 90ecd32 (PR #17: ``feat/task-handle-decoder``).
``Capability.IO_TASK_HANDLE``, the ``task_handle`` HEAVY_TYPES entry,
``TaskHandle`` dataclass, ``decode_task_handle_envelope``,
``encode_task_handle``, the dispatcher entry, the
``CLIENT_CAPABILITIES`` advertisement, and the smoke test are all
removed. Matching server-side revert in comfy-rnp-server.

Co-authored-by: Amp <amp@ampcode.com>
Amp-Thread-ID: https://ampcode.com/threads/T-019e3431-c416-742b-973c-e262ab13f2ff
2026-05-16 22:07:22 -07:00
Jedrzej Kosinski 90ecd3218c Merge pull request #17 from Comfy-Org/feat/task-handle-decoder
Add client-side cross-node task_handle decoder + TaskHandle dataclass
2026-05-16 21:34:42 -07:00
Jedrzej KosinskiandAmp 1f136526cf Add client-side cross-node task_handle decoder + TaskHandle dataclass
Mirrors comfy-rnp-server PR #39 on the client. Adds the surfaces
needed for chained partner provider nodes (Tripo Texture/Refine/
Rig/Retarget/Conversion, Kling VideoExtend, future partner chained
nodes) to pass vendor-native task / video / job IDs through a
workflow with their lineage metadata intact.

Surfaces:

* ``protocol.Capability.IO_TASK_HANDLE = "io:task_handle"`` and
  ``"task_handle"`` added to ``protocol.HEAVY_TYPES`` so
  ``is_envelope`` recognises the new envelope.
* ``serialization.TaskHandle`` dataclass with the wire-shape fields
  (``vendor``, ``kind``, ``native_id``, ``origin_node_id``,
  ``parent_chain``). ``__str__`` surfaces ``native_id`` for log /
  UI preview without leaking the lineage chain.
* ``serialization.decode_task_handle_envelope`` validates every
  field (mirroring server-side ``_validate_task_handle_ref`` so a
  bad envelope surfaces the same error message on both sides) and
  returns the dataclass. Rejects:
  - wrong encoding (only ``vendor_inline`` recognised);
  - missing / empty top-level fields;
  - comma-union ``kind`` strings (the input-acceptance string used
    by some provider nodes, NOT an emitted handle kind);
  - non-list ``parent_chain``;
  - malformed parent_chain entries (missing fields, non-dict,
    comma-union kind in lineage).
* ``serialization.encode_task_handle`` re-encodes the dataclass
  back to wire shape — symmetric counterpart of the decoder so a
  handle decoded from one provider's output and fed into another
  provider's input round-trips losslessly without ``proxy_node``
  having to know anything provider-specific.
* ``decode_envelope`` dispatcher routes ``type="task_handle"`` to
  the new decoder so the generic ``_deserialize_output`` path picks
  it up.
* ``client.CLIENT_CAPABILITIES`` advertises
  ``Capability.IO_TASK_HANDLE`` so server-side capability-gated
  descriptors (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) negotiate cleanly.

The dataclass return (NOT a plain ``str``) is deliberate: a plain
string would silently drop ``parent_chain`` when the value is
passed through the workflow, breaking the lineage the server uses
for future cascading-replay semantics (deferred to a follow-up PR
alongside server-side replay-cache storage — see PR #39 for the
replay design notes).

Smoke test ``notes/run_task_handle_decoder.py`` mirrors the pattern
used by ``run_model3d_bundle_decoder.py``: stubs out the ``client``
import for the relative ``from . import client as rnp_client`` at
the top of serialization.py, then exercises the 9 contract points
end-to-end. End-to-end check imports the server's
``make_task_handle_envelope`` from the comfy-rnp-server worktree
and verifies the wire shape round-trips losslessly through
decode -> encode.

Co-authored-by: Amp <amp@ampcode.com>
Amp-Thread-ID: https://ampcode.com/threads/T-019e3431-c416-742b-973c-e262ab13f2ff
2026-05-16 21:34:05 -07:00
Jedrzej Kosinski e5db513585 Merge pull request #16 from Comfy-Org/feat/model-3d-bundle-decoder
Add client-side multi-file 3D bundle decoder + BundledFile3D
2026-05-16 21:16:09 -07:00
Jedrzej KosinskiandAmp c962fa5adf Add client-side multi-file 3D bundle decoder + BundledFile3D
Closes the client side of the multi-file 3D bundle envelope landed in
comfy-rnp-server#38 (commit 50f9f4f on main).

* Vendored `protocol.py`: mirrors the new
  `Capability.MODEL_3D_BUNDLE_INLINE = "model_3d:bundle_inline"`
  constant with a full docstring describing the wire shape, path /
  role semantics, primary_path / format invariants, capability
  negotiation discipline, and the planned future
  `MODEL_3D_BUNDLE_URI` sibling. HEAVY_TYPES is unchanged
  (`type="model_3d"` already covers the new encoding).

* `serialization.py`: adds `BundledFile3D` as a `File3D` subclass
  built lazily via a `_bundled_file3d_class()` factory (so importing
  the heavy `comfy_api.latest._util` package doesn't happen at
  module load — keeps the smoke-test stubbing pattern intact for
  callers that don't exercise the bundle path). The class stores
  the primary mesh file plus N companion files in memory as a
  `path -> bytes` dict and:

  - `get_data()` / `get_bytes()` return the primary file's bytes
    without touching disk.
  - `get_source()` lazily materialises the whole bundle into a temp
    directory using the producer-supplied relative paths, then
    returns the on-disk path to the primary file — so a .obj that
    references its .mtl by filename (and the .mtl that references
    `textures/albedo.png`) resolves via the filesystem.
  - `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 that point at `model.mtl` still resolve when
    the user saves the model somewhere stable.
  - New convenience accessors `primary_path`, `file_paths`,
    `file_bytes(path)`, `file_role(path)`, `paths_with_role(role)`,
    and `materialize_to_temp_dir()`.
  - Temp directory cleanup on garbage collect (best-effort
    `shutil.rmtree(td, ignore_errors=True)` in `__del__`).
  - `super().__init__(BytesIO(primary), primary_fmt)` keeps the
    upstream `_source` / `_format` attributes populated so existing
    downstream consumers that read those plain attributes still
    work.

  `decode_model3d_envelope` now dispatches on `encoding`:
  `glb_inline` falls through to the existing single-file path
  unchanged; `bundle_inline` walks the new wire shape, decodes
  per-file base64 `data`, hoists `role` / `format` hints into
  parallel dicts, and constructs a `BundledFile3D`. Malformed
  envelopes (missing `primary_path`, missing `files`, primary not
  present in files, file entry missing `data`) raise
  `RnpProtocolError(code=INTERNAL)` rather than falling through to
  any opaque bucket.

* `client.py`: advertises `Capability.MODEL_3D_BUNDLE_INLINE` in
  `CLIENT_CAPABILITIES` with a full docstring matching the
  established pattern for the other 3D capabilities.

Design decisions (from oracle consult on wire-shape surface):

* Extend the existing `model_3d` heavy type with a new encoding
  (rather than introducing `model_3d_bundle` as a separate
  top-level type) — keeps HEAVY_TYPES unchanged, lets the
  single-file `glb_inline` decoder stay untouched, and lets
  capability gating stay parallel.

* `path` is the authoritative cross-file reference key (not
  `role`) — roles are conceptually non-unique (multiple meshes,
  LODs, buffers, animations) so `primary_path` pins the entry
  point unambiguously.

* `role` is an OPTIONAL producer hint, NOT a closed enum. Unknown
  roles round-trip without error.

* Subclass `File3D` directly rather than introducing a new opaque
  `ModelBundle3D` type — existing downstream loaders that
  type-check or duck-type-check on `File3D` keep working.
  `isinstance(bundled, File3D)` succeeds.

* Each file's `data` is base64 (mirrors the single-file
  `glb_inline` payload mode). Mixed `data`/`uri` inside one
  encoding was rejected per the parallel-encoding pattern already
  in use across the rest of the protocol; a future sibling
  `Capability.MODEL_3D_BUNDLE_URI` would handle URL-fetch
  semantics.

* Capability gating with whole-descriptor enforcement — 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.

Smoke test at `notes/run_model3d_bundle_decoder.py` — 9 checks:

1. Capability constant intact at protocol layer.
2. is_envelope recognises bundle_inline envelopes.
3. decode dispatch returns BundledFile3D (a File3D subclass) +
   primary bytes round-trip via get_data / get_bytes.
4. get_source materialises whole bundle on disk at producer-
   supplied paths.
5. save_to copies whole bundle: primary renamed, sidecars keep
   filenames.
6. Bundle accessors (primary_path / file_paths / file_role /
   paths_with_role) work; unknown roles round-trip.
7. Malformed bundle envelopes raise RnpProtocolError(INTERNAL).
8. End-to-end against server's make_model3d_bundle_envelope helper
   (from PR #38) — all files / roles / bytes preserve. Server-side
   path sanitization rejects `..` at envelope build time
   (defence-in-depth).
9. CLIENT_CAPABILITIES advertises model_3d:bundle_inline.

All 9 pass; PR #1's `notes/run_model3d_decoder.py` (8 checks) and
PR #2's `notes/run_dynamic_combo_parser.py` (8 checks) also still
pass — no regression on existing paths.

Hunyuan3D / Tencent / future Meshy / Rodin / Tripo bundle
providers land in a follow-up PR that consumes this decoder; this
PR is protocol + client decoder only per the planned sequence.

Co-authored-by: Amp <amp@ampcode.com>
Amp-Thread-ID: https://ampcode.com/threads/T-019e331c-c202-746b-96ec-86ce0b9c54a5
2026-05-16 21:15:29 -07:00
Jedrzej Kosinski d054d978cd Merge pull request #15 from Comfy-Org/feat/dynamic-combo-autogrow-parser
Parse DYNAMIC_COMBO + AUTOGROW inputs into real V3 IO objects
2026-05-16 16:42:33 -07:00
Jedrzej KosinskiandAmp 37b77f837a Parse DYNAMIC_COMBO + AUTOGROW inputs into real V3 IO objects
`proxy_node._parse_input_spec` previously had explicit branches only
for the primitive V3 input types (STRING / INT / FLOAT / BOOLEAN /
COMBO / IMAGE / VIDEO / AUDIO / MASK); every other `io_type` string —
including the two V3 dynamic input types `DYNAMIC_COMBO` and
`AUTOGROW` — fell through to the opaque `IO.Custom(io_type).Input`
bucket. Server-side runtime validation already worked (rules-
extraction at `_rules_from_input_spec` walks `__branches__` and
`__template__`), but on the frontend every Topaz / Magnific / Vidu /
Luma / Wan / Bria / Grok / ElevenLabs / HitPaw provider that declares
a `DynamicCombo` rendered as one opaque socket instead of the
branch-selector dropdown with per-branch sub-widgets the V3 IO is
designed to produce.

Refactor the parser so the construction logic is shared between two
entry points (top-level descriptor inputs use the 2-tuple
`[io_type, options]` form; DynamicCombo branch sub-inputs use the
3-tuple `[name, io_type, options]` form) and add real handlers for
both dynamic types:

* `_parse_input_spec(name, spec, optional)` — top-level entry, same
  signature as before. Strips the `[io_type, options]` head and
  delegates.
* `_parse_named_input_spec(spec, optional)` — branch-sub-input entry.
  Strips the leading name and delegates.
* `_build_input(name, io_type, options, optional)` — shared
  constructor dispatch. All primitive io_types route through the same
  bag of kwargs; the new `DYNAMIC_COMBO` branch builds
  `IO.DynamicCombo.Input(name, options=[IO.DynamicCombo.Option(key,
  inputs=[...]), ...])` by recursing into `_parse_named_input_spec`
  for each branch sub-input; the new `AUTOGROW` branch builds
  `IO.Autogrow.Input(name, template=IO.Autogrow.TemplatePrefix(...))`
  (or `TemplateNames`) by recursing into `_parse_input_spec` for the
  template entry.

The flattened RNP wire dialect for `AUTOGROW` (`template` is a flat
`[io_type, options]` entry directly, with `prefix` / `min` / `max` or
`names` / `min` as siblings of `template`) is what every server
provider already emits (e.g. `grok.py:_grok_v2_model_subinputs`); the
parser mirrors that shape rather than the upstream
`Autogrow.Input.as_dict` nested form.

Guard rails:

* Empty branch input lists are valid (Bria moderation's `"false"`
  branch carries `"inputs": []`).
* The branch selector's id is the outer input id (`"moderation"`,
  `"inputs"`, etc.) — not always `"model"`.
* Branch sub-inputs are treated as required at schema-construction
  time; the wire dialect doesn't express per-branch optionality, and
  server-side `__branches__` enforcement covers runtime presence.
* AUTOGROW template inputs that are themselves dynamic
  (`DynamicCombo.Input` / `Autogrow.Input`) are rejected at parse
  time, mirroring the upstream `_AutogrowTemplate.__init__` assert.
* Malformed `DYNAMIC_COMBO` / `AUTOGROW` specs return `None` (caller
  logs + skips the descriptor) instead of degrading to a single
  opaque socket — the opaque fallthrough is reserved for genuinely
  unknown io_type strings, not for known-but-broken dynamic specs.
* `advanced` only routes to primitive `WidgetInput` constructors; the
  dynamic constructors don't accept it.

No capability gating in this PR. The two dynamic io_type strings are
already V3-canonical so a legacy client without the parser fix sees
the same opaque-socket degradation it sees today (just rendered as
`Custom("DYNAMIC_COMBO")` instead of the real DynamicCombo input).
Adding `Capability.DYNAMIC_COMBO` / `Capability.AUTOGROW` would
require pairing the cap with real descriptor-load-time enforcement
(server-side filtering of `/object_info` or client-side rejection
based on `required_capabilities`); neither exists yet, so adding
caps now would just be inert metadata. The cap pair is the right
follow-up alongside the enforcement plumbing.

Smoke test at `notes/run_dynamic_combo_parser.py` — 8 checks covering
Bria's 2-branch DynamicCombo (with empty branch), Grok V2's 3-branch
DynamicCombo with nested AUTOGROW(IMAGE), ElevenLabs' branch mixing
custom-IO (voiceN) + primitive STRING, a synthetic nested
DynamicCombo-in-DynamicCombo (defensive recursion coverage), the
malformed-spec rejection path, a regression check that
`_collect_local_validate` output is byte-identical before and after,
the AUTOGROW-rejects-dynamic-template assertion, and a source-level
check that the new helpers + branches exist.

Co-authored-by: Amp <amp@ampcode.com>
Amp-Thread-ID: https://ampcode.com/threads/T-019e331c-c202-746b-96ec-86ce0b9c54a5
2026-05-16 16:41:38 -07:00
Jedrzej Kosinski 1113a80229 Merge pull request #14 from Comfy-Org/feat/client-model-3d-decoder
feat(model_3d): add client-side MODEL_3D decoder + capability advertisement
2026-05-16 16:25:03 -07:00
Jedrzej KosinskiandAmp 58d93daf2a feat(model_3d): add client-side MODEL_3D decoder + capability advertisement
Closes the loop on the just-merged comfy-rnp-server PR #37 which added
the four Tripo / Rodin 3D providers and the
``Capability.MODEL_3D_GLB_INLINE`` token. Without this client-side half
the providers were unreachable from any RNP client — descriptors would
fail negotiation because the proxy_node didn't advertise the token.

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

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

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

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

Co-authored-by: Amp <amp@ampcode.com>
Amp-Thread-ID: https://ampcode.com/threads/T-019e3314-4336-71d2-a333-6d1e33a9986e
2026-05-16 16:24:11 -07:00
Jedrzej Kosinski 97fbfd892c Merge pull request #13 from Comfy-Org/feat/audio-opus-base64
feat(audio): add Capability.AUDIO_OPUS_BASE64 + decode_audio_envelope branch
2026-05-15 18:25:32 -07:00
Jedrzej KosinskiandAmp cd3e051956 feat(audio): add Capability.AUDIO_OPUS_BASE64 + decode_audio_envelope branch
Pairs with comfy-rnp-server feat/audio-opus-base64. Lets the
proxy_node accept and decode `audio:opus_base64` envelopes (OGG/
Opus container as base64) so audio-out partner-node providers
whose upstream returns OGG/Opus bytes (e.g. ElevenLabs
`output_format=opus_48000_192`) can ship an honest envelope
instead of label-laundering the bytes as `mp3_base64`.

protocol.py:
- `Capability.AUDIO_OPUS_BASE64 = "audio:opus_base64"`.

client.py:
- `CLIENT_CAPABILITIES` now includes `Capability.AUDIO_OPUS_BASE64`,
  matching the AUDIO sweep precedent (mp3 + wav already in the list).

serialization.py:
- `decode_audio_envelope` allow-list extended with `opus_base64`.
  PyAV/FFmpeg already decode OGG/Opus natively in
  `audio_bytes_to_audio_input`, so the decoder branch is a one-line
  allow-list change — same path as MP3/WAV, just a different
  container. The wire-format hint is now honest rather than label-
  laundering, since the inbound envelope's `encoding` matches the
  actual container.

The capability negotiation is symmetric to the existing
`image:png_base64_batch` / `audio:mp3_base64` negotiation: servers
gate emission on the inbound advertise list, and the per-format
fallback (mp3) is required while opus is opportunistic.

Amp-Thread-ID: https://ampcode.com/threads/T-019e2e33-da42-74c2-a6cf-8c7942cb8180
Co-authored-by: Amp <amp@ampcode.com>
2026-05-15 17:46:20 -07:00
Jedrzej Kosinski fafa9b230d Merge pull request #12 from Comfy-Org/feat/batch-uri-followups
feat(serialization): tighten batch dispatch + accept response-side batch URIs
2026-05-15 13:32:43 -07:00
Jedrzej KosinskiandAmp 35b8a3b17f feat(serialization): tighten batch dispatch + accept response-side batch URIs
Three follow-ups identified in the PR-2 / PR-3 review pass:

1. ``serialization.maybe_externalize`` now keys batch dispatch off
   the ``encoding`` string (``png_base64_batch``) instead of the
   presence of a ``frames`` field. The old ``isinstance(frames, list)
   and frames`` heuristic would silently take the batch branch on a
   malformed envelope that mixed both fields and externalise garbage
   under a non-batch encoding.

2. An empty / missing ``frames`` list on a ``png_base64_batch``
   envelope now raises ``RnpProtocolError(INTERNAL)`` rather than
   silently no-op-ing through to the wire (where the server's
   ``resolve_image_envelope_pngs`` rejects it after the round trip
   anyway). Mirrors the server-side ``make_image_batch_envelope``
   contract: an empty batch is a hard wire bug.

3. ``proxy_node._deserialize_output`` now accepts the externalised
   ``png_base64_batch_uri`` shape on the response path. No server
   emits this shape today (future-proofing only — the symmetric
   counterpart of the request-side externalisation), 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 are fetched in parallel via ``asyncio.gather`` so
   a 9-frame batch import is one round-trip rather than N. Frame
   ordering is preserved (gather returns positional results) — the
   IMAGE tensor's batch axis carries semantic order. Fetched bytes
   are re-stamped as inline ``frames`` so the existing
   ``_decode_image_batch_envelope`` path handles them uniformly.

notes/run_batch_uri_followups.py — new standalone smoke test that
exercises all three changes (encoding-keyed dispatch, empty-frames
hard-error, response-side batch URI parallel fetch + ordering).

Amp-Thread-ID: https://ampcode.com/threads/T-019e2ce6-9403-720c-ae5b-1597b8068be7
Co-authored-by: Amp <amp@ampcode.com>
2026-05-15 12:05:58 -07:00
Jedrzej Kosinski 08981d9a0c Merge pull request #11 from Comfy-Org/fix/local-validate-nested-traversal
fix(local-validate): traverse AUTOGROW templates and DYNAMIC_COMBO branches
2026-05-15 02:38:17 -07:00
KosinkadinkandAmp 6ab23718a4 fix(local-validate): traverse AUTOGROW templates and DYNAMIC_COMBO branches
Pre-PR-4 the descriptor-driven `image_max_batch` cap fired only on
top-level IMAGE inputs (Flux2 `images`, LTXV i2v `image`).
Providers whose cap rides a nested template — GPT Image V2 (16),
Grok ImageEditV2 (3 / 1 by branch), Grok VideoReference (7) — silently
fell through to the server's defense-in-depth path, defeating the
"no paid round trip" guarantee PR-3 promised.

## Client

* `_collect_local_validate` now recurses via the new
  `_rules_from_input_spec` helper. AUTOGROW specs surface their
  template's rules under a private `__template__` key; DYNAMIC_COMBO
  specs surface a `__branches__` map of `{branch_key: {sub_name: rules}}`.
* `_enforce_local_validate` walks `__template__` against each
  Autogrow slot value (`{"image0": <env>, "image1": <env>, ...}`) and
  `__branches__` against the DynamicCombo runtime dict, dispatching
  on `value[<name>]` to pick the active branch's sub-input rules.
* `image_max_batch` checking lifted into a dedicated
  `_check_image_max_batch` helper so the recursive enforcer keeps a
  single rule-application code path.
* Failure messages now carry a `parent.slot[.sub]` label so the
  surfaced `INPUT_INVALID` says
  `input 'model.images.image0' accepts at most 3 image frames…`
  for nested cases instead of just the parent name.

## Docs

* No descriptor or matrix changes — the server already published the
  `local_validate.image_max_batch` rules on the nested templates in
  PR-3 (eat(descriptor): declare image_max_batch local_validate caps,
  comfy-rnp-server#15). The wire shape is unchanged; this PR just
  makes the client honour what the server already declared.

## Verification

* Extended `notes/run_image_max_batch.py` with two new fixtures
  (`_autogrow_descriptor` and `_dyncombo_with_autogrow_descriptor`)
  and 9 new sub-cases covering: AUTOGROW slot over-cap fires,
  AUTOGROW None-slot tolerance, AUTOGROW all-under-cap passes,
  DYNAMIC_COMBO branch-A over-cap fires, DYNAMIC_COMBO branch-A
  under-cap passes, DYNAMIC_COMBO branch-B (cap=1) fires the
  single-frame copy, DYNAMIC_COMBO unknown branch is a defensive
  no-op, plus collected-rules shape assertions.
* End-to-end check against live registered descriptors confirmed all
  four nested-cap providers fire client-side: GrokImageEditNodeV2_RNP
  (`model.images.image0` cap=3 and cap=1 branches),
  GrokVideoReferenceNode_RNP (`model.reference_images.reference0`
  cap=7), OpenAIGPTImageNodeV2_RNP (`model.images.image0` cap=16).

## Lockstep

Client-only fix. Server already publishes the rules; no paired
server PR required. The companion paper trail is comfy-rnp-server#15
(declarative caps) and the parent code-review note from the merged
PR-2/PR-3 wave that flagged this gap as the blocking follow-up.

Amp-Thread-ID: T-019e29b4-d606-7278-8584-52910661c981
Co-authored-by: Amp <amp@ampcode.com>
2026-05-15 02:37:41 -07:00
Jedrzej Kosinski 25b17020ca Merge pull request #10 from Comfy-Org/feat/image-max-batch-rule
feat(local-validate): fail-fast on over-cap image batches
2026-05-15 02:20:37 -07:00
KosinkadinkandAmp 37a4961d6d feat(local-validate): fail-fast on over-cap image batches
Wire ``proxy_node._encode_inputs`` to consult each input's
``descriptor.input.<section>.<name>[1].local_validate.image_max_batch``
rule. When set, the encoder asserts ``tensor.shape[0] <= cap`` for
IMAGE inputs (rank-4 ``[B,H,W,C]`` tensors) and raises
``RnpProtocolError(code=INPUT_INVALID, user_facing=True)`` *before*
any HTTP round trip — so a workflow with N>cap reference frames
surfaces the per-provider error in the standard friendly-message
dialog instead of paying for a doomed upstream call.

## Server

(See companion PR ``Comfy-Org/comfy-rnp-server#15`` —
``feat(descriptor): declare image_max_batch local_validate caps``.)
The server-side per-call caps inside each provider's ``execute()``
stay in place as defense-in-depth; non-compliant clients still fail.

## Client

- ``proxy_node._collect_local_validate(descriptor)`` walks
  ``descriptor.input.{required,optional}`` and returns
  ``{name: rules}`` for every input that carries a non-empty
  ``local_validate`` block. Inputs nested inside ``AUTOGROW`` /
  ``DYNAMIC_COMBO`` aren't surfaced — the parser doesn't yet flatten
  them into top-level inputs, so the cap declared on those templates
  documents the future contract rather than firing today.
- ``proxy_node._enforce_local_validate(name, value, rules,
  *, node_id)`` applies each supported rule to ``value``. Today only
  ``image_max_batch`` is implemented: rank-4 / rank-3-channels-last
  tensors with ``shape[0] > cap`` raise ``RnpProtocolError`` with
  code ``INPUT_INVALID`` and ``user_facing=True``. Friendly per-
  provider message echoes the upstream wording (Flux2's "The
  current maximum number of supported images is N" for cap≥2;
  a "expects a single image frame" copy for cap=1 like LTXV i2v).
- ``_encode_inputs`` runs the validator before the encoder so the
  raise happens before any tensor → PNG conversion or
  ``maybe_externalize`` upload — the cheapest possible fail point.
- Lookup is defensive at every layer: missing ``local_validate`` →
  no-op, missing ``image_max_batch`` sub-key → no-op, non-tensor
  input → no-op, non-image rank → no-op. Older descriptors emit
  no rule and the encoder behaves identically to before.

## Docs

(Matrix prose updated on the server side — the canonical
``notes/remote-nodes-port-matrix.md`` lives in the rnp-server repo.)

## Verification

- ``notes/run_image_max_batch.py`` — new standalone smoke test
  (loads the helper functions via AST extraction so the test runs
  in a bare venv with just ``torch`` installed; doesn't drag in
  the ComfyUI package). Asserts:
  1. ``_collect_local_validate`` extracts the rule from a Flux2-
     shaped descriptor.
  2. A 3-frame tensor against ``cap=2`` raises ``RnpProtocolError``
     with ``code=INPUT_INVALID``, ``user_facing=True``, and the
     ``node_id`` + counts mentioned in the message.
  3. A 2-frame tensor against ``cap=2`` encodes cleanly.
  4. ``cap=1`` surfaces the single-frame copy.
  5. Descriptor without ``local_validate`` is a no-op.
  6. ``local_validate`` block missing the ``image_max_batch`` key
     is a no-op.

  Run with: ``python notes/run_image_max_batch.py`` — all
  assertions pass on PASS.

## Lockstep

Paired with server PR ``Comfy-Org/comfy-rnp-server#15``
(``feat/image-max-batch-rule``). Both branches were cut off
``origin/main`` before PR-2 lands; the orchestrator will rebase PR-3
onto PR-2 before merging.

Co-authored-by: Amp <amp@ampcode.com>
Amp-Thread-ID: https://ampcode.com/threads/T-019e2adb-6ddf-72ad-88e1-8b30c57e873f
2026-05-15 02:19:59 -07:00
Jedrzej Kosinski 3dbb522771 Merge pull request #9 from Comfy-Org/feat/batch-envelope-externalization
feat(serialization): externalize oversize image batch envelopes via per-frame presigned URIs
2026-05-15 02:19:45 -07:00
KosinkadinkandAmp 6433a46c93 feat(serialization): externalize oversize image batch envelopes via per-frame presigned URIs
The pre-existing `serialization.maybe_externalize` only inspected the
singleton `data` field, so a multi-frame `frames=[…]` IMAGE envelope
whose summed base64 payload crossed the 8 MiB
`DEFAULT_MAX_INLINE_PAYLOAD_BYTES` cap silently rode the wire as a
30+ MiB POST body — defeating the entire reason the cap exists. A
9× 2048² Flux2 reference batch hits ~200 MB base64 inline; that's
the worst-case the new path is designed for.

This commit adds the **client side** of the lockstep fix: detect
oversize multi-frame batches (sum of per-frame raw bytes), upload
each frame as an individual presigned PUT in parallel via
`asyncio.gather`, and rewrite the envelope to the new
`image:png_base64_batch_uri` encoding with a parallel `uris` list.
Server side closes the loop with a matching encoding-string
dispatch in `resolve_image_envelope_pngs`.

Per-frame URIs (NOT a tar/zip blob) compose with the existing
per-frame fetch helpers and let the server fetch in parallel.

## Server
- See lockstep PR on `Comfy-Org/comfy-rnp-server` branch
  `feat/batch-envelope-externalization` for the
  `Capability.IMAGE_PNG_BASE64_BATCH_URI` token, the
  `_resolve_png_base64_batch_uri` per-frame fetch helper, and the
  encoding-string dispatch in `resolve_image_envelope_pngs`.

## Client
- `protocol.py`: mirror new `Capability.IMAGE_PNG_BASE64_BATCH_URI =
  "image:png_base64_batch_uri"` token next to
  `IMAGE_PNG_BASE64_BATCH`. Same comment block as the server-side
  copy so the two stay byte-comparable.
- `client.py`: append `Capability.IMAGE_PNG_BASE64_BATCH_URI` to
  `CLIENT_CAPABILITIES` so the server knows this client can emit
  the externalised batch shape. Comment notes the opportunistic
  posture (any input that already accepts `png_base64_batch`
  transparently accepts the externalised variant — no descriptor
  opt-in needed).
- `serialization.py`:
  - `maybe_externalize` now branches on payload shape: if
    `envelope["frames"]` is a populated list, dispatch to the new
    `_maybe_externalize_batch` helper; otherwise fall through to
    the existing singleton-`data` path. Docstring expanded to
    describe both shapes and the encoding-flip-on-spill behaviour.
  - New `_maybe_externalize_batch` helper sums per-frame raw bytes
    (`len(b64) * 3 // 4`), short-circuits when the batch fits under
    the cap, and otherwise uploads every frame in parallel via
    `asyncio.gather` (preserves frame ordering — critical because
    Flux2 maps `uris[k]` → `input_image_{k+1}` 1:1) before
    rewriting the envelope to `{type, encoding:
    png_base64_batch_uri, uris: [...], shape, dtype, color_space}`
    with `frames` stripped.
  - `import asyncio` added at module top so `asyncio.gather` is
    available without per-call imports.

## Docs
- The pilot side-by-side workflow `notes/sxs_flux2_pro_multi_ref_xl.json`
  (9× 2048² references) and the closed-gap entry in
  `notes/remote-nodes-vocabulary-gaps.md` live in the server
  worktree (this client repo has no `notes/` directory). Both ship
  in the paired server PR.

## Verification
- AST parse of `protocol.py`, `serialization.py`, `client.py`
  succeeds.
- In-process round-trip test (with a fake `comfy_api_nodes.util.client`
  shim and a stubbed `upload_asset`):
  * Under-cap batch envelope (1× 5 MiB raw, cap=8 MiB) → passes
    through unchanged (object identity preserved).
  * Over-cap batch envelope (3× 5 MiB raw, cap=8 MiB) → triggers
    3 PUTs via the stub, returns
    `{type: image, encoding: png_base64_batch_uri,
    uris: ['…/asset/1','…/asset/2','…/asset/3'], shape: [3,1,1,3],
    dtype: uint8, color_space: srgb}` with `frames` stripped.
  * Singleton `data` envelope (10 MiB raw, cap=8 MiB) → still gets
    the legacy `data → uri` swap (existing path unaffected).
  * `Capability.IMAGE_PNG_BASE64_BATCH_URI in CLIENT_CAPABILITIES`
    holds.

## Lockstep
- Paired server PR: https://github.com/Comfy-Org/comfy-rnp-server/pull/14
  (`feat/batch-envelope-externalization`). Server PR was opened
  first because the wire shape needs a server resolver to be
  meaningful — the orchestrator merges them in lockstep order.
- Capability gating posture: `IMAGE_PNG_BASE64_BATCH_URI` is
  opportunistic — this client emits it whenever a batch envelope it
  produced overflows the inline cap, regardless of whether the
  descriptor's `input_serialization` lists it. Server accepts on
  any image input that already accepts `png_base64_batch` via the
  shared `resolve_image_envelope_pngs` dispatch.
- Output-side decoder: server providers emit `png_base64_batch` (or
  single-frame `png_base64`) on responses today; the URI shape is
  consumed on input only, so the client decoder (`_DECODERS` /
  `_decode_image_batch_envelope`) is unchanged. The existing
  `decode_image_envelope` already dispatches on the `encoding`
  string (not field presence), so adding a `png_base64_batch_uri`
  decode branch is a drop-in if a future provider starts emitting
  the URI shape on outputs. Logged as a known follow-up.

Amp-Thread-ID: T-019e29b4-d606-7278-8584-52910661c981
Co-authored-by: Amp <amp@ampcode.com>
Amp-Thread-ID: https://ampcode.com/threads/T-019e2adb-6ddf-72ad-88e1-8b30c57e873f
Co-authored-by: Amp <amp@ampcode.com>
2026-05-15 02:12:23 -07:00
Jedrzej Kosinski c5ba3cb69d feat(capabilities): mirror missing audio/video/svg tokens (#8)
feat(capabilities): mirror missing audio/video/svg tokens, advertise round-trippable encodings
2026-05-15 01:57:12 -07:00
KosinkadinkandAmp 1c8d341918 feat(capabilities): mirror missing audio/video/svg tokens, advertise round-trippable encodings
Pre-F1 immediate-follow-up #1 (client side): the client ``Capability``
mirror in ``protocol.py`` had drifted away from the server's
``comfy_rnp_protocol/constants.py`` — the audio sweep that landed
``AUDIO_MP3_BASE64`` / ``AUDIO_WAV_BASE64`` / ``AUDIO_WAV_INLINE``,
the video sweep that landed ``VIDEO_MP4_BASE64``, and the SVG
encoding ``SVG_XML_BASE64`` were declared on the server but never
mirrored here. The mirror is supposed to be byte-equivalent so a
client that imports ``Capability.AUDIO_MP3_BASE64`` doesn't crash
when reading the same name a server descriptor emits.

This PR adds the five missing tokens to the client mirror and
expands ``CLIENT_CAPABILITIES`` to advertise the encodings the
client actually round-trips today (audio MP3 / WAV decode, audio
MP3 encode via ``encode_audio_input``, video MP4 base64 encode via
``encode_video_input``). ``SVG_XML_BASE64`` lands in the mirror but
is **not** added to ``CLIENT_CAPABILITIES`` yet — there is no SVG
envelope decoder in ``serialization.py`` today (Recraft SVG
providers currently only round-trip via the ``IO.Custom`` opaque
fallthrough Track §B added; a real ``decode_svg_envelope`` is a
follow-up).

Pairs with the matching ``feat/capability-mirror-sync`` PR on
``Comfy-Org/comfy-rnp-server`` (server-side capability gates: every
descriptor that publishes ``"opaque"`` or ``"png_base64_batch"`` in
its ``input_serialization`` now requires the matching token, so a
client that doesn't advertise ``IO_OPAQUE`` /
``IMAGE_PNG_BASE64_BATCH`` is filtered out at descriptor-publish
time); both PRs must merge for the gates to be useful end-to-end.

## Client

- ``protocol.py``: add ``VIDEO_MP4_BASE64``, ``AUDIO_MP3_BASE64``,
  ``AUDIO_WAV_BASE64``, ``AUDIO_WAV_INLINE``, ``SVG_XML_BASE64`` to
  the ``Capability`` mirror class (byte-equivalent with server).
- ``client.py``: add ``VIDEO_MP4_BASE64``, ``AUDIO_MP3_INLINE``,
  ``AUDIO_MP3_BASE64``, ``AUDIO_WAV_BASE64``, ``AUDIO_WAV_INLINE``
  to ``CLIENT_CAPABILITIES`` so servers gating audio / video
  descriptors on these tokens hand them to this client.
  ``AUDIO_MP3_INLINE`` was implicitly supported by
  ``decode_audio_envelope`` but had never been advertised; now it
  is.

## Verification

- ``serialization.encode_audio_input`` already emits
  ``mp3_base64`` envelopes (Stability audio sweep) — the client
  CAN produce this encoding, so advertising it is honest.
- ``serialization.decode_audio_envelope`` already accepts
  ``("mp3_base64", "mp3_inline", "wav_base64", "wav_inline")`` —
  the client CAN consume all four, so advertising them is honest.
- ``serialization.decode_video_envelope`` already accepts
  ``("mp4_inline", "mp4_base64")`` — both already advertised.
- ``serialization.decode_envelope`` does not currently handle
  ``"svg"`` envelopes (no entry in ``_DECODERS``) so
  ``SVG_XML_BASE64`` is mirrored for byte-equivalence but
  intentionally NOT advertised.

## Lockstep

- Client PR: this PR.
- Server PR: ``feat/capability-mirror-sync`` on
  ``Comfy-Org/comfy-rnp-server`` (descriptor capability gates).
- Both must merge; descriptor / advertisement-only change.

Co-authored-by: Amp <amp@ampcode.com>
Amp-Thread-ID: https://ampcode.com/threads/T-019e29b4-d606-7278-8584-52910661c981
2026-05-15 01:55:52 -07:00
Jedrzej Kosinski 6a82114ded feat(serialization): input-side image:png_base64_batch envelope (#7)
feat: input-side image:png_base64_batch envelope
2026-05-14 21:34:30 -07:00
KosinkadinkandAmp aad9f0ba43 feat(serialization): input-side image:png_base64_batch envelope
Closes the silent `tensor[0]` truncation bug for multi-image inputs:
when a descriptor advertises `"png_base64_batch"` for an IMAGE input
and the inbound tensor has `shape[0] > 1`, the encoder now emits a
multi-frame envelope (`frames=[…]`) instead of dropping `tensor[1:]`.
Lockstep server change ships in `comfy-rnp-server` PR (paired branch
`feat/input-batch-envelope`).

## Client
- `serialization.py`: extend `encode_image_tensor` with an
  `accepts_batch=False` keyword. When True AND `tensor.shape[0] > 1`,
  emit `{"type":"image","encoding":"png_base64_batch","frames":[…],
  "shape":[B,H,W,C],…}` — the same wire shape the existing
  `_decode_image_batch_envelope` consumes on the response path, so
  both directions round-trip symmetrically. Default-False keeps the
  legacy single-frame contract for descriptors that haven't migrated.
- `proxy_node.py`: plumb the per-input `input_serialization`
  allow-list from `descriptor.remote.input_serialization` through
  `build_node_class` → `_execute_remote` → `_encode_inputs` →
  `_encode_one`. The wire shape `{input_name: <str | list[str]>}`
  normalizes to `dict[str, list[str]]` once at descriptor parse time
  so the encoder can do a single `in` check per input. `_encode_one`
  passes `accepts_batch=("png_base64_batch" in encodings)` to the
  image encoder.
- `IMAGE_PNG_BASE64_BATCH` is already in `client.CLIENT_CAPABILITIES`
  (was added when the output-side decoder shipped) so no wire-header
  change.

## Verification
- Both modified files parse cleanly via `ast.parse`.
- `serialization._decode_image_batch_envelope` (output-side decoder
  already in this file) consumes the same `frames=[…]` shape the
  new encoder emits — round-trips symmetrically.

## Lockstep
- Paired server PR: `comfy-rnp-server` branch
  `feat/input-batch-envelope` — adds `"png_base64_batch"` to
  `DEFAULT_INPUT_SERIALIZATIONS["image"]`, the shared
  `comfy_rnp_protocol.envelopes.resolve_image_envelope_pngs(envelope)`
  helper, and migrates Flux2Pro/Max, OpenAI ChatNode + GPTImageV2,
  Grok ImageEditV2 + VideoReference, and LTXV ImageToVideo
  providers to consume the multi-frame envelope. Both PRs must
  merge together; no back-compat between unmatched versions.

Co-authored-by: Amp <amp@ampcode.com>
Amp-Thread-ID: https://ampcode.com/threads/T-019e29b4-d606-7278-8584-52910661c981
2026-05-14 21:22:30 -07:00
Jedrzej Kosinski 93ba8cba48 Merge pull request #6 from Comfy-Org/feat/opaque-custom-io
feat: opaque pass-through custom IO
2026-05-14 21:02:50 -07:00
KosinkadinkandAmp d67d42cd59 feat(custom-io): opaque pass-through custom IO bucket
Track B (opaque-custom-io) end-to-end on the proxy_node side. Adds a
generic ``IO.Custom(io_type)`` fallthrough in the parser so any
io_type string the client doesn't explicitly recognise becomes an
opaque socket with the original io_type preserved (so node-to-node
connections only chain when the strings match — an
``OPENAI_INPUT_FILES`` output cannot connect to a
``GEMINI_INPUT_FILES`` input). Pairs with the matching
``feat/opaque-custom-io`` PR on ``Comfy-Org/comfy-rnp-server`` (new
``Capability.IO_OPAQUE`` + the four deferred OpenAI / Gemini helper
node providers); both PRs must merge for the new path to light up
end-to-end.

## Client

- ``protocol.py``: mirror ``Capability.IO_OPAQUE = "io:opaque"``
  from the server's ``comfy_rnp_protocol/constants.py``.
- ``client.py``: advertise ``Capability.IO_OPAQUE`` in
  ``CLIENT_CAPABILITIES`` so servers gating descriptors on the
  token (every helper-config descriptor declares
  ``required_capabilities: [Capability.IO_OPAQUE]``) hand them to
  this client.
- ``proxy_node._parse_input_spec``: after the explicit STRING / INT
  / FLOAT / BOOLEAN / COMBO / IMAGE / VIDEO / AUDIO / MASK branches,
  any non-empty string io_type falls through to
  ``IO.Custom(io_type).Input(name, optional=…, tooltip=…)``. The
  V3 ``@comfytype`` decorator binds the original io_type string to
  the synthesised ComfyType, so the frontend's connection-validity
  check enforces strict string matching between sockets.
- ``proxy_node._parse_outputs``: mirror branch — when
  ``_OUTPUT_CLASSES.get(io_type)`` is None, build
  ``IO.Custom(io_type).Output(...)`` instead of failing the whole
  descriptor. Only a non-string / empty io_type still warns and
  skips.
- Encoder dispatch: opaque values flow through ``_encode_one``
  unchanged (already JSON-serialisable dicts/lists/strings — no
  envelope wrapping). Verified against the existing fallthrough at
  the end of ``_encode_one`` (returns ``value`` for non-tensor /
  non-audio inputs). Output deserialisation already passes
  non-envelope values through unchanged in
  ``_deserialize_output``.

## Docs

The proxy_node module-level docstring is updated to reflect the
new fallthrough (previously documented only the strict-skip
behaviour for unknown io_types).

## Verification

- Static AST parse of all four touched modules
  (``proxy_node.py``, ``client.py``, ``serialization.py``,
  ``protocol.py``).
- ``Capability.IO_OPAQUE`` resolves to ``"io:opaque"`` (matches the
  server's wire shape).
- ``IO.Custom("OPENAI_INPUT_FILES").Input("files", optional=True)``
  and ``.Output()`` both expose ``io_type ==
  "OPENAI_INPUT_FILES"``; ``IO.Custom("GEMINI_INPUT_FILES")``
  similarly exposes its own io_type, so the two sockets cannot
  cross-chain.

## Lockstep

Pairs with ``feat/opaque-custom-io`` on
``Comfy-Org/comfy-rnp-server`` (the new ``Capability.IO_OPAQUE``
constant lives there as the source of truth; this repo's
``protocol.py`` is the byte-compatible mirror). The server PR also
ports the four deferred helper nodes that exercise the new path
end-to-end (``OpenAIInputFiles_RNP``, ``OpenAIChatConfig_RNP``,
``GeminiInputFiles_RNP``, plus the chat node's ``files=`` /
``advanced_options=`` inputs). Until that PR merges, this client
parser fallthrough lights up only against the existing Recraft
custom IO descriptors (``RECRAFT_COLOR``, ``RECRAFT_CONTROLS``,
``RECRAFT_V3_STYLE``) — which previously caused the whole
descriptor to be skipped because the parser had no fallthrough.

Co-authored-by: Amp <amp@ampcode.com>
Amp-Thread-ID: https://ampcode.com/threads/T-019e29b4-d606-7278-8584-52910661c981
2026-05-14 21:00:56 -07:00
Jedrzej Kosinski d450bf3769 Merge pull request #5 from Comfy-Org/feat/async-polling-mode
feat(async-polling): advertise ASYNC_EXECUTE capability and derive poll cap from duration
2026-05-14 19:52:01 -07:00
KosinkadinkandAmp 94f995bb05 feat(async-polling): advertise ASYNC_EXECUTE capability and derive poll cap from duration
Final pieces of Track A (async-polling mode) on the client side. The
proxy_node already routes between ``/execute`` and ``/execute_async``
on the descriptor's ``remote.execution.mode``; this commit makes the
contract honest in two ways:

- ``client.py``: advertise the new ``Capability.ASYNC_EXECUTE`` token
  so the server can gate ``async_polling`` descriptors on its
  presence. An older client that only knows ``/execute`` is now
  visible to the server and won't be silently routed to a long-poll
  provider whose lifecycle it can't drive.
- ``protocol.py`` + ``proxy_node.py``: drop the
  ``DEFAULT_MAX_POLL_ATTEMPTS`` wire-default in favour of
  duration-based ``DEFAULT_SOFT_TIMEOUT_S`` /
  ``DEFAULT_HARD_TIMEOUT_S`` / ``DEFAULT_POLL_GRACE_S``. The
  poll-loop cap is now a *derived* value
  (``ceil((hard_timeout_s + grace) / poll_interval_s)``) so changing
  the poll cadence can never silently shrink or stretch the contract.
  Industry precedent: OAuth2 device flow §3.5, Kubernetes
  ``--timeout``, gRPC deadlines all express bounds as durations.

Lockstep with the server side of Track A
(comfy-rnp-server feat/async-polling-mode), which now enforces
``mode`` dispatch on ``/execute`` vs ``/execute_async`` and migrates
existing providers (BFL/Bria/LTXV/Grok video/OpenAIChatNode/Stability
UpscaleCreative → ``async_polling``; DALL-E/Ideogram/Recraft/sync
Stability → ``request_response``).

Co-authored-by: Amp <amp@ampcode.com>
Amp-Thread-ID: https://ampcode.com/threads/T-019e28eb-31ae-76a1-bb45-c0922f43bcc3
2026-05-14 19:50:47 -07:00
Jedrzej Kosinski 8835c4ee02 Merge pull request #4 from Comfy-Org/feat/batch-image-envelope
feat(serialization): decode image:png_base64_batch envelope
2026-05-13 22:01:34 -07:00
Jedrzej KosinskiandAmp ed63c5ae49 feat(serialization): decode image:png_base64_batch envelope
Mirrors the server-side change in comfy-rnp-server PR #7. Adds the
client-side half of the multi-image envelope contract:

- Capability.IMAGE_PNG_BASE64_BATCH (\image:png_base64_batch\) added
  to protocol.Capability, mirroring comfy_rnp_protocol/constants.py.
- decode_image_envelope() now dispatches png_base64_batch to a new
  _decode_image_batch_envelope() helper that stacks each PNG frame
  into a [B, H, W, C] tensor — the native ComfyUI image-batch shape
  — so providers whose upstream returns n>1 surface as a real
  batched tensor in the workflow runtime instead of being silently
  truncated to the first frame.
- Frame-count integrity check: if the envelope declares shape[0]=B
  but \rames\ has a different length, fail hard rather than
  drop frames silently. Matches the SVG decoder's \count\ check.
- Per-frame H/W mismatch is also a hard error — the server-side
  encoder enforces uniform sizing already, so a mismatch on the
  wire indicates a wire bug worth surfacing.
- IMAGE_PNG_BASE64_BATCH added to client.CLIENT_CAPABILITIES so the
  server's per-request negotiator (\client_advertises_capability\
  in comfy_rnp_protocol.capabilities) lights up the batch path.

Single-frame envelopes (\png_base64\ / \jpeg_base64\ /
\webp_base64\) remain unchanged — back-compat with every
already-merged provider.

Verified end-to-end with a synthetic 2-frame envelope (white +
black 32x64 PNGs) plus the mismatch / empty-frames error paths.

Co-authored-by: Amp <amp@ampcode.com>
Amp-Thread-ID: https://ampcode.com/threads/T-019e24d1-250c-750a-ab00-2eb5e9cb2500
2026-05-13 22:00:52 -07:00
10 changed files with 4045 additions and 76 deletions
+68 -2
View File
@@ -30,13 +30,79 @@ 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) need, plus the ``schema:v3`` marker so descriptors come back
# in the V3 ``get_v1_info()`` shape.
# Gemini, Ideogram, ...) need, plus the ``schema:v3`` marker so
# descriptors come back in the V3 ``get_v1_info()`` shape. Servers use
# this list to negotiate per-request: e.g. a provider only emits the
# multi-frame ``image:png_base64_batch`` envelope when this list
# contains :data:`Capability.IMAGE_PNG_BASE64_BATCH`, otherwise it
# falls back to the single-frame ``image:png_base64`` shape (with a
# server-side truncation warning logged when extras are dropped).
CLIENT_CAPABILITIES = [
Capability.SCHEMA_V3,
Capability.VIDEO_MP4_INLINE,
Capability.VIDEO_MP4_BASE64,
Capability.IMAGE_PNG_BASE64,
Capability.IMAGE_PNG_BASE64_BATCH,
# Advertises that this client can emit a multi-frame IMAGE
# envelope with per-frame presigned-GET URIs instead of inline
# base64 frames when the summed batch payload would overflow the
# 8 MiB inline cap (see ``serialization.maybe_externalize``).
# Opportunistic: the client falls back to it whenever a batch
# envelope it produced exceeds the cap, so descriptors don't
# need to publish it in their ``input_serialization`` to
# benefit — any input that already accepts ``png_base64_batch``
# transparently accepts the externalised variant on the server
# via the shared ``resolve_image_envelope_pngs`` dispatch.
Capability.IMAGE_PNG_BASE64_BATCH_URI,
Capability.MASK_PNG_BASE64,
Capability.AUDIO_MP3_INLINE,
Capability.AUDIO_MP3_BASE64,
Capability.AUDIO_WAV_BASE64,
Capability.AUDIO_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,
]
# Reported in the X-RNP-Client-Version header.
+352
View File
@@ -0,0 +1,352 @@
"""Smoke test for the three batch-URI followups (PR-2/PR-3 cleanup):
1. ``serialization.maybe_externalize`` keys batch dispatch off the
``encoding`` string (``png_base64_batch``) — NOT field presence
(``frames``).
2. ``serialization.maybe_externalize`` rejects an empty/missing
``frames`` list on a ``png_base64_batch`` envelope as a hard
``RnpProtocolError`` rather than silently no-op-ing.
3. ``proxy_node._deserialize_output`` accepts the externalised
``png_base64_batch_uri`` shape on the response path: fetches every
``uri`` (in parallel via ``asyncio.gather``), rewrites to inline
``frames``, and decodes into a ``[B,H,W,C]`` tensor with frame
ordering preserved.
Run with any python that has ``torch`` and ``Pillow`` available:
python notes/run_batch_uri_followups.py
"""
from __future__ import annotations
import asyncio
import base64
import http.server
import importlib.util
import io
import os
import socketserver
import sys
import threading
import time
import types
import numpy as np
import torch
from PIL import Image
# ---------------------------------------------------------------------------
# Locate the worktree under test.
# ---------------------------------------------------------------------------
_HERE = os.path.dirname(os.path.abspath(__file__))
_CLIENT_DIR = os.path.dirname(_HERE)
def _load_module(name: str, path: str) -> types.ModuleType:
spec = importlib.util.spec_from_file_location(name, path)
mod = importlib.util.module_from_spec(spec)
sys.modules[name] = mod
spec.loader.exec_module(mod)
return mod
# Stub out client.upload_asset so the dispatch tests don't need a real
# RNP server — the externalize batch path will call this once per frame.
_uploads: list[bytes] = []
async def _fake_upload_asset(server_url, raw, *, content_type, auth_headers=None): # noqa: ANN001
_uploads.append(bytes(raw))
return f"https://fake-storage/{len(_uploads) - 1}"
def _install_fake_client() -> None:
fake_client = types.ModuleType("comfy_remote_nodes.client")
fake_client.upload_asset = _fake_upload_asset
sys.modules["comfy_remote_nodes.client"] = fake_client
def _png_b64_for_color(color: tuple[int, int, int]) -> str:
img = Image.new("RGB", (4, 4), color=color)
buf = io.BytesIO()
img.save(buf, format="PNG")
return base64.b64encode(buf.getvalue()).decode("ascii")
def _png_bytes_for_color(color: tuple[int, int, int]) -> bytes:
img = Image.new("RGB", (4, 4), color=color)
buf = io.BytesIO()
img.save(buf, format="PNG")
return buf.getvalue()
# ---------------------------------------------------------------------------
# Test 1+2: maybe_externalize dispatch + empty-frames
# ---------------------------------------------------------------------------
async def _test_maybe_externalize() -> list[str]:
failures: list[str] = []
# Re-import serialization fresh because it caches a `client` ref.
sys.modules.pop("comfy_remote_nodes", None)
sys.modules.pop("comfy_remote_nodes.serialization", None)
sys.modules.pop("comfy_remote_nodes.protocol", None)
# Build a minimal package shim so `from . import client` works.
pkg = types.ModuleType("comfy_remote_nodes")
pkg.__path__ = [_CLIENT_DIR]
sys.modules["comfy_remote_nodes"] = pkg
_install_fake_client()
protocol = _load_module(
"comfy_remote_nodes.protocol",
os.path.join(_CLIENT_DIR, "protocol.py"),
)
serialization = _load_module(
"comfy_remote_nodes.serialization",
os.path.join(_CLIENT_DIR, "serialization.py"),
)
# --- Case A: batch envelope with payload bigger than cap → externalises
_uploads.clear()
big_frames = [_png_b64_for_color((i * 30, 0, 0)) for i in range(4)]
env = {
"type": "image",
"encoding": "png_base64_batch",
"frames": big_frames,
"shape": [4, 4, 4, 3],
"dtype": "uint8",
"color_space": "srgb",
}
out = await serialization.maybe_externalize(
env, server_url="http://fake", max_inline_bytes=1, auth_headers={},
)
if out.get("encoding") != "png_base64_batch_uri":
failures.append(
f"A: expected encoding=png_base64_batch_uri, got {out.get('encoding')!r}"
)
if "frames" in out:
failures.append("A: 'frames' field still present after externalise")
if not isinstance(out.get("uris"), list) or len(out["uris"]) != 4:
failures.append(f"A: expected 4 uris, got {out.get('uris')!r}")
if len(_uploads) != 4:
failures.append(f"A: expected 4 uploads, got {len(_uploads)}")
# --- Case B: dispatch keys off encoding, NOT field presence.
# An envelope with a stray `frames` field but a non-batch encoding
# must NOT be treated as a batch (would have under the old
# `isinstance(frames, list) and frames` dispatch).
_uploads.clear()
rogue = {
"type": "image",
"encoding": "png_base64", # singleton encoding
"data": _png_b64_for_color((10, 20, 30)) * 200, # > cap
"frames": ["bogus-extra-field"],
}
out_b = await serialization.maybe_externalize(
rogue, server_url="http://fake", max_inline_bytes=1, auth_headers={},
)
if out_b.get("encoding") == "png_base64_batch_uri":
failures.append(
"B: dispatch took batch branch on a singleton encoding (rogue frames field)"
)
if "uri" not in out_b:
failures.append("B: singleton encoding should have externalised to 'uri'")
if len(_uploads) != 1:
failures.append(f"B: expected 1 upload (singleton path), got {len(_uploads)}")
# --- Case C: empty frames on a batch encoding is a hard error.
bad = {
"type": "image",
"encoding": "png_base64_batch",
"frames": [],
"shape": [0, 4, 4, 3],
}
try:
await serialization.maybe_externalize(
bad, server_url="http://fake", max_inline_bytes=1, auth_headers={},
)
except protocol.RnpProtocolError:
pass
else:
failures.append("C: empty 'frames' on batch encoding did not raise RnpProtocolError")
# --- Case D: under-cap batch: no upload, envelope unchanged.
_uploads.clear()
small = {
"type": "image",
"encoding": "png_base64_batch",
"frames": big_frames,
"shape": [4, 4, 4, 3],
}
out_d = await serialization.maybe_externalize(
small,
server_url="http://fake",
max_inline_bytes=10 * 1024 * 1024,
auth_headers={},
)
if out_d is not small and out_d != small:
failures.append("D: under-cap batch should be a no-op")
if _uploads:
failures.append(f"D: under-cap batch should not upload, got {len(_uploads)}")
return failures
# ---------------------------------------------------------------------------
# Test 3: _deserialize_output handles png_base64_batch_uri on response
# ---------------------------------------------------------------------------
DELAY_S = 0.4
N_FRAMES = 4
class _SlowPngHandler(http.server.BaseHTTPRequestHandler):
def do_GET(self) -> None: # noqa: N802
time.sleep(DELAY_S)
try:
idx = int(self.path.rsplit("/", 1)[-1])
except ValueError:
self.send_response(404); self.end_headers(); return
# Distinct color per frame so we can verify ordering after decode.
body = _png_bytes_for_color((idx * 60 % 256, 80, 120))
self.send_response(200)
self.send_header("Content-Type", "image/png")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def log_message(self, *_a, **_kw) -> None:
return
async def _test_deserialize_output() -> list[str]:
failures: list[str] = []
server = socketserver.ThreadingTCPServer(("127.0.0.1", 0), _SlowPngHandler)
host, port = server.server_address
th = threading.Thread(target=server.serve_forever, daemon=True)
th.start()
try:
# Stub the proxy_node's _fetch_url_envelope so we don't have to
# bring up the whole comfy_api_nodes.util.client stack.
import urllib.request as _urlreq
async def fake_fetch(envelope, policy, *, auth_headers, node_cls): # noqa: ANN001, ARG001
# Real ``_fetch_url_envelope`` is properly async (aiohttp);
# the smoke stub uses sync urllib so we hop to a thread to
# preserve the parallel-gather semantics under test.
uri = envelope["uri"]
def _do_fetch() -> bytes:
with _urlreq.urlopen(uri, timeout=10.0) as resp:
return resp.read()
return await asyncio.to_thread(_do_fetch)
# Re-import proxy_node helpers via AST to avoid loading ComfyUI.
import ast
with open(os.path.join(_CLIENT_DIR, "proxy_node.py"), "r", encoding="utf-8") as fh:
src = fh.read()
tree = ast.parse(src)
wanted = {"_deserialize_output"}
nodes = [n for n in tree.body if isinstance(n, ast.AsyncFunctionDef) and n.name in wanted]
if len(nodes) != len(wanted):
failures.append(f"missing AST nodes; found {[n.name for n in nodes]}")
return failures
# Re-import the protocol + serialization modules with the package
# shim so `from . import` works inside serialization.
sys.modules.pop("comfy_remote_nodes.serialization", None)
sys.modules.pop("comfy_remote_nodes.protocol", None)
pkg = types.ModuleType("comfy_remote_nodes")
pkg.__path__ = [_CLIENT_DIR]
sys.modules["comfy_remote_nodes"] = pkg
_install_fake_client()
protocol = _load_module(
"comfy_remote_nodes.protocol",
os.path.join(_CLIENT_DIR, "protocol.py"),
)
serialization = _load_module(
"comfy_remote_nodes.serialization",
os.path.join(_CLIENT_DIR, "serialization.py"),
)
ns: dict = {
"Any": object,
"RnpProtocolError": protocol.RnpProtocolError,
"ErrorCode": protocol.ErrorCode,
"is_envelope": protocol.is_envelope,
"serialization": serialization,
"_fetch_url_envelope": fake_fetch,
"log": types.SimpleNamespace(warning=lambda *a, **kw: None),
}
mod_src = ast.Module(body=nodes, type_ignores=[])
exec(compile(mod_src, "<batch_uri_smoke>", "exec"), ns)
deserialize = ns["_deserialize_output"]
# Build the externalised batch envelope on the response path.
uris = [f"http://{host}:{port}/frame/{i}" for i in range(N_FRAMES)]
envelope = {
"type": "image",
"encoding": "png_base64_batch_uri",
"uris": uris,
"shape": [N_FRAMES, 4, 4, 3],
"dtype": "uint8",
"color_space": "srgb",
}
url_fetch_policy = {"image": {"url_kind": "presigned", "timeout_s": 10}}
t0 = time.monotonic()
tensor = await deserialize(
envelope, meta=None,
url_fetch_policy=url_fetch_policy,
server_url="http://fake",
auth_headers={},
node_cls=None,
)
elapsed = time.monotonic() - t0
if not isinstance(tensor, torch.Tensor):
failures.append(f"expected torch.Tensor, got {type(tensor).__name__}")
return failures
if tuple(tensor.shape) != (N_FRAMES, 4, 4, 3):
failures.append(f"shape {tuple(tensor.shape)} != ({N_FRAMES}, 4, 4, 3)")
# Each frame's R-channel should match the expected color.
for i in range(N_FRAMES):
expected_r = (i * 60 % 256) / 255.0
actual_r = float(tensor[i, 0, 0, 0])
if abs(actual_r - expected_r) > 1e-3:
failures.append(
f"frame {i}: R channel {actual_r:.3f} != expected {expected_r:.3f} "
f"(ordering broken or wrong fetch)"
)
# Wall-clock: parallel should be ~DELAY_S, sequential would be ~N*DELAY_S.
sequential_lower = N_FRAMES * DELAY_S * 0.7
if elapsed >= sequential_lower:
failures.append(
f"wall clock {elapsed:.2f}s >= sequential lower bound "
f"{sequential_lower:.2f}s — gather not running in parallel"
)
return failures
finally:
server.shutdown(); server.server_close()
def main() -> int:
failures: list[str] = []
failures += asyncio.run(_test_maybe_externalize())
failures += asyncio.run(_test_deserialize_output())
if failures:
print()
print("FAIL:")
for f in failures:
print(" -", f)
return 1
print("PASS: maybe_externalize dispatches on encoding + rejects empty frames; "
"_deserialize_output handles png_base64_batch_uri in parallel")
return 0
if __name__ == "__main__":
sys.exit(main())
+526
View File
@@ -0,0 +1,526 @@
"""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.")
+360
View File
@@ -0,0 +1,360 @@
"""Smoke test for nested dict recursion in ``_encode_one`` /
``_encode_inputs`` / ``_externalize_nested``.
Background: ComfyUI's V3 ``build_nested_inputs`` rebuilds AUTOGROW and
DYNAMIC_COMBO runtime kwargs into nested dicts (e.g. ``{"reference_images":
{"image1": <tensor>, "image2": <tensor>}}`` for a top-level AG, or
``{"model": {"branch": "wan2.7-r2v", "video_refs": {"video1":
<VideoInput>}}}`` for AG-inside-DC). Before this fix, ``_encode_one``
returned a non-tensor non-VideoInput non-audio dict unchanged, which
meant per-slot tensors / VideoInputs survived past encoding and would
either crash at ``json.dumps`` time or violate the server's contract
(comfy_rnp_server's ``_ordered_wan_videoedit_image_envelopes`` etc.
expect each slot value to already be an RNP envelope).
Asserts:
1. Top-level AG-of-IMAGE dict: each ``image<N>`` tensor becomes an
image envelope; the wrapping dict structure is preserved.
2. Top-level AG-of-VIDEO dict: each ``video<N>`` VideoInput becomes a
video envelope.
3. DC-wrapping-AG: top-level dict carries a branch key (string) AND a
nested AG dict of tensors / VideoInputs; encoder recurses one level
deeper and encodes the leaves while leaving the branch key untouched.
4. AUDIO input is NOT swallowed by the dict-recursion branch: the
``{"waveform": ..., "sample_rate": ...}`` shape still routes to
``encode_audio_input``.
5. Already-encoded envelope dicts pass through unchanged (no double-
encoding) — protects against re-running the encoder on a value that
the caller pre-encoded.
6. Plain scalar config dicts (``{"width": 1024, "height": 576}``) pass
through unchanged.
7. ``_externalize_nested`` walks the same shape: a nested AG dict of
oversize envelopes uploads each envelope via the stubbed externalize
path and the envelope's ``data`` field is swapped for ``uri``;
non-envelope siblings are untouched; the wrapping dict structure is
preserved.
8. JSON-serializability post-condition: ``json.dumps`` round-trips the
full encoded payload (regression guard for the exact bug fixed).
9. Single-tensor / single-VideoInput / single-audio top-level inputs
still encode the same way they did before (no regression for the
existing leaf path).
Run with any python that has ``torch`` available (uses the ComfyUI
venv on this workstation):
python notes/run_encode_nested_dicts.py
The test loads ``_encode_one`` / ``_encode_inputs`` / ``_externalize_nested``
out of ``proxy_node.py`` via AST extraction so we don't pay the cost
of importing ComfyUI's ``comfy_api`` / ``comfy_api_nodes`` packages
(which those helpers don't touch). Same standalone pattern as
``run_image_max_batch.py``.
"""
from __future__ import annotations
import ast
import asyncio
import importlib.util
import json
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
class _FakeVideoInput:
"""Duck-typed VideoInput stand-in: ``save_to`` + ``get_duration``
are the only attrs ``serialization.is_video_input`` checks for."""
def __init__(self, tag: str, duration: float = 4.0) -> None:
self.tag = tag
self._duration = duration
def save_to(self, *args, **kwargs): # pragma: no cover — never called by stub
return None
def get_duration(self) -> float:
return self._duration
def _build_serialization_stub(protocol: types.ModuleType) -> types.ModuleType:
"""Minimal ``serialization`` shim covering exactly the surface the
helpers under test use."""
mod = types.ModuleType("rnp_smoke_serialization")
def _is_torch_tensor(value): # noqa: ANN001
return isinstance(value, torch.Tensor)
def is_audio_input(value): # noqa: ANN001
return (
isinstance(value, dict)
and {"waveform", "sample_rate"} <= set(value.keys())
)
def is_video_input(value): # noqa: ANN001
return (
callable(getattr(value, "save_to", None))
and callable(getattr(value, "get_duration", None))
)
def is_model3d_input(value): # noqa: ANN001
return False # unused in this test
def encode_image_tensor(tensor, *, accepts_batch=False): # noqa: ANN001, ARG001
return {
"type": "image",
"encoding": "png_base64",
"data": f"img:{int(tensor.shape[0])}",
}
def encode_mask_tensor(tensor): # noqa: ANN001, ARG001
return {"type": "mask", "encoding": "png_base64", "data": "mask"}
def encode_audio_input(value): # noqa: ANN001, ARG001
return {"type": "audio", "encoding": "mp3_base64", "data": "aud"}
def encode_video_input(value): # noqa: ANN001
return {
"type": "video",
"encoding": "mp4_base64",
"data": f"vid:{getattr(value, 'tag', '?')}",
}
def encode_model3d_input(value): # noqa: ANN001, ARG001 # pragma: no cover
return {"type": "model_3d", "encoding": "glb_base64", "data": "m3d"}
HEAVY_TYPES = {"image", "mask", "audio", "video", "model_3d"}
def is_envelope(value): # noqa: ANN001
return (
isinstance(value, dict)
and isinstance(value.get("type"), str)
and value.get("type") in HEAVY_TYPES
and isinstance(value.get("encoding"), str)
)
# Cap-aware ``maybe_externalize`` stub: swap ``data`` for ``uri`` when
# the inline payload would exceed ``max_inline_bytes``. Lets us prove
# that ``_externalize_nested`` reaches every per-slot envelope.
upload_log: list[str] = []
async def maybe_externalize(
envelope, # noqa: ANN001
*, server_url=None, max_inline_bytes=None, auth_headers=None, # noqa: ANN001, ARG001
):
if not server_url or max_inline_bytes is None:
return envelope
data = envelope.get("data")
if not isinstance(data, str):
return envelope
if len(data) <= max_inline_bytes:
return envelope
upload_log.append(f"{envelope.get('type')}:{data}")
out = {k: v for k, v in envelope.items() if k != "data"}
out["uri"] = f"https://upload/{envelope.get('type')}/{len(upload_log)}"
return out
mod._is_torch_tensor = _is_torch_tensor
mod.is_audio_input = is_audio_input
mod.is_video_input = is_video_input
mod.is_model3d_input = is_model3d_input
mod.encode_image_tensor = encode_image_tensor
mod.encode_mask_tensor = encode_mask_tensor
mod.encode_audio_input = encode_audio_input
mod.encode_video_input = encode_video_input
mod.encode_model3d_input = encode_model3d_input
mod.is_envelope = is_envelope
mod.maybe_externalize = maybe_externalize
mod._upload_log = upload_log # exposed for assertions
return mod
def _build_proxy_node_stub(
protocol: types.ModuleType, serialization: types.ModuleType,
) -> types.ModuleType:
"""Extract the helpers under test from ``proxy_node.py``."""
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 = {
"_enforce_local_validate",
"_check_image_max_batch",
"_encode_one",
"_encode_inputs",
"_externalize_nested",
}
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
# ---------------------------------------------------------------------------
# Tests
# ---------------------------------------------------------------------------
async def main() -> int:
protocol = _load_protocol()
serialization = _build_serialization_stub(protocol)
proxy_node = _build_proxy_node_stub(protocol, serialization)
img1 = torch.zeros((1, 64, 64, 3), dtype=torch.float32)
img2 = torch.zeros((1, 32, 32, 3), dtype=torch.float32)
vid1 = _FakeVideoInput("v1")
vid2 = _FakeVideoInput("v2")
audio = {
"waveform": torch.zeros((1, 1, 1000), dtype=torch.float32),
"sample_rate": 16000,
}
# ---- 1. Top-level AG-of-IMAGE dict encodes per-slot
out = await proxy_node._encode_inputs(
{"reference_images": {"image1": img1, "image2": img2}},
)
refs = out["reference_images"]
assert isinstance(refs, dict) and set(refs.keys()) == {"image1", "image2"}, refs
assert refs["image1"]["type"] == "image" and refs["image1"]["data"] == "img:1"
assert refs["image2"]["type"] == "image" and refs["image2"]["data"] == "img:1"
print("ok: top-level AG-of-IMAGE encoded per slot")
# ---- 2. Top-level AG-of-VIDEO dict encodes per-slot
out = await proxy_node._encode_inputs(
{"video_refs": {"video1": vid1, "video2": vid2}},
)
vrefs = out["video_refs"]
assert isinstance(vrefs, dict) and set(vrefs.keys()) == {"video1", "video2"}, vrefs
assert vrefs["video1"]["type"] == "video" and vrefs["video1"]["data"] == "vid:v1"
assert vrefs["video2"]["type"] == "video" and vrefs["video2"]["data"] == "vid:v2"
print("ok: top-level AG-of-VIDEO encoded per slot")
# ---- 3. DC-wrapping-AG: branch key untouched, nested AG dict encoded
dc_value = {
"model": "wan2.7-r2v",
"video_refs": {"video1": vid1},
"reference_images": {"image1": img1, "image2": img2},
"duration": 5,
}
out = await proxy_node._encode_inputs({"model": dc_value})
enc = out["model"]
assert enc["model"] == "wan2.7-r2v", enc
assert enc["duration"] == 5, enc
assert enc["video_refs"]["video1"]["type"] == "video"
assert enc["reference_images"]["image1"]["type"] == "image"
assert enc["reference_images"]["image2"]["type"] == "image"
print("ok: DC-wrapping-AG encoded leaves at every level")
# ---- 4. AUDIO dict is encoded, NOT recursed into
out = await proxy_node._encode_inputs({"audio": audio})
assert out["audio"] == {"type": "audio", "encoding": "mp3_base64", "data": "aud"}
print("ok: AUDIO dict still routes to encode_audio_input")
# ---- 5. Already-encoded envelope passes through (no double-encode)
pre_encoded = {"type": "image", "encoding": "png_base64", "data": "preimg"}
out = await proxy_node._encode_inputs({"images": pre_encoded})
assert out["images"] == pre_encoded, out
print("ok: pre-encoded envelope passed through unchanged")
# ---- 6. Plain scalar config dict passes through unchanged
out = await proxy_node._encode_inputs(
{"config": {"width": 1024, "height": 576, "name": "x"}},
)
assert out["config"] == {"width": 1024, "height": 576, "name": "x"}, out
print("ok: scalar config dict passed through unchanged")
# ---- 7. _externalize_nested walks per-slot envelopes
serialization._upload_log.clear()
big_img = torch.zeros((1, 64, 64, 3), dtype=torch.float32)
# encode_image_tensor stub emits ``data="img:1"`` (5 chars). Cap at 3
# to force every slot to externalize.
out = await proxy_node._encode_inputs(
{"reference_images": {"image1": big_img, "image2": big_img}},
server_url="https://fake.server",
max_inline_bytes=3,
)
refs = out["reference_images"]
assert refs["image1"].get("uri", "").startswith("https://upload/image/"), refs
assert refs["image2"].get("uri", "").startswith("https://upload/image/"), refs
assert "data" not in refs["image1"] and "data" not in refs["image2"]
assert len(serialization._upload_log) == 2, serialization._upload_log
print("ok: nested AG envelopes externalized per slot")
# ---- 7b. _externalize_nested leaves non-envelope siblings alone
serialization._upload_log.clear()
out = await proxy_node._encode_inputs(
{"model": {
"model": "wan2.7-r2v",
"duration": 5,
"reference_images": {"image1": big_img},
}},
server_url="https://fake.server",
max_inline_bytes=3,
)
enc = out["model"]
assert enc["model"] == "wan2.7-r2v" and enc["duration"] == 5, enc
assert enc["reference_images"]["image1"]["uri"].startswith("https://upload/image/")
print("ok: scalar siblings preserved alongside externalized envelopes")
# ---- 8. JSON-serializability post-condition
payload = await proxy_node._encode_inputs({
"reference_images": {"image1": img1, "image2": img2},
"video_refs": {"video1": vid1},
"audio": audio,
"config": {"width": 1024},
"scalar": "hello",
"n": 42,
})
json.dumps(payload) # would raise TypeError pre-fix
print("ok: full encoded payload is JSON-serializable")
# ---- 9. Leaf-path regression: single-tensor / single-video / single-audio
out = await proxy_node._encode_inputs({"image": img1})
assert out["image"]["type"] == "image", out
out = await proxy_node._encode_inputs({"video": vid1})
assert out["video"]["type"] == "video", out
out = await proxy_node._encode_inputs({"audio": audio})
assert out["audio"]["type"] == "audio", out
out = await proxy_node._encode_inputs({"prompt": "hi", "seed": 7})
assert out == {"prompt": "hi", "seed": 7}, out
print("ok: top-level leaf path unchanged")
print("ALL OK")
return 0
if __name__ == "__main__":
raise SystemExit(asyncio.run(main()))
+441
View File
@@ -0,0 +1,441 @@
"""Smoke test for the descriptor-driven ``local_validate.image_max_batch``
client-side fail-fast.
Exercises the proxy_node's ``_collect_local_validate`` /
``_enforce_local_validate`` pair against a fake descriptor that mirrors
the Flux2 wire shape. Asserts that:
1. ``_collect_local_validate`` extracts ``{"images": {"image_max_batch": 2}}``
from the descriptor's IMAGE-input options dict.
2. Encoding a 3-frame tensor for that input raises ``RnpProtocolError``
with ``code=INPUT_INVALID`` and ``user_facing=True`` *before* any
serialization / HTTP attempt.
3. A 1-frame tensor passes through cleanly.
4. A descriptor without ``local_validate`` is a no-op.
5. ``cap=1`` surfaces the single-frame copy.
6. A ``local_validate`` block missing the ``image_max_batch`` key is a no-op.
Run with any python that has ``torch`` available:
python notes/run_image_max_batch.py
The test loads the helper functions out of ``proxy_node.py`` via AST
extraction so it doesn't pay the cost of importing ComfyUI's
``comfy_api`` / ``comfy_api_nodes`` packages (which the helper
functions don't touch). Standalone — no test runner / fixtures needed.
"""
from __future__ import annotations
import ast
import asyncio
import importlib.util
import os
import sys
import types
import torch
# ---------------------------------------------------------------------------
# Locate the worktree under test (this file's parent's parent).
# ---------------------------------------------------------------------------
_HERE = os.path.dirname(os.path.abspath(__file__))
_CLIENT_DIR = os.path.dirname(_HERE)
def _load_protocol() -> types.ModuleType:
"""Load ``protocol.py`` standalone (no relative imports)."""
path = os.path.join(_CLIENT_DIR, "protocol.py")
spec = importlib.util.spec_from_file_location("rnp_smoke_protocol", path)
mod = importlib.util.module_from_spec(spec)
sys.modules[spec.name] = mod
spec.loader.exec_module(mod)
return mod
def _build_serialization_stub(protocol: types.ModuleType) -> types.ModuleType:
"""Minimal ``serialization`` shim covering exactly the surface
``_encode_one`` / ``_enforce_local_validate`` / ``_encode_inputs`` use:
* ``_is_torch_tensor(value)`` — duck-type check for torch tensors.
* ``is_audio_input(value)`` — always False (no audio in this test).
* ``encode_image_tensor(tensor, *, accepts_batch)`` — fake envelope.
* ``encode_mask_tensor(tensor)`` — fake envelope.
* ``is_envelope(value)`` — False so ``maybe_externalize`` is skipped.
"""
mod = types.ModuleType("rnp_smoke_serialization")
def _is_torch_tensor(value): # noqa: ANN001
try:
import torch as _torch
except ImportError:
return False
return isinstance(value, _torch.Tensor)
def is_audio_input(value): # noqa: ANN001
return isinstance(value, dict) and {"waveform", "sample_rate"} <= set(value.keys())
def encode_image_tensor(tensor, *, accepts_batch=False): # noqa: ANN001, ARG001
return {"type": "image", "encoding": "png_base64", "frames": int(tensor.shape[0])}
def encode_mask_tensor(tensor): # noqa: ANN001, ARG001
return {"type": "mask", "encoding": "png_base64"}
def encode_audio_input(value): # noqa: ANN001, ARG001
return {"type": "audio", "encoding": "mp3_base64"}
def is_envelope(value): # noqa: ANN001
return False # smoke test never externalizes
mod._is_torch_tensor = _is_torch_tensor
mod.is_audio_input = is_audio_input
mod.encode_image_tensor = encode_image_tensor
mod.encode_mask_tensor = encode_mask_tensor
mod.encode_audio_input = encode_audio_input
mod.is_envelope = is_envelope
async def maybe_externalize(*args, **kwargs): # pragma: no cover — unused
return args[0] if args else None
mod.maybe_externalize = maybe_externalize
return mod
def _build_proxy_node_stub(
protocol: types.ModuleType, serialization: types.ModuleType,
) -> types.ModuleType:
"""Extract the four helper functions under test from ``proxy_node.py``
and exec them in a minimal namespace so we don't drag the rest of
the module's heavy ComfyUI imports into the test process.
"""
mod = types.ModuleType("rnp_smoke_proxy_node")
mod.__dict__.update({
"Any": object,
"log": types.SimpleNamespace(warning=lambda *a, **kw: None),
"serialization": serialization,
"RnpProtocolError": protocol.RnpProtocolError,
"ErrorCode": protocol.ErrorCode,
})
src_path = os.path.join(_CLIENT_DIR, "proxy_node.py")
with open(src_path, "r", encoding="utf-8") as fh:
src = fh.read()
tree = ast.parse(src)
wanted = {
"_collect_local_validate",
"_rules_from_input_spec",
"_enforce_local_validate",
"_check_image_max_batch",
"_encode_one",
"_encode_inputs",
}
nodes = [
n for n in tree.body
if isinstance(n, (ast.AsyncFunctionDef, ast.FunctionDef)) and n.name in wanted
]
snippet = ast.Module(body=nodes, type_ignores=[])
code = compile(snippet, src_path, "exec")
exec(code, mod.__dict__)
missing = wanted - set(mod.__dict__)
if missing:
raise RuntimeError(f"proxy_node helpers missing from extract: {missing}")
return mod
# ---------------------------------------------------------------------------
# Test fixtures
# ---------------------------------------------------------------------------
def _flux2_like_descriptor(cap: int) -> dict:
"""Mirrors the Flux2 ``images`` input descriptor surface."""
return {
"name": "FakeFlux2_RNP",
"input": {
"required": {
"prompt": ["STRING", {"multiline": True, "default": ""}],
},
"optional": {
"images": ["IMAGE", {
"tooltip": "Up to N images to be used as references.",
"local_validate": {"image_max_batch": cap},
}],
},
"hidden": {},
},
}
def _no_validate_descriptor() -> dict:
return {
"name": "FakeNoValidate_RNP",
"input": {
"required": {},
"optional": {
"images": ["IMAGE", {"tooltip": "no cap"}],
},
"hidden": {},
},
}
def _autogrow_descriptor(cap: int, prefix: str = "reference") -> dict:
"""Mirrors GrokVideoReferenceNode_RNP's top-level Autogrow input
after server descriptor unwraps the nested DynamicCombo branch
(kept simpler here for direct AUTOGROW coverage)."""
return {
"name": "FakeAutogrow_RNP",
"input": {
"required": {
"reference_images": ["AUTOGROW", {
"template": ["IMAGE", {
"local_validate": {"image_max_batch": cap},
}],
"prefix": prefix,
"min": 1,
"max": cap,
}],
},
"optional": {},
"hidden": {},
},
}
def _dyncombo_with_autogrow_descriptor() -> dict:
"""Mirrors GrokImageEditNodeV2_RNP's ``model`` DynamicCombo with
two branches each carrying an Autogrow ``images`` template — one
capped at 3, the other at 1."""
def branch(key: str, cap: int) -> dict:
return {
"key": key,
"inputs": [
["images", "AUTOGROW", {
"template": ["IMAGE", {
"local_validate": {"image_max_batch": cap},
}],
"prefix": "image",
"min": 1,
"max": cap,
}],
["resolution", "COMBO", {"options": ["1K", "2K"]}],
],
}
return {
"name": "FakeDynComboAutogrow_RNP",
"input": {
"required": {
"model": ["DYNAMIC_COMBO", {
"options": [
branch("grok-imagine-image-quality", 3),
branch("grok-imagine-image-pro", 1),
],
}],
},
"optional": {},
"hidden": {},
},
}
# ---------------------------------------------------------------------------
# Tests
# ---------------------------------------------------------------------------
async def main() -> int:
protocol = _load_protocol()
serialization = _build_serialization_stub(protocol)
proxy_node = _build_proxy_node_stub(protocol, serialization)
# ---- 1. _collect_local_validate extracts the rule
desc = _flux2_like_descriptor(cap=2)
rules = proxy_node._collect_local_validate(desc)
assert rules == {"images": {"image_max_batch": 2}}, rules
print("ok: _collect_local_validate extracted", rules)
# ---- 2. over-cap tensor raises INPUT_INVALID before encoding
over_cap = torch.zeros((3, 64, 64, 3), dtype=torch.float32)
raised = False
try:
await proxy_node._encode_inputs(
{"images": over_cap},
input_serialization={"images": ["png_base64", "png_base64_batch"]},
local_validate=rules,
node_id="FakeFlux2_RNP",
)
except protocol.RnpProtocolError as e:
raised = True
assert e.code == protocol.ErrorCode.INPUT_INVALID, e.code
assert e.user_facing is True, e.user_facing
msg = str(e)
assert "FakeFlux2_RNP" in msg, msg
assert "3" in msg and "2" in msg, msg
print("ok: over-cap raised INPUT_INVALID:", msg)
assert raised, "over-cap tensor did not raise"
# ---- 3. at-cap tensor passes through (no exception)
at_cap = torch.zeros((2, 64, 64, 3), dtype=torch.float32)
out = await proxy_node._encode_inputs(
{"images": at_cap},
input_serialization={"images": ["png_base64", "png_base64_batch"]},
local_validate=rules,
node_id="FakeFlux2_RNP",
)
assert "images" in out
print("ok: at-cap (B=2) encoded without error")
# ---- 4. cap=1 / multi-frame raises with the single-frame copy
desc1 = _flux2_like_descriptor(cap=1)
rules1 = proxy_node._collect_local_validate(desc1)
raised = False
try:
await proxy_node._encode_inputs(
{"images": over_cap},
local_validate=rules1,
node_id="FakeI2V_RNP",
)
except protocol.RnpProtocolError as e:
raised = True
assert e.code == protocol.ErrorCode.INPUT_INVALID
assert "single image" in str(e), str(e)
print("ok: cap=1 single-frame copy fired:", str(e))
assert raised
# ---- 5. descriptor without local_validate is a no-op
desc_n = _no_validate_descriptor()
rules_n = proxy_node._collect_local_validate(desc_n)
assert rules_n == {}, rules_n
out = await proxy_node._encode_inputs(
{"images": over_cap},
input_serialization={"images": ["png_base64", "png_base64_batch"]},
local_validate=rules_n,
node_id="FakeNoValidate_RNP",
)
assert "images" in out
print("ok: missing local_validate is a no-op")
# ---- 6. local_validate block without image_max_batch sub-key is a no-op
out = await proxy_node._encode_inputs(
{"images": over_cap},
input_serialization={"images": ["png_base64", "png_base64_batch"]},
local_validate={"images": {"some_other_rule": 7}},
node_id="FakeOther_RNP",
)
assert "images" in out
print("ok: missing image_max_batch sub-key is a no-op")
# ---- 7. AUTOGROW: each slot value enforced against the template cap
desc_ag = _autogrow_descriptor(cap=2)
rules_ag = proxy_node._collect_local_validate(desc_ag)
assert "reference_images" in rules_ag, rules_ag
assert rules_ag["reference_images"].get("__template__") == {"image_max_batch": 2}, rules_ag
print("ok: AUTOGROW collected __template__ rules:", rules_ag)
one_frame = torch.zeros((1, 64, 64, 3), dtype=torch.float32)
raised = False
try:
# Slot 1 is over-cap (B=3 > cap=2) → should fire on that slot
await proxy_node._encode_inputs(
{"reference_images": {"reference0": one_frame, "reference1": over_cap}},
local_validate=rules_ag,
node_id="FakeAutogrow_RNP",
)
except protocol.RnpProtocolError as e:
raised = True
assert e.code == protocol.ErrorCode.INPUT_INVALID
msg = str(e)
# Slot path label should mention reference_images.<slot_key>
assert "reference_images" in msg, msg
assert "reference1" in msg, msg
print("ok: AUTOGROW slot over-cap fired:", msg)
assert raised, "AUTOGROW slot over-cap did not raise"
# All-under-cap autogrow passes (no exception)
out = await proxy_node._encode_inputs(
{"reference_images": {"reference0": one_frame, "reference1": one_frame, "reference2": None}},
local_validate=rules_ag,
node_id="FakeAutogrow_RNP",
)
assert "reference_images" in out
print("ok: AUTOGROW all-under-cap (None slots tolerated) passed")
# ---- 8. DYNAMIC_COMBO with nested AUTOGROW: branch-aware enforcement
desc_dc = _dyncombo_with_autogrow_descriptor()
rules_dc = proxy_node._collect_local_validate(desc_dc)
assert "model" in rules_dc, rules_dc
branches = rules_dc["model"].get("__branches__")
assert isinstance(branches, dict), rules_dc
assert "grok-imagine-image-quality" in branches
assert "grok-imagine-image-pro" in branches
assert branches["grok-imagine-image-quality"]["images"]["__template__"] == {"image_max_batch": 3}
assert branches["grok-imagine-image-pro"]["images"]["__template__"] == {"image_max_batch": 1}
print("ok: DYNAMIC_COMBO collected branch-keyed templates")
# Branch A (cap=3): slot with B=4 should raise
over_4 = torch.zeros((4, 64, 64, 3), dtype=torch.float32)
raised = False
try:
await proxy_node._encode_inputs(
{"model": {
"model": "grok-imagine-image-quality",
"resolution": "1K",
"images": {"image0": over_4},
}},
local_validate=rules_dc,
node_id="FakeDynComboAutogrow_RNP",
)
except protocol.RnpProtocolError as e:
raised = True
assert e.code == protocol.ErrorCode.INPUT_INVALID
msg = str(e)
assert "image0" in msg and "3" in msg and "4" in msg, msg
print("ok: DYNAMIC_COMBO branch-A (cap=3) fired on B=4 slot:", msg)
assert raised
# Branch A (cap=3): all slots within cap → passes
out = await proxy_node._encode_inputs(
{"model": {
"model": "grok-imagine-image-quality",
"resolution": "2K",
"images": {"image0": one_frame, "image1": one_frame, "image2": one_frame},
}},
local_validate=rules_dc,
node_id="FakeDynComboAutogrow_RNP",
)
assert "model" in out
print("ok: DYNAMIC_COMBO branch-A all-under-cap passed")
# Branch B (cap=1): a B=3 slot fires the single-frame copy
raised = False
try:
await proxy_node._encode_inputs(
{"model": {
"model": "grok-imagine-image-pro",
"resolution": "2K",
"images": {"image0": over_cap},
}},
local_validate=rules_dc,
node_id="FakeDynComboAutogrow_RNP",
)
except protocol.RnpProtocolError as e:
raised = True
assert e.code == protocol.ErrorCode.INPUT_INVALID
assert "single image" in str(e), str(e)
print("ok: DYNAMIC_COMBO branch-B (cap=1) fired single-frame copy:", str(e))
assert raised
# Unknown branch key → no enforcement (defensive)
out = await proxy_node._encode_inputs(
{"model": {
"model": "unknown-branch-key",
"images": {"image0": over_cap},
}},
local_validate=rules_dc,
node_id="FakeDynComboAutogrow_RNP",
)
assert "model" in out
print("ok: DYNAMIC_COMBO unknown branch key is a defensive no-op")
print("PASS: image_max_batch local_validate smoke test (incl. nested traversal)")
return 0
if __name__ == "__main__":
sys.exit(asyncio.run(main()))
+461
View File
@@ -0,0 +1,461 @@
"""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.")
+316
View File
@@ -0,0 +1,316 @@
"""Smoke test for the client-side MODEL_3D decoder.
Verifies:
1. The vendored ``protocol.HEAVY_TYPES`` now recognises ``model_3d`` so
``is_envelope`` returns True for an inbound MODEL_3D envelope.
2. ``serialization.decode_model3d_envelope`` round-trips an inline
``glb_inline`` envelope into a ``comfy_api.latest._util.File3D``
whose ``BytesIO`` payload matches the original GLB bytes.
3. ``decode_envelope`` dispatches ``type=model_3d`` to the new decoder.
4. The proxy_node ``_OUTPUT_CLASSES`` now maps ``"MODEL_3D"`` to
``IO.File3DGLB.Output`` so a descriptor with ``output=["MODEL_3D"]``
parses without falling through to the opaque IO.Custom branch.
5. ``client.CLIENT_CAPABILITIES`` advertises
``model_3d:glb_inline`` on every outbound request so descriptors
gated on the token negotiate cleanly.
6. End-to-end against the server's ``make_model3d_envelope`` helper
(imported from the comfy-rnp-server worktree) — the bytes the
server packs are exactly the bytes the client decodes.
Run with the ComfyUI venv (needs torch / comfy_api on PYTHONPATH):
python notes/run_model3d_decoder.py
"""
from __future__ import annotations
import base64
import importlib.util
import io
import json
import os
import sys
import types
# ---------------------------------------------------------------------------
# Locate the worktree under test.
# ---------------------------------------------------------------------------
_HERE = os.path.dirname(os.path.abspath(__file__))
_CLIENT_DIR = os.path.dirname(_HERE)
_COMFYUI_DIR = os.path.dirname(os.path.dirname(_CLIENT_DIR))
_SERVER_DIR = os.path.normpath(
os.path.join(_COMFYUI_DIR, "..", "comfy-rnp-server")
)
def _load_module(name: str, path: str) -> types.ModuleType:
spec = importlib.util.spec_from_file_location(name, path)
mod = importlib.util.module_from_spec(spec)
sys.modules[name] = mod
spec.loader.exec_module(mod)
return mod
# Make ComfyUI importable so ``comfy_api.latest._util.File3D`` resolves.
if _COMFYUI_DIR not in sys.path:
sys.path.insert(0, _COMFYUI_DIR)
# And the server worktree so we can import the ENCODER side end-to-end.
if _SERVER_DIR not in sys.path:
sys.path.insert(0, _SERVER_DIR)
# ComfyUI's full ``comfy_api`` tree pulls torch / PIL / comfy /
# comfy_execution at import time — way too heavy for a decoder
# round-trip check. Stub just the leaf module our decoder imports
# (``comfy_api.latest._util``) with a faithful pure-Python ``File3D``.
class _StubFile3D:
def __init__(self, source, file_format: str = "") -> None:
self._source = source
self._format = file_format
_comfy_api = types.ModuleType("comfy_api")
_comfy_api_latest = types.ModuleType("comfy_api.latest")
_comfy_api_latest_util = types.ModuleType("comfy_api.latest._util")
_comfy_api_latest_util.File3D = _StubFile3D
_comfy_api.latest = _comfy_api_latest
_comfy_api_latest._util = _comfy_api_latest_util
sys.modules.setdefault("comfy_api", _comfy_api)
sys.modules.setdefault("comfy_api.latest", _comfy_api_latest)
sys.modules.setdefault("comfy_api.latest._util", _comfy_api_latest_util)
# Load the client modules as if they were a package — the ``protocol``
# module imports from ``.protocol`` relative to its package, so we
# emulate the package layout under the name ``comfy_remote_nodes``.
pkg = types.ModuleType("comfy_remote_nodes_test")
pkg.__path__ = [_CLIENT_DIR]
sys.modules["comfy_remote_nodes_test"] = pkg
protocol = _load_module(
"comfy_remote_nodes_test.protocol",
os.path.join(_CLIENT_DIR, "protocol.py"),
)
# serialization imports ``client`` for ``rnp_client`` but we don't need
# the HTTP client for the decoder path; provide a tiny stub so the
# import succeeds.
client_stub = types.ModuleType("comfy_remote_nodes_test.client")
sys.modules["comfy_remote_nodes_test.client"] = client_stub
serialization = _load_module(
"comfy_remote_nodes_test.serialization",
os.path.join(_CLIENT_DIR, "serialization.py"),
)
# ---------------------------------------------------------------------------
# Build a minimal valid binary glTF 2.0 (GLB) payload for the round-trip.
# Format: 12-byte header + N chunks. Header: magic(4) + version(4) + length(4).
# Magic must be b"glTF", version=2. We include exactly one JSON chunk that
# describes an empty asset (enough to satisfy any GLB-magic check; downstream
# nodes do their own validation).
# ---------------------------------------------------------------------------
def _make_minimal_glb() -> bytes:
json_payload = b'{"asset":{"version":"2.0"}}'
# Chunks must be 4-byte aligned; pad JSON with spaces (0x20).
pad = (4 - (len(json_payload) % 4)) % 4
json_payload = json_payload + (b" " * pad)
json_chunk = (
len(json_payload).to_bytes(4, "little") # chunk length
+ b"JSON" # chunk type
+ json_payload # chunk data
)
total = 12 + len(json_chunk)
header = (
b"glTF"
+ (2).to_bytes(4, "little")
+ total.to_bytes(4, "little")
)
return header + json_chunk
# ---------------------------------------------------------------------------
# 1. HEAVY_TYPES recognises model_3d.
# ---------------------------------------------------------------------------
assert "model_3d" in protocol.HEAVY_TYPES, (
f"HEAVY_TYPES missing 'model_3d': {sorted(protocol.HEAVY_TYPES)}"
)
assert protocol.Capability.MODEL_3D_GLB_INLINE == "model_3d:glb_inline", (
f"Capability.MODEL_3D_GLB_INLINE wrong value: "
f"{protocol.Capability.MODEL_3D_GLB_INLINE!r}"
)
print("PASS: HEAVY_TYPES includes 'model_3d'; Capability constant intact.")
# ---------------------------------------------------------------------------
# 2. is_envelope returns True for a MODEL_3D envelope.
# ---------------------------------------------------------------------------
glb_bytes = _make_minimal_glb()
envelope = {
"type": "model_3d",
"encoding": "glb_inline",
"data": base64.b64encode(glb_bytes).decode("ascii"),
"format": "glb",
"byte_size": len(glb_bytes),
}
assert protocol.is_envelope(envelope), (
"is_envelope returned False for a well-formed MODEL_3D envelope; "
"HEAVY_TYPES update did not take effect."
)
print("PASS: is_envelope recognises model_3d envelopes.")
# ---------------------------------------------------------------------------
# 3. decode_model3d_envelope round-trips inline bytes.
# ---------------------------------------------------------------------------
file3d = serialization.decode_model3d_envelope(envelope)
from comfy_api.latest._util import File3D as _File3D_for_check # noqa: E402
assert isinstance(file3d, _File3D_for_check), (
f"Expected File3D, got {type(file3d).__name__}"
)
# File3D exposes get_data() / get_source(); the public API on this
# version returns a BytesIO when constructed with one. Drain bytes:
src = file3d._source # internal, but stable across this code path
assert hasattr(src, "read"), f"File3D source not BytesIO-like: {type(src)}"
src.seek(0)
round_trip = src.read()
assert round_trip == glb_bytes, (
f"GLB byte mismatch after decode: input={len(glb_bytes)} bytes, "
f"output={len(round_trip)} bytes"
)
assert round_trip[:4] == b"glTF", "GLB magic bytes lost in round-trip"
print(f"PASS: decode_model3d_envelope round-trips {len(glb_bytes)} GLB bytes.")
# ---------------------------------------------------------------------------
# 4. decode_envelope dispatches model_3d to the new decoder.
# ---------------------------------------------------------------------------
file3d2 = serialization.decode_envelope(envelope)
assert isinstance(file3d2, _File3D_for_check), (
f"decode_envelope dispatch failed for type=model_3d: got "
f"{type(file3d2).__name__}"
)
print("PASS: decode_envelope dispatches type='model_3d' correctly.")
# ---------------------------------------------------------------------------
# 5. Unsupported encoding raises a hard error (defence-in-depth so a
# future encoding never silently downgrades to garbage bytes).
# ---------------------------------------------------------------------------
bad = dict(envelope, encoding="obj_inline")
try:
serialization.decode_model3d_envelope(bad)
except protocol.RnpProtocolError as e:
assert "Unsupported model_3d encoding" in str(e), (
f"Wrong error message for unsupported encoding: {e!r}"
)
assert e.code == protocol.ErrorCode.INTERNAL
print("PASS: unsupported model_3d encoding raises RnpProtocolError.")
else:
raise AssertionError(
"decode_model3d_envelope accepted an unsupported encoding"
)
# ---------------------------------------------------------------------------
# 6. End-to-end: server's make_model3d_envelope -> client decode.
# Skips when the server worktree isn't on disk (e.g. CI running
# against just the client repo).
# ---------------------------------------------------------------------------
server_protocol_path = os.path.join(
_SERVER_DIR, "comfy_rnp_protocol", "envelopes.py",
)
if os.path.exists(server_protocol_path):
# Import the server-side envelopes module standalone (it has no
# ComfyUI deps and uses its own sibling ``constants`` import).
server_pkg = types.ModuleType("comfy_rnp_protocol_e2e")
server_pkg.__path__ = [os.path.join(_SERVER_DIR, "comfy_rnp_protocol")]
sys.modules["comfy_rnp_protocol_e2e"] = server_pkg
server_constants = _load_module(
"comfy_rnp_protocol_e2e.constants",
os.path.join(_SERVER_DIR, "comfy_rnp_protocol", "constants.py"),
)
server_envelopes = _load_module(
"comfy_rnp_protocol_e2e.envelopes",
os.path.join(_SERVER_DIR, "comfy_rnp_protocol", "envelopes.py"),
)
# Build using the server helper exactly as a provider would.
server_env = server_envelopes.make_model3d_envelope(
encoding="glb_inline",
data=base64.b64encode(glb_bytes).decode("ascii"),
format="glb",
byte_size=len(glb_bytes),
)
assert server_env["type"] == "model_3d", server_env
assert server_env["encoding"] == "glb_inline", server_env
assert "model_3d" in server_envelopes.HEAVY_TYPES, (
"Server HEAVY_TYPES missing 'model_3d' — server PR #37 should "
"have added it; reinstall comfy_rnp_protocol if this fires."
)
# Pipe through the client decoder.
file3d3 = serialization.decode_envelope(server_env)
src = file3d3._source
src.seek(0)
assert src.read() == glb_bytes, "End-to-end byte mismatch"
print(
"PASS: end-to-end server make_model3d_envelope -> client "
"decode_envelope preserves bytes."
)
else:
print(
f"SKIP end-to-end: server worktree not found at {_SERVER_DIR!r}"
)
# ---------------------------------------------------------------------------
# 7. proxy_node._OUTPUT_CLASSES maps "MODEL_3D" -> IO.File3DGLB.Output.
# proxy_node pulls in the full comfy_api / server / comfy_api_nodes
# tree at import time (too heavy to stub here), so we verify the
# mapping with a source-level check instead. The pattern is exact:
# a quoted ``"MODEL_3D":`` key whose value is the
# ``IO.File3DGLB.Output`` class reference inside the
# ``_OUTPUT_CLASSES`` dict literal. The same source check would
# catch a typo (e.g. ``IO.File3DAny.Output``) that a smoke test
# importing the live module would miss too — the dict literal is
# the contract.
# ---------------------------------------------------------------------------
proxy_src = open(
os.path.join(_CLIENT_DIR, "proxy_node.py"), "r", encoding="utf-8",
).read()
assert "_OUTPUT_CLASSES = {" in proxy_src, "proxy_node missing _OUTPUT_CLASSES dict"
assert '"MODEL_3D":' in proxy_src and "IO.File3DGLB.Output" in proxy_src, (
"proxy_node._OUTPUT_CLASSES does not appear to map "
"'MODEL_3D' -> IO.File3DGLB.Output"
)
# Verify the mapping is in the same dict literal (not somewhere else).
oc_start = proxy_src.index("_OUTPUT_CLASSES = {")
oc_end = proxy_src.index("}", oc_start)
oc_block = proxy_src[oc_start:oc_end]
assert '"MODEL_3D":' in oc_block, (
"'MODEL_3D' key not inside the _OUTPUT_CLASSES dict literal"
)
assert "IO.File3DGLB.Output" in oc_block, (
"IO.File3DGLB.Output value not inside the _OUTPUT_CLASSES dict literal"
)
print(
"PASS: proxy_node._OUTPUT_CLASSES dict literal contains "
"'MODEL_3D' -> IO.File3DGLB.Output."
)
# ---------------------------------------------------------------------------
# 8. CLIENT_CAPABILITIES advertises model_3d:glb_inline.
# client.py imports aiohttp / comfy_api_nodes too — same source-level
# contract check.
# ---------------------------------------------------------------------------
client_src = open(
os.path.join(_CLIENT_DIR, "client.py"), "r", encoding="utf-8",
).read()
assert "CLIENT_CAPABILITIES = [" in client_src, (
"client.py missing CLIENT_CAPABILITIES list"
)
cc_start = client_src.index("CLIENT_CAPABILITIES = [")
cc_end = client_src.index("]", cc_start)
cc_block = client_src[cc_start:cc_end]
assert "Capability.MODEL_3D_GLB_INLINE" in cc_block, (
"Capability.MODEL_3D_GLB_INLINE not advertised in CLIENT_CAPABILITIES"
)
print("PASS: CLIENT_CAPABILITIES advertises Capability.MODEL_3D_GLB_INLINE.")
print("ALL CHECKS PASSED.")
+129 -2
View File
@@ -21,7 +21,18 @@ DEFAULT_POLL_INTERVAL_S = 1.5
# Values mirror comfy_api_nodes.util.client.poll_op_raw so RNP-routed
# nodes behave identically to a direct upstream call when the descriptor
# leaves a field unset.
DEFAULT_MAX_POLL_ATTEMPTS = 160
#
# Lifetime budget design: the wire format declares ``soft_timeout_s``
# (server-side stuck threshold) and ``hard_timeout_s`` (server-side
# hard kill) as durations. The client's poll-loop cap is a derived
# value — ``max_poll_attempts = ceil((hard_timeout_s + DEFAULT_POLL_GRACE_S)
# / poll_interval_s)`` — not a separate wire field, so changing the
# poll cadence can never silently shrink or stretch the contract.
# Industry precedent: OAuth2 device flow §3.5, Kubernetes
# ``--timeout``, gRPC deadlines all express bounds as durations.
DEFAULT_SOFT_TIMEOUT_S = 240.0
DEFAULT_HARD_TIMEOUT_S = 300.0
DEFAULT_POLL_GRACE_S = 30.0
DEFAULT_TIMEOUT_PER_POLL_S = 120.0
DEFAULT_MAX_RETRIES_PER_POLL = 10
DEFAULT_RETRY_DELAY_PER_POLL_S = 1.0
@@ -93,14 +104,130 @@ 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"
HEAVY_TYPES = frozenset({"image", "video", "audio", "mask"})
HEAVY_TYPES = frozenset({"image", "video", "audio", "mask", "model_3d"})
def is_envelope(value: Any) -> bool:
+689 -55
View File
@@ -3,9 +3,17 @@
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; an input outside that
set causes the whole node to be skipped with a warning rather than
half-built.
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.
Hidden inputs (``auth_token_comfy_org`` / ``api_key_comfy_org`` /
``unique_id`` / ``prompt`` / ``extra_pnginfo`` / ``dynprompt``) are
@@ -22,6 +30,7 @@ from __future__ import annotations
import asyncio
import contextlib
import logging
import math
import time
import uuid
from typing import Any
@@ -36,8 +45,9 @@ from . import client as rnp_client
from . import serialization
from .protocol import (
DEFAULT_CANCEL_TIMEOUT_S,
DEFAULT_MAX_POLL_ATTEMPTS,
DEFAULT_HARD_TIMEOUT_S,
DEFAULT_MAX_RETRIES_PER_POLL,
DEFAULT_POLL_GRACE_S,
DEFAULT_POLL_INTERVAL_S,
DEFAULT_RETRY_BACKOFF_PER_POLL,
DEFAULT_RETRY_DELAY_PER_POLL_S,
@@ -146,19 +156,32 @@ def _extract_execution_policy(execution: dict[str, Any]) -> dict[str, Any]:
node with an empty ``execution`` block behaves identically to a
direct upstream call. The schema is the wire contract from
``comfy_rnp_protocol.constants``: ``poll_interval_s``,
``max_poll_attempts``, ``timeout_per_poll_s``, ``cancel_timeout_s``,
``max_retries_per_poll``, ``retry_delay_per_poll_s``,
``retry_backoff_per_poll``, ``retry: {max, delay_s, backoff,
retry_on}``, ``idempotency``.
``soft_timeout_s``, ``hard_timeout_s``, ``timeout_per_poll_s``,
``cancel_timeout_s``, ``max_retries_per_poll``,
``retry_delay_per_poll_s``, ``retry_backoff_per_poll``,
``retry: {max, delay_s, backoff, retry_on}``, ``idempotency``.
The client's poll-loop cap (``max_poll_attempts``) is **derived**
from ``hard_timeout_s + DEFAULT_POLL_GRACE_S`` and
``poll_interval_s`` — there is no wire field for it. Coupling a
contract bound to a cadence value silently breaks every time you
tune the cadence; expressing the bound as a duration keeps the
contract invariant under cadence changes (see protocol.py
"Lifetime budget design" comment).
"""
retry_block = execution.get("retry") if isinstance(execution.get("retry"), dict) else {}
poll_interval_s = _coerce_pos(
execution.get("poll_interval_s"), DEFAULT_POLL_INTERVAL_S,
)
hard_timeout_s = _coerce_pos(
execution.get("hard_timeout_s"),
_coerce_pos(execution.get("soft_timeout_s"), DEFAULT_HARD_TIMEOUT_S),
)
poll_budget_s = hard_timeout_s + DEFAULT_POLL_GRACE_S
max_poll_attempts = max(1, math.ceil(poll_budget_s / poll_interval_s))
return {
"poll_interval_s": _coerce_pos(
execution.get("poll_interval_s"), DEFAULT_POLL_INTERVAL_S,
),
"max_poll_attempts": _coerce_pos_int(
execution.get("max_poll_attempts"), DEFAULT_MAX_POLL_ATTEMPTS,
),
"poll_interval_s": poll_interval_s,
"max_poll_attempts": max_poll_attempts,
"timeout_per_poll_s": _coerce_pos(
execution.get("timeout_per_poll_s"), DEFAULT_TIMEOUT_PER_POLL_S,
),
@@ -290,6 +313,32 @@ def build_node_class(
schema_hash = remote.get("schema_hash")
execute_path = execute_endpoint["path"]
execution = remote.get("execution") or {}
# ``descriptor.remote.input_serialization`` advertises the wire
# encodings each input accepts. The wire shape is
# ``{input_name: <str | list[str]>}`` — single string for a
# single-encoding input (e.g. ``"png_base64"``), list when the
# input accepts multiple encodings (e.g.
# ``["png_base64", "png_base64_batch"]`` for a multi-frame IMAGE).
# Normalize to ``dict[str, list[str]]`` so the encoder can do a
# single ``in`` check per input without re-parsing per call.
raw_input_serialization = (
remote.get("input_serialization") if isinstance(remote.get("input_serialization"), dict) else {}
)
input_serialization: dict[str, list[str]] = {}
for name, val in raw_input_serialization.items():
if isinstance(val, str):
input_serialization[name] = [val]
elif isinstance(val, list):
input_serialization[name] = [v for v in val if isinstance(v, str)]
# ``descriptor.input.<section>.<name>[1].local_validate`` carries
# declarative client-side validation rules (e.g.
# ``{"image_max_batch": 9}`` for a Flux2 reference-image input).
# The encoder consults this map and raises ``INPUT_INVALID``
# locally before any HTTP round trip when an over-cap batch is
# wired in. Lookup is defensive: missing ``local_validate`` →
# no validation; missing per-rule key → no validation.
local_validate = _collect_local_validate(descriptor)
timeout_s = float(
execution.get("hard_timeout_s")
or execution.get("soft_timeout_s")
@@ -370,6 +419,8 @@ def build_node_class(
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
@@ -395,6 +446,111 @@ _HIDDEN_MARKERS = {
}
def _collect_local_validate(descriptor: dict[str, Any]) -> dict[str, dict[str, Any]]:
"""Pull ``local_validate`` rule dicts off every required/optional input
spec in the descriptor, recursing into AUTOGROW templates and
DYNAMIC_COMBO branches.
Returns ``{input_name: rules_dict}`` for every input that carries a
non-empty ``local_validate`` block (or whose nested template /
branches do). The returned ``rules_dict`` may carry the literal
``local_validate`` keys (e.g. ``{"image_max_batch": 9}``) plus two
private composite keys consumed by ``_enforce_local_validate``:
* ``__template__`` — for AUTOGROW inputs, the rules harvested from
the template's options dict. The enforcer applies these to every
slot value in the runtime Autogrow dict
(``{"image0": <env>, "image1": <env>, ...}``).
* ``__branches__`` — for DYNAMIC_COMBO inputs, a
``{branch_key: {sub_name: rules}}`` map. The enforcer reads the
selected branch from ``value[<name>]`` and walks the matching
branch's sub-inputs (used by GPT Image V2 / Grok ImageEditV2 /
Grok VideoReference, where the cap lives on a nested Autogrow
template inside a DynamicCombo branch).
Inputs whose entire subtree is empty are omitted so the encoder can
use a single ``in`` check per call.
"""
out: dict[str, dict[str, Any]] = {}
input_def = descriptor.get("input") or {}
for kind in ("required", "optional"):
defs = input_def.get(kind) or {}
for name, spec in defs.items():
rules = _rules_from_input_spec(spec)
if rules:
out[name] = rules
return out
def _rules_from_input_spec(spec: Any) -> dict[str, Any]:
"""Extract ``local_validate`` rules from one V3 input spec.
Accepts both wire shapes:
* ``[io_type, options_dict]`` — top-level inputs (the descriptor's
``input.required[name]`` value, or an AUTOGROW template).
* ``[name, io_type, options_dict]`` — DynamicCombo branch sub-inputs
(each entry of ``options.options[i].inputs``).
For ``AUTOGROW`` specs, recurses into ``options["template"]`` and
stashes the result under ``__template__``. For ``DYNAMIC_COMBO``,
walks each branch's sub-inputs and stashes the resulting
``{branch_key: {sub_name: sub_rules}}`` map under ``__branches__``.
"""
if not isinstance(spec, (list, tuple)) or len(spec) < 2:
return {}
if (
len(spec) >= 3
and isinstance(spec[1], str)
and isinstance(spec[2], dict)
):
# [name, io_type, options]
io_type, options = spec[1], spec[2]
elif isinstance(spec[1], dict):
# [io_type, options]
io_type, options = spec[0], spec[1]
else:
return {}
rules: dict[str, Any] = {}
direct = options.get("local_validate")
if isinstance(direct, dict) and direct:
rules.update(direct)
if io_type == "AUTOGROW":
tmpl_spec = options.get("template")
tmpl_rules = (
_rules_from_input_spec(tmpl_spec)
if isinstance(tmpl_spec, (list, tuple))
else {}
)
if tmpl_rules:
rules["__template__"] = tmpl_rules
elif io_type == "DYNAMIC_COMBO":
branches: dict[str, dict[str, Any]] = {}
for opt in (options.get("options") or []):
if not isinstance(opt, dict):
continue
key = opt.get("key")
if not isinstance(key, str):
continue
sub_rules: dict[str, dict[str, Any]] = {}
for sub_spec in (opt.get("inputs") or []):
if (
not isinstance(sub_spec, (list, tuple))
or len(sub_spec) < 1
or not isinstance(sub_spec[0], str)
):
continue
sub_name = sub_spec[0]
r = _rules_from_input_spec(sub_spec)
if r:
sub_rules[sub_name] = r
if sub_rules:
branches[key] = sub_rules
if branches:
rules["__branches__"] = branches
return rules
def _parse_inputs(descriptor: dict[str, Any]) -> tuple[list[Any] | None, list[str]]:
"""Return ``(input_objs, hidden_input_names)``.
@@ -424,34 +580,90 @@ 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."""
"""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`.
"""
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:
common["advanced"] = bool(options.get("advanced"))
primitive_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"), **common)
default=options.get("default"), **primitive_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"), **common)
default=options.get("default"), **primitive_common)
if io_type == "STRING":
return IO.String.Input(
name,
default=options.get("default", ""),
multiline=bool(options.get("multiline", False)),
**common,
**primitive_common,
)
if io_type == "INT":
return IO.Int.Input(
@@ -460,7 +672,7 @@ def _parse_input_spec(name: str, spec: list[Any], optional: bool) -> Any | None:
min=options.get("min", 0),
max=options.get("max", 2147483647),
step=options.get("step", 1),
**common,
**primitive_common,
)
if io_type == "FLOAT":
return IO.Float.Input(
@@ -469,20 +681,160 @@ def _parse_input_spec(name: str, spec: list[Any], optional: bool) -> Any | None:
min=options.get("min", 0.0),
max=options.get("max", 1.0),
step=options.get("step", 0.01),
**common,
**primitive_common,
)
if io_type == "BOOLEAN":
return IO.Boolean.Input(
name, default=bool(options.get("default", False)), **common,
name, default=bool(options.get("default", False)), **primitive_common,
)
if io_type == "IMAGE":
return IO.Image.Input(name, **common)
return IO.Image.Input(name, **primitive_common)
if io_type == "VIDEO":
return IO.Video.Input(name, **common)
return IO.Video.Input(name, **primitive_common)
if io_type == "AUDIO":
return IO.Audio.Input(name, **common)
return IO.Audio.Input(name, **primitive_common)
if io_type == "MASK":
return IO.Mask.Input(name, **common)
return IO.Mask.Input(name, **primitive_common)
if io_type == "MODEL_3D":
# 3D MODEL input: symmetric counterpart of the ``"MODEL_3D"``
# entry in ``_OUTPUT_CLASSES`` above. The Tencent Hunyuan3D
# MODEL_3D-input nodes (``TencentModelTo3DUVNode`` /
# ``Tencent3DTextureEditNode`` / ``Tencent3DPartNode`` /
# ``TencentSmartTopologyNode``) all declare their model_3d
# input as ``IO.MultiType.Input(types=[IO.File3DGLB,
# IO.File3DOBJ, IO.File3DFBX, IO.File3DAny])`` so the socket
# accepts any upstream 3D source: ``Load3D``'s ``File3DAny``
# output, Meshy / Hunyuan3D / Tripo / Rodin's per-format
# outputs (``FILE_3D_GLB`` / ``FILE_3D_OBJ`` / ``FILE_3D_FBX``),
# etc. Mirror that union here so the proxy socket accepts the
# same set of comfytypes upstream does. Per-node format
# validation (e.g. 3DTextureEdit's FBX-only enforcement) is
# done server-side in the provider's ``execute()`` against
# the inline envelope's ``format`` extra; this socket-level
# filter is purely about frontend connection validity.
return IO.MultiType.Input(
name,
types=[
IO.File3DGLB, IO.File3DOBJ, IO.File3DFBX, IO.File3DAny,
],
**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,
)
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),
):
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:
return None
return IO.Autogrow.Input(
name,
template=template,
tooltip=options.get("tooltip"),
optional=optional,
)
# 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 isinstance(io_type, str) and io_type:
return IO.Custom(io_type).Input(name, **common)
return None
@@ -567,23 +919,47 @@ def _parse_outputs(descriptor: dict[str, Any]) -> list[Any] | None:
tip = output_tooltips[i] if i < len(output_tooltips) else None
cls = _OUTPUT_CLASSES.get(io_type)
if cls is None:
log.warning(
"Skipping descriptor: unsupported output type %r", io_type,
)
return 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:
log.warning(
"Skipping descriptor: unsupported output type %r",
io_type,
)
return None
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,
"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,
}
@@ -831,6 +1207,8 @@ async def _execute_remote(
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 {}
@@ -864,11 +1242,18 @@ async def _execute_remote(
# 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)
@@ -1043,8 +1428,54 @@ async def _deserialize_output(
# advisory here, the value already matches the native ComfyUI
# type for STRING / INT / FLOAT / BOOLEAN.
return value
if "data" not in value and "uri" in value:
env_type = value.get("type", "")
encoding = value.get("encoding")
env_type = value.get("type", "")
if encoding == "png_base64_batch_uri":
# Externalised IMAGE batch on the response path (mirror of the
# request-side ``serialization.maybe_externalize`` behaviour).
# No server emits this shape today, but the decoder owns the
# symmetry: any envelope shape the client may externalise on
# the way out, the client must also accept on the way in.
# Per-frame URIs fetched in parallel via ``asyncio.gather`` so
# a 9-frame batch import is one round-trip rather than N — the
# GET fan-out is the dominant latency for an out-of-band batch.
# Frame ordering is preserved (gather returns positional
# results) — critical because the IMAGE tensor's batch axis
# carries semantic order.
policy = url_fetch_policy.get(env_type)
if not isinstance(policy, dict):
raise RnpProtocolError(
f"Server returned a URL batch envelope for output type "
f"{env_type!r} but the descriptor declared no "
f"remote.url_fetch policy for that type",
code=ErrorCode.INTERNAL,
)
uris = value.get("uris")
if not isinstance(uris, list) or not uris:
raise RnpProtocolError(
"png_base64_batch_uri envelope requires a non-empty 'uris' list",
code=ErrorCode.INTERNAL,
)
import asyncio as _asyncio
import base64 as _b64
per_frame_envs = [
{"type": env_type, "encoding": "png_base64", "uri": u}
for u in uris
]
raws = await _asyncio.gather(*[
_fetch_url_envelope(
env, policy,
auth_headers=auth_headers,
node_cls=node_cls,
)
for env in per_frame_envs
])
# Re-stamp as the inline batch shape so the existing
# ``_decode_image_batch_envelope`` path handles it uniformly.
value = {k: v for k, v in value.items() if k != "uris"}
value["encoding"] = "png_base64_batch"
value["frames"] = [_b64.b64encode(r).decode("ascii") for r in raws]
elif "data" not in value and "uri" in value:
policy = url_fetch_policy.get(env_type)
if not isinstance(policy, dict):
raise RnpProtocolError(
@@ -1126,6 +1557,9 @@ async def _encode_inputs(
server_url: str | None = None,
max_inline_bytes: int | None = None,
auth_headers: dict[str, str] | None = None,
input_serialization: dict[str, list[str]] | None = None,
local_validate: dict[str, dict[str, Any]] | None = None,
node_id: str | None = None,
) -> dict[str, Any]:
"""Convert in-place: replace heavy-typed input values with the
matching RNP value envelope. Scalars pass through unchanged.
@@ -1136,40 +1570,240 @@ async def _encode_inputs(
* ``torch.Tensor`` rank 3 with last-dim != 3/4 → mask envelope (B,H,W).
Rank-2 (H,W) tensors are also treated as masks.
* ``dict`` with ``waveform`` + ``sample_rate`` → audio envelope.
* ``VideoInput`` subclass (duck-typed via ``save_to`` +
``get_duration``) → mp4-base64 video envelope.
* Everything else passes through as-is (already JSON-serializable).
When ``max_inline_bytes`` is set, any envelope whose inline payload
exceeds the threshold is externalized via ``POST /rnp/v1/uploads``
(see ``serialization.maybe_externalize``).
``input_serialization`` is the descriptor's per-input encoding
allow-list (e.g. ``{"images": ["png_base64", "png_base64_batch"]}``).
The encoder consults it to decide whether a multi-frame IMAGE input
can ride a single batch envelope vs. falling back to single-frame.
``local_validate`` is the per-input declarative-rule map
(e.g. ``{"images": {"image_max_batch": 9}}``). For each rule the
encoder enforces the constraint locally and raises
``RnpProtocolError(code=INPUT_INVALID, user_facing=True)`` so the
proxy_node's friendly-message dispatch surfaces a per-provider
error before any paid HTTP round trip. Lookup is defensive: a
missing ``local_validate`` entry or missing per-rule key is a
no-op.
"""
serialization_map = input_serialization or {}
validate_map = local_validate or {}
out: dict[str, Any] = {}
for name, value in inputs.items():
encoded = _encode_one(name, value)
# Only envelope-shaped values can be externalized; scalars pass
# straight through ``maybe_externalize`` unchanged.
if serialization.is_envelope(encoded):
encoded = await serialization.maybe_externalize(
encoded,
server_url=server_url,
max_inline_bytes=max_inline_bytes,
auth_headers=auth_headers,
)
_enforce_local_validate(
name, value, validate_map.get(name) or {}, node_id=node_id,
)
encoded = _encode_one(name, value, serialization_map.get(name))
# Externalize any envelope payloads, walking into nested dicts so
# AUTOGROW slot values and DYNAMIC_COMBO sub-inputs whose encoded
# form is a per-slot envelope are uploaded the same way a
# top-level singleton envelope is. ``_externalize_nested`` is a
# no-op on plain scalars and on dicts that contain no envelopes.
encoded = await _externalize_nested(
encoded,
server_url=server_url,
max_inline_bytes=max_inline_bytes,
auth_headers=auth_headers,
)
out[name] = encoded
return out
def _encode_one(name: str, value: Any) -> Any:
async def _externalize_nested(
value: Any,
*,
server_url: str | None,
max_inline_bytes: int | None,
auth_headers: dict[str, str] | None,
) -> Any:
"""Recursively externalize any RNP envelopes inside ``value``.
Mirrors :func:`_encode_one`'s dict-recursion: ComfyUI hands the
proxy_node nested AUTOGROW / DYNAMIC_COMBO runtime dicts
(e.g. ``{"reference_images": {"image1": <env>, "image2": <env>}}``
or ``{"model": {"branch": "wan2.7-r2v",
"video_refs": {"video1": <env>}}}``); without this walk the
per-slot envelopes would ship inline regardless of
``max_inline_bytes`` because :func:`_encode_inputs` only inspected
the top-level value.
Stops at envelope leaves (delegates the per-envelope decision to
:func:`serialization.maybe_externalize`) and at plain scalars (leaves
them unchanged). Lists are intentionally left alone — AG/DC runtime
shapes always materialize as dicts, never lists, and recursing into
arbitrary lists would touch opaque payloads. Recursion is bounded
by the AUTOGROW max slot count (100) and the maximum DC branch
nesting depth (1 today), so an in-place ``async def`` walk is fine.
"""
if serialization.is_envelope(value):
return await serialization.maybe_externalize(
value,
server_url=server_url,
max_inline_bytes=max_inline_bytes,
auth_headers=auth_headers,
)
if isinstance(value, dict):
out: dict[str, Any] = {}
for k, v in value.items():
out[k] = await _externalize_nested(
v,
server_url=server_url,
max_inline_bytes=max_inline_bytes,
auth_headers=auth_headers,
)
return out
return value
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.
Rules currently supported:
* ``image_max_batch`` — for IMAGE inputs (rank-4 tensors), assert
``tensor.shape[0] <= cap``. Mirrors the per-call cap raised on
the server (e.g. Flux2 raises "The current maximum number of
supported images is 9.") so a workflow with N>cap reference
frames fails fast on the client before any paid HTTP round trip.
Two private composite keys are walked transparently:
* ``__template__`` (AUTOGROW) — ``value`` is expected to be a
runtime Autogrow dict (``{"image0": <env>, "image1": <env>, ...}``).
The template's rules are applied to each non-``None`` slot value.
* ``__branches__`` (DYNAMIC_COMBO) — ``value`` is expected to be the
DynamicCombo runtime dict whose ``value[<name>]`` carries the
selected branch key. The matching branch's per-sub-input rules are
applied recursively. This is what makes GPT Image V2 / Grok
ImageEditV2 / Grok VideoReference's nested per-slot caps fire
client-side before any HTTP round trip.
"""
if not rules:
return
_check_image_max_batch(name, value, rules.get("image_max_batch"), node_id=node_id)
tmpl = rules.get("__template__")
if isinstance(tmpl, dict) and isinstance(value, dict):
for slot_key, slot_value in value.items():
if slot_value is None:
continue
_enforce_local_validate(
f"{name}.{slot_key}", slot_value, tmpl, node_id=node_id,
)
branches = rules.get("__branches__")
if isinstance(branches, dict) and isinstance(value, dict):
branch_key = value.get(name)
if isinstance(branch_key, str):
sub_rules_map = branches.get(branch_key)
if isinstance(sub_rules_map, dict):
for sub_name, sub_rules in sub_rules_map.items():
if sub_name in value:
_enforce_local_validate(
f"{name}.{sub_name}", value[sub_name], sub_rules,
node_id=node_id,
)
def _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:
# ComfyUI's V3 ``build_nested_inputs`` rebuilds AUTOGROW / DYNAMIC_COMBO
# runtime kwargs into nested dicts (``{"reference_images": {"image1":
# <tensor>, ...}}`` for top-level AG; ``{"<dc-name>": branch_key,
# "<nested-ag>": {"video1": <VideoInput>, ...}}`` for AG-inside-DC).
# Recurse into plain dicts so heavy leaves anywhere in that tree get
# encoded — without this an AG-of-IMAGE slot would ship the raw
# torch.Tensor inside a dict and crash at ``json.dumps`` time. Stops
# at envelope dicts (already-encoded payloads) so we don't double-
# encode wire-shape values that happen to be dicts.
if serialization.is_envelope(value):
return value
if isinstance(value, dict) and not serialization.is_audio_input(value):
return {
k: _encode_one(f"{name}.{k}", v, accepted_encodings)
for k, v in value.items()
}
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 serialization.is_model3d_input(value):
return serialization.encode_model3d_input(value)
if not serialization._is_torch_tensor(value):
return value
rank = value.dim()
last_dim = int(value.shape[-1]) if rank >= 1 else 0
if rank == 4 or (rank == 3 and last_dim in (3, 4)):
# Per-image tensors: a length-1 batch is the common case; longer
# batches send only the first frame (the framework hasn't asked
# for batch-of-envelopes semantics yet).
return serialization.encode_image_tensor(value)
# Per-image tensors: a length-1 batch is the common case.
# When the descriptor advertises ``"png_base64_batch"`` for
# this input AND the tensor has > 1 frame, ship the multi-
# frame envelope; otherwise fall back to single-frame
# (``tensor[0]`` for batches >1, full tensor for length-1).
accepts_batch = bool(
accepted_encodings and "png_base64_batch" in accepted_encodings
)
return serialization.encode_image_tensor(
value, accepts_batch=accepts_batch,
)
if rank in (2, 3):
return serialization.encode_mask_tensor(value)
log.warning(
+703 -17
View File
@@ -17,6 +17,7 @@ tensor — no parallel dialect.
"""
from __future__ import annotations
import asyncio
import base64
import logging
from io import BytesIO
@@ -56,14 +57,64 @@ async def maybe_externalize(
max_inline_bytes: int | None,
auth_headers: dict[str, str] | None = None,
) -> dict[str, Any]:
"""If the envelope's inline data exceeds ``max_inline_bytes``,
upload the bytes and replace ``data`` with ``uri``.
"""If the envelope's inline payload exceeds ``max_inline_bytes``,
upload it via the presigned-PUT path and rewrite the envelope to
reference the uploaded asset(s) instead of inlining base64.
Two payload shapes are recognised:
* Singleton ``data`` envelopes (the common single-frame IMAGE /
MASK / VIDEO / AUDIO case). Oversize payloads upload as one
asset and the envelope's ``data`` field is swapped for ``uri``.
* Multi-frame ``frames=[…]`` IMAGE envelopes (encoding
``png_base64_batch``). The original ``maybe_externalize`` only
inspected ``data`` and so silently let a 9× 2048² Flux2 batch
ride the wire as a 200 MB POST body — defeating the entire
reason the cap exists. The new path computes the summed raw
byte budget (``sum(len(b64)) * 3 // 4`` — base64 expands 4:3 so
this is the exact payload the server would receive) and, if it
crosses the cap, uploads each frame as an individual presigned
PUT in parallel via :func:`asyncio.gather` then rewrites the
envelope to ``encoding=png_base64_batch_uri`` with a parallel
``uris`` list. ``shape`` / ``dtype`` / ``color_space`` ride
through unchanged so the server-side decoder still sees the
batch dimension. Per-frame URIs (NOT a tar/zip blob) compose
with the existing per-frame fetch helpers and let the server
fetch in parallel.
No-op when either argument is falsy, the envelope already has a
``uri``, or the inline payload fits under the cap.
``uri`` / ``uris``, or the inline payload fits under the cap.
Dispatch keys off the ``encoding`` string — NOT field presence
(``frames`` vs. ``data``). An encoder that emitted ``frames`` for
a non-batch encoding by accident, or shipped both fields together,
used to silently take the batch branch and externalise garbage;
keying off the canonical encoding catches that as a hard ``frames``
check inside the batch branch itself.
"""
if not server_url or not max_inline_bytes:
return envelope
encoding = envelope.get("encoding")
if encoding == "png_base64_batch":
frames = envelope.get("frames")
if not isinstance(frames, list) or not frames:
# Mirror the server-side make_image_batch_envelope contract:
# an empty/missing frames list is a hard wire bug, NOT a
# silent no-op (which used to ship through under the old
# ``isinstance(frames, list) and frames`` dispatch and then
# blow up on the server's resolve_image_envelope_pngs check).
raise RnpProtocolError(
"png_base64_batch envelope requires a non-empty 'frames' list",
code=ErrorCode.INTERNAL,
)
return await _maybe_externalize_batch(
envelope, frames,
server_url=server_url,
max_inline_bytes=max_inline_bytes,
auth_headers=auth_headers,
)
data = envelope.get("data")
if not isinstance(data, str):
return envelope
@@ -84,22 +135,122 @@ async def maybe_externalize(
return out
async def _maybe_externalize_batch(
envelope: dict[str, Any],
frames: list[Any],
*,
server_url: str,
max_inline_bytes: int,
auth_headers: dict[str, str] | None,
) -> dict[str, Any]:
"""Externalise a multi-frame IMAGE envelope when its summed inline
payload overflows ``max_inline_bytes``.
Per-frame upload (NOT a tar/zip blob): each frame becomes its own
presigned-PUT asset so the server can fetch in parallel and the
existing per-frame ``resolve_image_envelope_pngs`` helper picks
up the URI shape transparently. ``asyncio.gather`` issues every
PUT in parallel — the dominant latency for a 9-frame Flux2 batch
is the upload, not the server hop, so serialising would multiply
end-to-end submit time by N. Frame ordering is preserved (gather
returns positional results) — critical because a Flux2 provider
maps ``uris[k]`` → ``input_image_{k+1}`` 1:1.
"""
# Cap on raw byte count (base64 expands 4:3, so multiplying by
# 3/4 recovers the actual payload the server would receive).
total_raw = 0
for f in frames:
if isinstance(f, str):
total_raw += len(f) * 3 // 4
if total_raw <= max_inline_bytes:
return envelope
env_type = envelope.get("type", "")
content_type = _CONTENT_TYPE_FOR_TYPE.get(env_type, "application/octet-stream")
async def _upload_one(idx: int, frame_b64: Any) -> str:
if not isinstance(frame_b64, str):
raise RnpProtocolError(
f"image batch frame {idx} is not a base64 string "
f"(got {type(frame_b64).__name__})",
code=ErrorCode.INTERNAL,
)
raw = base64.b64decode(frame_b64)
return await rnp_client.upload_asset(
server_url, raw,
content_type=content_type,
auth_headers=auth_headers,
)
uris = await asyncio.gather(
*(_upload_one(i, f) for i, f in enumerate(frames))
)
log.info(
"RNP: externalized %s batch envelope (%d frames, %d raw bytes) "
"-> %d presigned URIs",
env_type, len(frames), total_raw, len(uris),
)
# Strip the inline ``frames`` field; preserve ``shape`` / ``dtype``
# / ``color_space`` so the server-side decoder still sees the
# batch dimension and per-frame metadata. Encoding flips to the
# URI variant so dispatch keys cleanly off the string (NOT field
# presence — frames vs. uris is a payload field, not a shape
# selector).
out: dict[str, Any] = {
k: v for k, v in envelope.items() if k != "frames"
}
out["encoding"] = "png_base64_batch_uri"
out["uris"] = list(uris)
return out
# ---------------------------------------------------------------------------
# Image encode / decode
# ---------------------------------------------------------------------------
def encode_image_tensor(tensor: Any) -> dict[str, Any]:
"""Encode a ComfyUI IMAGE tensor (B,H,W,C float32 in 0..1) as a
PNG-base64 image envelope.
def encode_image_tensor(tensor: Any, *, accepts_batch: bool = False) -> dict[str, Any]:
"""Encode a ComfyUI IMAGE tensor (B,H,W,C float32 in 0..1) as an
image envelope.
Multi-image batches send only the first frame for now; the
framework hasn't grown batch-of-envelopes semantics yet.
``accepts_batch`` reflects whether the descriptor's
``input_serialization`` for this input lists ``"png_base64_batch"``.
When True AND ``tensor.shape[0] > 1``, emit a multi-frame
``png_base64_batch`` envelope (``frames=[<png_b64>, …]``) so the
server decodes a real ``[B, H, W, C]`` source. Otherwise emit a
single-frame ``png_base64`` envelope for ``tensor[0]`` — preserves
backwards compat with descriptors that haven't migrated yet, and
matches the wire shape of the existing ``_decode_image_batch_envelope``
on the response path so both directions round-trip symmetrically.
"""
import numpy as np
from PIL import Image
# Strip the batch dim if present.
if hasattr(tensor, "dim") and tensor.dim() == 4:
rank = getattr(tensor, "dim", lambda: 0)()
if rank == 4 and accepts_batch and int(tensor.shape[0]) > 1:
frames_arr = (tensor.cpu().clamp(0.0, 1.0).numpy() * 255).astype(np.uint8)
b, height, width = frames_arr.shape[:3]
channels = frames_arr.shape[3] if frames_arr.ndim == 4 else 1
pil_mode = {1: "L", 3: "RGB", 4: "RGBA"}.get(channels, "RGB")
frames_b64: list[str] = []
for i in range(b):
img = Image.fromarray(frames_arr[i], mode=pil_mode)
buf = BytesIO()
img.save(buf, format="PNG")
frames_b64.append(base64.b64encode(buf.getvalue()).decode("ascii"))
return {
"type": "image",
"encoding": "png_base64_batch",
"frames": frames_b64,
"shape": [b, height, width, channels],
"dtype": "uint8",
"color_space": "srgb",
}
# Single-frame fallback: matches the old ``encode_image_tensor``
# contract (sends ``tensor[0]`` for a multi-frame batch when the
# descriptor doesn't advertise the batch encoding).
if rank == 4:
frame = tensor[0]
else:
frame = tensor
@@ -121,8 +272,18 @@ def encode_image_tensor(tensor: Any) -> dict[str, Any]:
def decode_image_envelope(envelope: dict[str, Any]) -> Any:
"""Decode an image envelope into a ComfyUI IMAGE tensor (B,H,W,C float32)."""
"""Decode an image envelope into a ComfyUI IMAGE tensor.
Single-frame encodings (``png_base64``, ``jpeg_base64``,
``webp_base64``) decode into ``[1, H, W, C]``. The multi-frame
``png_base64_batch`` encoding (a list of base64-PNG ``frames``)
stacks each frame into ``[B, H, W, C]`` — the native ComfyUI
image-batch shape — so providers whose upstream returns ``n > 1``
surface as a real batched tensor in the workflow runtime.
"""
encoding = envelope.get("encoding")
if encoding == "png_base64_batch":
return _decode_image_batch_envelope(envelope)
if encoding not in ("png_base64", "jpeg_base64", "webp_base64"):
raise RnpProtocolError(
f"Unsupported image encoding: {encoding!r}",
@@ -141,6 +302,69 @@ def decode_image_envelope(envelope: dict[str, Any]) -> Any:
return torch.from_numpy(arr).unsqueeze(0)
def _decode_image_batch_envelope(envelope: dict[str, Any]) -> Any:
"""Decode a ``png_base64_batch`` envelope into ``[B, H, W, C]``.
The wire shape is ``{"type":"image", "encoding":"png_base64_batch",
"frames":[<png_b64>, ...], "shape":[B,H,W,C], ...}``. Each frame is
a PNG-encoded image; all frames must share the same H/W after the
RGB conversion (the server-side encoder enforces this). A frame
count mismatch against the ``shape[0]`` field surfaces as a hard
parse error rather than silently dropping frames — matches the SVG
decoder's ``count`` integrity check.
"""
import base64 as _b64
import numpy as np
import torch
from PIL import Image
frames = envelope.get("frames")
if not isinstance(frames, list) or not frames:
raise RnpProtocolError(
"image batch envelope missing or empty 'frames' list",
code=ErrorCode.INTERNAL,
)
declared_shape = envelope.get("shape")
if isinstance(declared_shape, list) and len(declared_shape) >= 1:
declared_b = int(declared_shape[0])
if declared_b != len(frames):
raise RnpProtocolError(
f"image batch envelope frame count {len(frames)} does not "
f"match declared shape[0]={declared_b}",
code=ErrorCode.INTERNAL,
)
arrs: list[Any] = []
base_hw: tuple[int, int] | None = None
for idx, frame_b64 in enumerate(frames):
if not isinstance(frame_b64, str):
raise RnpProtocolError(
f"image batch frame {idx} is not a base64 string",
code=ErrorCode.INTERNAL,
)
try:
raw = _b64.b64decode(frame_b64)
img = Image.open(BytesIO(raw)).convert("RGB")
except Exception as e:
raise RnpProtocolError(
f"image batch frame {idx} failed to decode: {e}",
code=ErrorCode.INTERNAL,
) from e
arr = np.array(img).astype(np.float32) / 255.0
if base_hw is None:
base_hw = (arr.shape[0], arr.shape[1])
elif (arr.shape[0], arr.shape[1]) != base_hw:
raise RnpProtocolError(
f"image batch frame {idx} size {arr.shape[:2]} does not "
f"match frame 0 size {base_hw} (server should have rejected "
f"this; likely a wire bug)",
code=ErrorCode.INTERNAL,
)
arrs.append(arr)
stacked = np.stack(arrs, axis=0)
return torch.from_numpy(stacked)
# ---------------------------------------------------------------------------
# Mask encode / decode
# ---------------------------------------------------------------------------
@@ -227,7 +451,11 @@ def encode_audio_input(audio: dict[str, Any]) -> dict[str, Any]:
def decode_audio_envelope(envelope: dict[str, Any]) -> dict[str, Any]:
"""Decode an audio envelope back into a ComfyUI AUDIO dict."""
encoding = envelope.get("encoding")
if encoding not in ("mp3_base64", "mp3_inline", "wav_base64", "wav_inline"):
if encoding not in (
"mp3_base64", "mp3_inline",
"wav_base64", "wav_inline",
"opus_base64",
):
raise RnpProtocolError(
f"Unsupported audio encoding: {encoding!r}",
code=ErrorCode.INTERNAL,
@@ -237,9 +465,44 @@ 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 VIDEO (``VideoInput`` subclass) as an mp4-base64
video envelope.
Mirrors the AUDIO encoder above: re-encodes the Video object to an
in-memory mp4/H.264 byte buffer (matches upstream
``comfy_api_nodes/util/conversions.py`` ``video_to_base64_string``)
and base64-encodes it. Populates ``duration_s`` metadata when the
Video object exposes ``get_duration()`` so server-side providers
(e.g. ``GrokVideoExtendProvider._validate_video_duration_envelope``)
can range-check without demuxing the MP4.
"""
from comfy_api_nodes.util.conversions import video_to_base64_string
b64 = video_to_base64_string(video)
extra: dict[str, Any] = {}
duration_s: float | None = None
try:
getter = getattr(video, "get_duration", None)
if callable(getter):
d = getter()
if d is not None:
duration_s = float(d)
except Exception:
duration_s = None
if duration_s is not None:
extra["duration_s"] = duration_s
return {
"type": "video",
"encoding": "mp4_base64",
"data": b64,
**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")
@@ -252,15 +515,391 @@ def decode_video_envelope(envelope: dict[str, Any]) -> Any:
return InputImpl.VideoFromFile(BytesIO(decode_envelope_data(envelope)))
# ---------------------------------------------------------------------------
# 3D model encode / decode
# ---------------------------------------------------------------------------
def encode_model3d_input(model_3d: Any) -> dict[str, Any]:
"""Encode a ComfyUI ``File3D`` (or compatible) as an inline
single-file 3D-model envelope.
Mirrors the AUDIO / VIDEO encoders above: reads the primary mesh
bytes (via the ``File3D`` API: ``get_data().read()`` returns a
``BytesIO`` payload; falling back to ``read_bytes()`` for older
sources) and base64-encodes them. The envelope's ``encoding`` is
``"glb_inline"`` (the generic single-file inline encoding the
server-side decoder dispatches on at :func:`decode_model3d_envelope`
— the decoder hands the bytes to ``File3D(BytesIO(...), format)``
regardless of whether the actual format is glb / obj / fbx / etc.,
so the ``encoding`` name is a logical "inline single-file 3D" tag
and the ``format`` extra picks the file extension).
Multi-file bundles (``BundledFile3D``-style values where every
companion file matters — e.g. an OBJ with a sibling .mtl + texture
PNGs) are not yet supported on the encode path: this PR ships the
minimal wire shape needed for the Tencent Hunyuan3D MODEL_3D-input
nodes which all upload a single file via ``upload_3d_model_to_comfyapi``
upstream. Bundle encode (``bundle_inline``) lands when a partner
node actually needs the companion bytes.
"""
file_format = ""
fmt_attr = getattr(model_3d, "format", None)
if isinstance(fmt_attr, str):
file_format = fmt_attr.lower()
# ``File3D.get_data()`` returns a ``BytesIO`` (per
# ``comfy_api.latest._util.geometry_types.File3D``). Other File3D-
# compatible classes may instead expose ``read_bytes`` returning
# raw bytes; fall back to that for compatibility.
raw: bytes
get_data = getattr(model_3d, "get_data", None)
if callable(get_data):
buf = get_data()
if hasattr(buf, "seek"):
buf.seek(0)
raw = buf.read() if hasattr(buf, "read") else bytes(buf)
else:
read_bytes = getattr(model_3d, "read_bytes", None)
if callable(read_bytes):
raw = read_bytes()
else:
raise RnpProtocolError(
f"unsupported 3D model value: {type(model_3d).__name__!r} "
f"has no get_data() or read_bytes()",
code=ErrorCode.INTERNAL,
)
envelope: dict[str, Any] = {
"type": "model_3d",
"encoding": "glb_inline",
"data": base64.b64encode(raw).decode("ascii"),
}
if file_format:
envelope["format"] = file_format
return envelope
# ---------------------------------------------------------------------------
# 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.
"""
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
def decode_model3d_envelope(envelope: dict[str, Any]) -> Any:
"""Decode a 3D-model envelope into a ComfyUI ``File3D`` object.
Dispatches on the envelope's ``encoding``:
* ``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``.
"""
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,
)
# ---------------------------------------------------------------------------
# Dispatch by envelope type
# ---------------------------------------------------------------------------
_DECODERS = {
"image": decode_image_envelope,
"mask": decode_mask_envelope,
"audio": decode_audio_envelope,
"video": decode_video_envelope,
"image": decode_image_envelope,
"mask": decode_mask_envelope,
"audio": decode_audio_envelope,
"video": decode_video_envelope,
"model_3d": decode_model3d_envelope,
}
@@ -293,6 +932,47 @@ def is_audio_input(value: Any) -> bool:
)
def is_video_input(value: Any) -> bool:
"""Best-effort check for a ComfyUI ``VideoInput`` subclass.
Duck-types on the ``save_to`` + ``get_duration`` method pair: both
are defined on the ``VideoInput`` abstract base in
``comfy_api.latest._input`` and present on every concrete subclass
(``VideoFromFile`` / ``VideoFromComponents``). Avoids importing
``VideoInput`` at module load (which would pull torch via the
``comfy_api`` tree).
"""
if isinstance(value, dict):
return False
return (
callable(getattr(value, "save_to", None))
and callable(getattr(value, "get_duration", None))
)
def is_model3d_input(value: Any) -> bool:
"""Best-effort check for a ComfyUI ``File3D`` (or compatible).
Duck-types on the ``format`` string attribute + ``get_data``
callable: both are present on the ``File3D`` class in
``comfy_api.latest._util.geometry_types`` and the
``BundledFile3D`` subclass built lazily in
:func:`_bundled_file3d_class`. Avoids importing ``File3D`` at
module load (which would pull torch via the ``comfy_api`` tree).
The ``VideoInput`` shape (``save_to`` + ``get_duration``) does not
expose a ``format`` string property, so VIDEO and MODEL_3D values
don't collide; caller (:func:`_encode_one`) still dispatches
AUDIO / VIDEO before MODEL_3D as defense-in-depth.
"""
if isinstance(value, dict):
return False
return (
isinstance(getattr(value, "format", None), str)
and callable(getattr(value, "get_data", None))
)
def _is_torch_tensor(value: Any) -> bool:
try:
import torch
@@ -309,10 +989,16 @@ __all__ = [
"decode_mask_envelope",
"encode_audio_input",
"decode_audio_envelope",
"encode_video_input",
"decode_video_envelope",
"encode_model3d_input",
"decode_model3d_envelope",
"_bundled_file3d_class",
"decode_envelope",
"is_envelope",
"is_image_tensor",
"is_mask_tensor",
"is_audio_input",
"is_video_input",
"is_model3d_input",
]