"""Build dynamic V3 ``IO.ComfyNode`` subclasses from RNP NodeDescriptors. The descriptor wire format is V3 ``Schema.get_v1_info()`` verbatim, so per-input dicts are handed straight to V3's IO classes without reinterpretation. ``_parse_input_spec`` knows STRING / INT / FLOAT / BOOLEAN / COMBO / IMAGE / VIDEO / AUDIO / MASK plus the two V3 dynamic input types DYNAMIC_COMBO (branch-selector dropdown with per-branch sub-widgets) and AUTOGROW (variable-arity slot template); any other io_type string falls through to ``IO.Custom(io_type)`` (the §B opaque pass-through bucket — partner helper-config types like RECRAFT_*, OPENAI_INPUT_FILES, OPENAI_CHAT_CONFIG, GEMINI_INPUT_FILES round-trip as raw JSON blobs with the original io_type preserved on the V3 socket, so node-to-node connections only chain when the strings match). Only a non-string / empty io_type causes a skip. Malformed DYNAMIC_COMBO / AUTOGROW specs skip the descriptor (return ``None``) rather than degrading to a single opaque socket. Hidden inputs (``auth_token_comfy_org`` / ``api_key_comfy_org`` / ``unique_id`` / ``prompt`` / ``extra_pnginfo`` / ``dynprompt``) are declared on the schema so the executor populates ``cls.hidden``; auth credentials are then lifted into outbound HTTP headers and everything else flows into the request ``context`` map. Only ``execute`` is wired today; the validate / fingerprint / check_lazy_status RNP methods fall through to ComfyUI's defaults until their server endpoints exist. """ from __future__ import annotations import asyncio import contextlib import logging import math import time import uuid from typing import Any from comfy_api.latest import IO from server import PromptServer from comfy_api_nodes.util.client import ApiEndpoint, poll_op_raw from comfy_api_nodes.util.common_exceptions import ProcessingInterrupted from . import client as rnp_client from . import serialization from .protocol import ( DEFAULT_CANCEL_TIMEOUT_S, 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, DEFAULT_TIMEOUT_PER_POLL_S, ErrorCode, ExecutionMode, Header, PROTOCOL_VERSION, RnpProtocolError, TaskStatus, is_envelope, ) # ComfyUI's frontend matches these strings literally on # ``exception_message.includes(...)`` to fire dedicated dialogs, so the # wording here must stay byte-identical with # ``comfy_api_nodes.util.client._friendly_http_message``. _AUTH_FAILED_MESSAGE = "Unauthorized: Please login first to use this node." _INSUFFICIENT_CREDITS_MESSAGE = ( "Payment Required: Please add credits to your account to use this node." ) _ACCOUNT_PROBLEM_MESSAGE = ( "There is a problem with your account. Please contact support@comfy.org." ) _RATE_LIMIT_AFTER_RETRY_MESSAGE = ( "Rate Limit Exceeded: The server returned 429 after all retry attempts. " "Please wait and try again." ) _PROVIDER_UNAVAILABLE_MESSAGE = ( "The remote node provider is temporarily unavailable. Please try again " "in a few minutes." ) _SERVER_BUSY_MESSAGE = ( "The remote node server is at capacity. Please try again in a moment." ) _MAINTENANCE_MESSAGE = ( "The remote node server is in maintenance. Please try again shortly." ) _TASK_LOST_MESSAGE = ( "The remote node server lost the running task and the resume attempt " "did not succeed. Please re-run the workflow." ) def _to_friendly_message(e: RnpProtocolError, node_id: str) -> str: upstream = (e.details or {}).get("upstream_status") if e.details else None if e.code == ErrorCode.AUTH_FAILED or upstream in (401, 403): return _AUTH_FAILED_MESSAGE if e.code == ErrorCode.INSUFFICIENT_CREDITS or upstream == 402: return _INSUFFICIENT_CREDITS_MESSAGE # 409 from the upstream gateway means "account in bad standing". # comfy_api_nodes.util.client._friendly_http_message uses a dedicated # support-contact prompt for this; mirror it byte-identically so the # same frontend dialog fires. if upstream == 409: return _ACCOUNT_PROBLEM_MESSAGE # 429 reaches the client only after the server has exhausted its own # rate-limit retries — surface the post-retry copy so users don't # think a single transient 429 leaked through. if e.code == ErrorCode.RATE_LIMITED or upstream == 429: return _RATE_LIMIT_AFTER_RETRY_MESSAGE if e.code == ErrorCode.SERVER_BUSY: return _SERVER_BUSY_MESSAGE if e.code == ErrorCode.MAINTENANCE: # Server-friendly message often embeds an "until" timestamp in # the wire ``message`` — prefer it over the canned copy. return str(e) if e.user_facing else _MAINTENANCE_MESSAGE if e.code == ErrorCode.TASK_LOST: return _TASK_LOST_MESSAGE # PROVIDER_UNAVAILABLE with a retryable hint or 5xx upstream # collapses to the generic "try again" copy. if e.code == ErrorCode.PROVIDER_UNAVAILABLE and ( upstream in (502, 503, 504) or e.retryable ): return _PROVIDER_UNAVAILABLE_MESSAGE if e.user_facing: return str(e) return f"Remote node {node_id} failed ({e.code})" def _coerce_pos(value: Any, default: float) -> float: """Best-effort numeric coerce — falls back to ``default`` for anything unparseable or non-positive (the descriptor came over the wire so we don't trust it).""" try: f = float(value) except (TypeError, ValueError): return float(default) return f if f > 0 else float(default) def _coerce_pos_int(value: Any, default: int) -> int: try: i = int(value) except (TypeError, ValueError): return int(default) return i if i > 0 else int(default) def _extract_execution_policy(execution: dict[str, Any]) -> dict[str, Any]: """Lift ``descriptor.remote.execution`` policy fields into a flat dict of ``poll_op_raw`` kwargs. Defaults mirror ``comfy_api_nodes.util.client.poll_op_raw`` so an RNP 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``, ``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": 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, ), "cancel_timeout_s": _coerce_pos( execution.get("cancel_timeout_s"), DEFAULT_CANCEL_TIMEOUT_S, ), "max_retries_per_poll": _coerce_pos_int( execution.get("max_retries_per_poll"), DEFAULT_MAX_RETRIES_PER_POLL, ), "retry_delay_per_poll_s": _coerce_pos( execution.get("retry_delay_per_poll_s"), DEFAULT_RETRY_DELAY_PER_POLL_S, ), "retry_backoff_per_poll": _coerce_pos( execution.get("retry_backoff_per_poll"), DEFAULT_RETRY_BACKOFF_PER_POLL, ), # Submit-side retry block — currently unused by execute_remote / # execute_remote_async (they hard-code max_retries=2 in # rnp_client) but lifted out here so future steps can plumb it # through without re-parsing the descriptor. "submit_retry": { "max": _coerce_pos_int(retry_block.get("max"), 0) if retry_block.get("max") is not None else 0, "delay_s": _coerce_pos(retry_block.get("delay_s"), 1.0), "backoff": _coerce_pos(retry_block.get("backoff"), 2.0), "retry_on": list(retry_block.get("retry_on") or []), }, "idempotency": execution.get("idempotency") or "none", } def _node_id_for_ticker(cls: type[IO.ComfyNode]) -> str | None: """Lift ``cls.hidden.unique_id`` for the per-second progress ticker. Returns ``None`` if the hidden holder isn't populated (test paths that build a class but don't run it through the executor). """ hidden = getattr(cls, "hidden", None) if hidden is None: return None uid = getattr(hidden, "unique_id", None) return str(uid) if uid is not None else None async def _progress_ticker( node_id: str | None, estimated_duration_s: int | float | None, *, status_override: dict[str, Any] | None = None, ) -> None: """Emit one ``send_progress_text`` per second until cancelled. Format mirrors ``comfy_api_nodes.util.client._display_time_progress`` so the progress badge looks identical to local partner nodes: ``Status: Processing\\nTime elapsed: s`` (with ``(~Ns remaining)`` when the descriptor declared an estimate). ``status_override`` is a shared mutable holder ``{"label": str|None, "expires_at": float|None}`` the caller can mutate to swap "Processing" for a transient message like "Server busy, retrying in 30s..." while a backpressure 503 is being waited out. The ticker auto-clears the override once ``time.monotonic() >= expires_at`` so a stale label doesn't linger if the post-wait request takes longer than the announced Retry-After window. """ if node_id is None: return started = time.monotonic() while True: elapsed = int(time.monotonic() - started) if estimated_duration_s is not None and estimated_duration_s > 0: remaining = max(0, int(estimated_duration_s) - elapsed) time_line = f"Time elapsed: {elapsed}s (~{remaining}s remaining)" else: time_line = f"Time elapsed: {elapsed}s" label = "Processing" if status_override is not None: expires_at = status_override.get("expires_at") if expires_at is not None and time.monotonic() >= expires_at: status_override["label"] = None status_override["expires_at"] = None override = status_override.get("label") if override: label = override text = f"Status: {label}\n{time_line}" with contextlib.suppress(Exception): instance = PromptServer.instance if instance is not None: instance.send_progress_text(text, node_id) try: await asyncio.sleep(1.0) except asyncio.CancelledError: return log = logging.getLogger("comfy_remote_nodes.proxy_node") def build_node_class( node_id: str, descriptor: dict[str, Any], server_url: str, *, max_inline_bytes: int | None = None, ) -> type[IO.ComfyNode] | None: """Build an ``IO.ComfyNode`` subclass for one descriptor. Returns ``None`` if the descriptor can't be parsed (e.g. an input type ``_parse_input_spec`` doesn't recognise). The caller logs and skips the node so one bad descriptor doesn't break the whole extension. """ remote = descriptor.get("remote") or {} if not isinstance(remote, dict): log.warning("Descriptor %s missing 'remote' block — skipping", node_id) return None endpoints = remote.get("endpoints") or {} execute_endpoint = endpoints.get("execute") if not isinstance(execute_endpoint, dict) or not execute_endpoint.get("path"): log.warning("Descriptor %s has no remote.endpoints.execute — skipping", node_id) return None inputs, hidden_names = _parse_inputs(descriptor) if inputs is None: return None outputs = _parse_outputs(descriptor) if outputs is None: return None 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: }`` — 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.
.[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") or 600.0 ) estimated_duration_s = execution.get("estimated_duration_s") if not isinstance(estimated_duration_s, (int, float)): estimated_duration_s = None execution_mode = execution.get("mode") or ExecutionMode.REQUEST_RESPONSE execution_policy = _extract_execution_policy(execution) # ``descriptor.remote.url_fetch`` declares per-output-type fetch # posture for envelopes that arrive with ``uri`` instead of inline # ``data``. Empty / absent → all outputs must inline their bytes. url_fetch_policy = ( remote.get("url_fetch") if isinstance(remote.get("url_fetch"), dict) else {} ) display_name = descriptor.get("display_name") or node_id category = descriptor.get("category") or "remote" description = descriptor.get("description") or "" is_api_node = bool(descriptor.get("api_node", False)) is_experimental = bool(descriptor.get("experimental", False)) is_deprecated = bool(descriptor.get("deprecated", False)) schema_inputs = list(inputs) schema_outputs = list(outputs) hidden_decls = _hidden_decls_from_names(hidden_names, is_api_node) # Forward every other V3 NodeInfoV1 field the descriptor surfaces. # Anything not constructed here is silently dropped; the # ``_parse_input_spec`` / ``_parse_outputs`` warnings are the # signal a deployment needs to add a missing type to this builder. is_input_list = bool(descriptor.get("is_input_list", False)) is_output_node = bool(descriptor.get("output_node", False)) is_dev_only = bool(descriptor.get("dev_only", False)) has_intermediate_output = bool(descriptor.get("has_intermediate_output", False)) search_aliases = descriptor.get("search_aliases") or [] essentials_category = descriptor.get("essentials_category") price_badge = _build_price_badge(descriptor.get("price_badge")) class RemoteProxyNode(IO.ComfyNode): @classmethod def define_schema(cls) -> IO.Schema: return IO.Schema( node_id=node_id, display_name=display_name, category=category, description=description, inputs=schema_inputs, outputs=schema_outputs, hidden=hidden_decls, is_input_list=is_input_list, is_output_node=is_output_node, is_api_node=is_api_node, is_experimental=is_experimental, is_deprecated=is_deprecated, is_dev_only=is_dev_only, has_intermediate_output=has_intermediate_output, search_aliases=list(search_aliases), essentials_category=essentials_category, price_badge=price_badge, ) @classmethod async def execute(cls, **kwargs: Any) -> IO.NodeOutput: return await _execute_remote( cls, node_id=node_id, server_url=server_url, execute_path=execute_path, schema_hash=schema_hash, timeout_s=timeout_s, outputs_meta=schema_outputs, hidden_names=hidden_names, inputs=kwargs, max_inline_bytes=max_inline_bytes, estimated_duration_s=estimated_duration_s, execution_mode=execution_mode, execution_policy=execution_policy, url_fetch_policy=url_fetch_policy, input_serialization=input_serialization, local_validate=local_validate, ) RemoteProxyNode.__name__ = node_id RemoteProxyNode.__qualname__ = node_id RemoteProxyNode.RELATIVE_PYTHON_MODULE = "custom_nodes.comfy_remote_nodes" return RemoteProxyNode # --------------------------------------------------------------------------- # Input parsing # --------------------------------------------------------------------------- # The hidden input markers V3 emits in get_v1_info() — clients forward # their values through the request ``context`` instead of treating them # as user-facing inputs. _HIDDEN_MARKERS = { "AUTH_TOKEN_COMFY_ORG": IO.Hidden.auth_token_comfy_org, "API_KEY_COMFY_ORG": IO.Hidden.api_key_comfy_org, "UNIQUE_ID": IO.Hidden.unique_id, "PROMPT": IO.Hidden.prompt, "EXTRA_PNGINFO": IO.Hidden.extra_pnginfo, "DYNPROMPT": IO.Hidden.dynprompt, } 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": , "image1": , ...}``). * ``__branches__`` — for DYNAMIC_COMBO inputs, a ``{branch_key: {sub_name: rules}}`` map. The enforcer reads the selected branch from ``value[]`` 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)``. Returns ``(None, [])`` if any required/optional input can't be parsed (caller logs and skips). """ input_def = descriptor.get("input") or {} input_order = descriptor.get("input_order") or {} out: list[Any] = [] for kind, optional in (("required", False), ("optional", True)): order = input_order.get(kind) or list((input_def.get(kind) or {}).keys()) defs = input_def.get(kind) or {} for name in order: spec = defs.get(name) if spec is None: continue inp = _parse_input_spec(name, spec, optional=optional) if inp is None: log.warning( "Skipping descriptor: unsupported input %r (%s)", name, spec, ) return None, [] out.append(inp) hidden_names = list((input_def.get("hidden") or {}).keys()) return out, hidden_names def _parse_input_spec(name: str, spec: list[Any], optional: bool) -> Any | None: """Map one V3-shaped ``[io_type, options_dict]`` entry to a V3 IO Input. This is the top-level entry used by ``_parse_inputs`` (where the name comes from the descriptor's ``input.required[name]`` / ``input.optional[name]`` key) and recursively as the AUTOGROW template entry (which also lacks a leading name field — the slot name is provided by the surrounding AUTOGROW). Branch sub-inputs inside a DYNAMIC_COMBO option use the 3-tuple ``[name, io_type, options]`` form instead — see :func:`_parse_named_input_spec`. """ if not isinstance(spec, (list, tuple)) or len(spec) < 1: return None io_type = spec[0] options: dict[str, Any] = spec[1] if len(spec) > 1 and isinstance(spec[1], dict) else {} return _build_input(name, io_type, options, optional) def _parse_named_input_spec(spec: list[Any], optional: bool) -> Any | None: """Map one V3-shaped ``[name, io_type, options_dict]`` entry to a V3 IO Input. This is the shape used inside ``DYNAMIC_COMBO`` options' ``inputs`` list (each branch sub-input carries its own name as ``spec[0]``). Top-level descriptor inputs use the 2-tuple form handled by :func:`_parse_input_spec`. """ if ( not isinstance(spec, (list, tuple)) or len(spec) < 2 or not isinstance(spec[0], str) ): return None name = spec[0] io_type = spec[1] options: dict[str, Any] = spec[2] if len(spec) > 2 and isinstance(spec[2], dict) else {} return _build_input(name, io_type, options, optional) def _build_input(name: str, io_type: Any, options: dict[str, Any], optional: bool) -> Any | None: """Shared dispatch: construct one V3 IO Input from a parsed ``(name, io_type, options)`` triple. Recurses for the two V3 dynamic input types: * ``DYNAMIC_COMBO`` — selector dropdown whose per-branch sub-inputs live in ``options.options[i].inputs`` (each entry is the 3-tuple ``[name, io_type, options]`` form). * ``AUTOGROW`` — variable-arity slot template. The RNP wire dialect flattens upstream's ``Autogrow.Input.as_dict`` shape: ``template`` is a 2-tuple ``[io_type, options]`` entry directly, with ``prefix`` / ``min`` / ``max`` (or ``names`` / ``min``) as siblings of ``template`` rather than nested under ``template.template``. Malformed dynamic specs return ``None`` (caller logs + skips the whole descriptor) rather than falling through to the opaque ``IO.Custom`` bucket — that would mask a protocol/schema bug as "looks fine but renders as opaque socket". """ common = { "tooltip": options.get("tooltip"), "optional": optional, } # ``advanced`` is only accepted by WidgetInput subclasses — the # dynamic constructors (DynamicCombo.Input / Autogrow.Input) do # not take it. Build a separate kwargs bag for primitives. primitive_common = dict(common) if options.get("advanced") is not None: primitive_common["advanced"] = bool(options.get("advanced")) if isinstance(io_type, list): # Legacy V1-style combo: io_type *is* the options list. return IO.Combo.Input(name, options=list(io_type), default=options.get("default"), **primitive_common) if io_type == "COMBO": opts = options.get("options") if not isinstance(opts, (list, tuple)): return None return IO.Combo.Input(name, options=list(opts), default=options.get("default"), **primitive_common) if io_type == "STRING": return IO.String.Input( name, default=options.get("default", ""), multiline=bool(options.get("multiline", False)), **primitive_common, ) if io_type == "INT": return IO.Int.Input( name, default=options.get("default", 0), min=options.get("min", 0), max=options.get("max", 2147483647), step=options.get("step", 1), **primitive_common, ) if io_type == "FLOAT": return IO.Float.Input( name, default=options.get("default", 0.0), min=options.get("min", 0.0), max=options.get("max", 1.0), step=options.get("step", 0.01), **primitive_common, ) if io_type == "BOOLEAN": return IO.Boolean.Input( name, default=bool(options.get("default", False)), **primitive_common, ) if io_type == "IMAGE": return IO.Image.Input(name, **primitive_common) if io_type == "VIDEO": return IO.Video.Input(name, **primitive_common) if io_type == "AUDIO": return IO.Audio.Input(name, **primitive_common) if io_type == "MASK": 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 def _build_price_badge(wire: Any) -> Any | None: """Convert the wire-format ``price_badge`` dict back to an ``IO.PriceBadge`` instance. The wire shape (from ``PriceBadge.as_dict()``) carries ``depends_on.widgets`` as a list of ``{name, type}`` objects so the frontend can resolve types; the V3 constructor only needs the ``name`` strings since it re-derives types from the schema's actual inputs at ``as_dict()`` time. Inputs / input_groups round-trip as-is. Returns ``None`` if the descriptor has no ``price_badge`` field. """ if not isinstance(wire, dict): return None expr = wire.get("expr") if not isinstance(expr, str) or not expr.strip(): return None dep = wire.get("depends_on") or {} widgets = [ w["name"] for w in (dep.get("widgets") or []) if isinstance(w, dict) and "name" in w ] inputs = list(dep.get("inputs") or []) input_groups = list(dep.get("input_groups") or []) engine = wire.get("engine", "jsonata") try: return IO.PriceBadge( expr=expr, depends_on=IO.PriceBadgeDepends( widgets=widgets, inputs=inputs, input_groups=input_groups, ), engine=engine, ) except Exception as e: # noqa: BLE001 log.warning("Skipping price_badge — could not construct: %s", e) return None def _hidden_decls_from_names(hidden_names: list[str], is_api_node: bool) -> list[Any]: """Build the ``hidden=`` list for IO.Schema. The descriptor lists each hidden input by name (e.g. ``auth_token_comfy_org``); the ``_HIDDEN_MARKERS`` table above resolves those names back to V3 ``IO.Hidden`` enum members. Names that aren't in the table are dropped with a debug log so a server that adds a new hidden marker doesn't break older clients. """ out: list[Any] = [] seen: set[Any] = set() for name in hidden_names: marker = getattr(IO.Hidden, name, None) if marker is None: log.debug("Unknown hidden input %s — dropping", name) continue if marker not in seen: out.append(marker) seen.add(marker) # IO.Schema.finalize() will add auth/api hidden if is_api_node, so # we don't need to add them again here. if not is_api_node: # Non-api-node descriptors that happened to advertise auth # hidden are forwarded as-is. pass return out # --------------------------------------------------------------------------- # Output parsing # --------------------------------------------------------------------------- def _parse_outputs(descriptor: dict[str, Any]) -> list[Any] | None: output_types = descriptor.get("output") or [] output_names = descriptor.get("output_name") or output_types output_is_list = descriptor.get("output_is_list") or [False] * len(output_types) output_tooltips = descriptor.get("output_tooltips") or [None] * len(output_types) out: list[Any] = [] for i, io_type in enumerate(output_types): name = output_names[i] if i < len(output_names) else io_type is_list = bool(output_is_list[i]) if i < len(output_is_list) else False tip = output_tooltips[i] if i < len(output_tooltips) else None cls = _OUTPUT_CLASSES.get(io_type) if cls is None: # Opaque custom-IO fallthrough (mirror of # ``_parse_input_spec``) — any unknown output io_type # string becomes ``IO.Custom(io_type).Output`` so partner # helper-config nodes can publish their own opaque # outputs (RECRAFT_*, OPENAI_INPUT_FILES, GEMINI_INPUT_FILES, # OPENAI_CHAT_CONFIG, …). The original io_type string # rides on the resulting Output via V3's ``@comfytype`` # decorator, so the frontend only chains output→input # sockets when the strings match. if isinstance(io_type, str) and io_type: cls = IO.Custom(io_type).Output else: 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, # 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, } # --------------------------------------------------------------------------- # Execute dispatch # --------------------------------------------------------------------------- def _extract_progress_pct(response: dict[str, Any]) -> int | None: """Map the RNP ``progress`` block to the 0-100 int that ``poll_op_raw`` feeds to its (single) ``ProgressBar(100)``. Returns ``None`` when the server doesn't yet have progress to report so the bar stays at its current position rather than snapping back to 0. """ progress = response.get("progress") if not isinstance(progress, dict): return None value = progress.get("value") total = progress.get("max") if not isinstance(value, (int, float)) or not isinstance(total, (int, float)) or total <= 0: return None pct = int(round((float(value) / float(total)) * 100)) return max(0, min(100, pct)) def _rnp_protocol_headers(extra: dict[str, str] | None = None) -> dict[str, str]: """Headers every RNP request carries — protocol-version + client version. Pass ``extra`` to merge per-request headers (e.g. the Idempotency-Key the server uses to authorize a TASK_LOST resume, or the freshly-rotated auth token a long-running poll wants to forward to the server).""" headers = { Header.PROTOCOL_VERSION: PROTOCOL_VERSION, Header.CLIENT_VERSION: rnp_client.CLIENT_VERSION, } if extra: headers.update(extra) return headers async def _run_async_polling( cls: type[IO.ComfyNode], *, node_id: str, server_url: str, schema_hash: str | None, payload_inputs: dict[str, Any], context: dict[str, Any], auth_headers: dict[str, str], idempotency_key: str, poll_interval_s: float, estimated_duration_s: int | float | None, max_poll_attempts: int, timeout_per_poll_s: float, cancel_timeout_s: float, max_retries_per_poll: int, retry_delay_per_poll_s: float, retry_backoff_per_poll: float, ) -> dict[str, Any]: """Drive the ``execute_async`` + ``poll`` flow via ``poll_op_raw``. Submits the task with ``execute_async``, then hands the polling loop off to ``comfy_api_nodes.util.client.poll_op_raw`` so RNP inherits the same per-poll retry/backoff, queued-vs-active accounting, max-poll-attempts cap, ``ProgressBar`` plumbing, ticker UI, and cancel-endpoint propagation as local partner nodes. All of DONE/ERROR/CANCELLED are mapped to poll_op_raw's "completed" set so the function returns the terminal poll-response dict and we dispatch on ``status`` here — that preserves ``RnpProtocolError`` typing for ERROR and ``ProcessingInterrupted`` for CANCELLED instead of flattening both to ``Exception("Task failed: …")``. On ``TASK_LOST`` (the server lost the task store between submit and poll, typically a restart), the function resubmits exactly once with the same idempotency key and resumes polling. A second ``TASK_LOST`` from the resume gives up — at that point the failure is structural rather than transient. """ resume_attempted = False while True: submit = await rnp_client.execute_remote_async( server_url, schema_hash, node_id, payload_inputs, context, extra_headers=auth_headers or None, idempotency_key=idempotency_key, node_cls=cls, ) task_id = submit["task_id"] interval = float(submit.get("poll_interval") or poll_interval_s) try: response = await _poll_until_terminal( cls, node_id=node_id, server_url=server_url, task_id=task_id, auth_headers=auth_headers, idempotency_key=idempotency_key, poll_interval_s=interval, estimated_duration_s=estimated_duration_s, max_poll_attempts=max_poll_attempts, timeout_per_poll_s=timeout_per_poll_s, cancel_timeout_s=cancel_timeout_s, max_retries_per_poll=max_retries_per_poll, retry_delay_per_poll_s=retry_delay_per_poll_s, retry_backoff_per_poll=retry_backoff_per_poll, ) except RnpProtocolError as e: if e.code == ErrorCode.TASK_LOST and not resume_attempted: resume_attempted = True _send_progress_text( cls, "Server restarted, resuming…", ) log.warning( "RNP %s: TASK_LOST during poll (task_id=%s); " "resubmitting with idempotency_key=%s", node_id, task_id, idempotency_key, ) continue raise return _interpret_terminal_response(task_id, response) def _send_progress_text(cls: type[IO.ComfyNode], text: str) -> None: """Best-effort progress badge update — invisible when the executor isn't running this class through PromptServer (e.g. unit tests).""" node_id = _node_id_for_ticker(cls) if node_id is None: return with contextlib.suppress(Exception): instance = PromptServer.instance if instance is not None: instance.send_progress_text(text, node_id) async def _poll_until_terminal( cls: type[IO.ComfyNode], *, node_id: str, server_url: str, task_id: str, auth_headers: dict[str, str], idempotency_key: str, poll_interval_s: float, estimated_duration_s: int | float | None, max_poll_attempts: int, timeout_per_poll_s: float, cancel_timeout_s: float, max_retries_per_poll: int, retry_delay_per_poll_s: float, retry_backoff_per_poll: float, ) -> dict[str, Any]: """Run ``poll_op_raw`` against an existing task_id and return the terminal poll-response dict. Wraps the per-request header build so the resume path in ``_run_async_polling`` can reuse it.""" # Forward the idempotency key on every poll/cancel: the server # uses it to gate TASK_LOST recovery (a typo'd task_id without an # idempotency key falls through to a generic NOT_FOUND so it # cannot trigger a phantom resubmit). Auth headers ride alongside # so a token rotated mid-task picks up on the next upstream call # the server makes. per_request_headers = _rnp_protocol_headers({ Header.IDEMPOTENCY_KEY: idempotency_key, **(auth_headers or {}), }) poll_endpoint = ApiEndpoint( f"rnp/v1/tasks/{task_id}", "GET", headers=per_request_headers, ) cancel_endpoint = ApiEndpoint( f"rnp/v1/tasks/{task_id}/cancel", "POST", headers=per_request_headers, ) return await poll_op_raw( cls, poll_endpoint, status_extractor=lambda r: r.get("status"), progress_extractor=_extract_progress_pct, completed_statuses=[TaskStatus.DONE, TaskStatus.ERROR, TaskStatus.CANCELLED], failed_statuses=[], queued_statuses=[TaskStatus.PENDING], poll_interval=poll_interval_s, max_poll_attempts=max_poll_attempts, timeout_per_poll=timeout_per_poll_s, max_retries_per_poll=max_retries_per_poll, retry_delay_per_poll=retry_delay_per_poll_s, retry_backoff_per_poll=retry_backoff_per_poll, estimated_duration=int(estimated_duration_s) if estimated_duration_s else None, cancel_endpoint=cancel_endpoint, cancel_timeout=cancel_timeout_s, base_url=server_url, auth_headers={}, error_parser=rnp_client._parse_rnp_error, is_rate_limited=rnp_client._is_rnp_backpressure, rate_limit_label=rnp_client._rnp_rate_limit_label, ) def _interpret_terminal_response( task_id: str, response: dict[str, Any], ) -> dict[str, Any]: """Map a terminal poll-response dict to the same ``{"outputs": [...]}`` shape sync ``execute`` returns, or raise the appropriate typed exception for ERROR / CANCELLED.""" status = response.get("status") if status == TaskStatus.DONE: outputs = response.get("outputs") if not isinstance(outputs, list): raise RnpProtocolError( f"task {task_id} done but no outputs list: {response!r}", ) return {"outputs": outputs} if status == TaskStatus.ERROR: err_body = response.get("exception") if not isinstance(err_body, dict): raise RnpProtocolError( f"task {task_id} errored without exception body", ) raise RnpProtocolError.from_response_body(err_body, 500) if status == TaskStatus.CANCELLED: raise ProcessingInterrupted(f"task {task_id} cancelled") raise RnpProtocolError( f"task {task_id} returned unrecognised terminal status: {status!r}", ) async def _execute_remote( cls: type[IO.ComfyNode], *, node_id: str, server_url: str, execute_path: str, schema_hash: str | None, timeout_s: float, outputs_meta: list[Any], hidden_names: list[str], inputs: dict[str, Any], max_inline_bytes: int | None = None, estimated_duration_s: int | float | None = None, execution_mode: str = ExecutionMode.REQUEST_RESPONSE, execution_policy: dict[str, Any] | None = None, url_fetch_policy: dict[str, Any] | None = None, input_serialization: dict[str, list[str]] | None = None, local_validate: dict[str, dict[str, Any]] | None = None, ) -> IO.NodeOutput: policy = execution_policy or _extract_execution_policy({}) url_fetch_policy = url_fetch_policy or {} # Strip hidden inputs from the user-supplied kwargs. Auth credentials # (``auth_token_comfy_org`` / ``api_key_comfy_org``) travel as standard # HTTP headers so the server can forward them to api.comfy.org without # parsing the body; everything else rides in the request ``context``. context: dict[str, Any] = {} auth_headers: dict[str, str] = {} hidden_holder = getattr(cls, "hidden", None) for hname in hidden_names: # Prefer cls.hidden (the standard executor path); fall back to a # value piped directly into kwargs for advanced workflows that # wire strings into hidden slots manually. Either way, strip the # name from the user-input kwargs so it doesn't leak into the # request body. hidden_value = ( getattr(hidden_holder, hname, None) if hidden_holder is not None else None ) piped_value = inputs.pop(hname, None) value = hidden_value if hidden_value is not None else piped_value if value is None: continue if hname == "auth_token_comfy_org": auth_headers["Authorization"] = f"Bearer {value}" elif hname == "api_key_comfy_org": auth_headers["X-API-KEY"] = str(value) else: context[hname] = value # Encode heavy-typed inputs (IMAGE/MASK/AUDIO) as RNP value # envelopes; scalars pass through. Oversized envelopes are # uploaded out-of-band and rewritten to ``{..., uri: }``. # ``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) # Stable per-execute() idempotency key — when the underlying HTTP # client retries the POST after a connection drop, the server sees # the same key and returns the original task_id rather than # spawning a duplicate run. idempotency_key = uuid.uuid4().hex async def _run_with_ticker( execute_schema_hash: str | None, ) -> dict[str, Any]: if execution_mode == ExecutionMode.ASYNC_POLLING: # poll_op_raw runs its own _display_time_progress ticker, # so we don't start a parallel one here — the sync path # below still uses the local ticker since execute_remote # blocks on a single HTTP request without UI feedback. return await _run_async_polling( cls, node_id=node_id, server_url=server_url, schema_hash=execute_schema_hash, payload_inputs=payload_inputs, context=context, auth_headers=auth_headers, idempotency_key=idempotency_key, poll_interval_s=policy["poll_interval_s"], estimated_duration_s=estimated_duration_s, max_poll_attempts=policy["max_poll_attempts"], timeout_per_poll_s=policy["timeout_per_poll_s"], cancel_timeout_s=policy["cancel_timeout_s"], max_retries_per_poll=policy["max_retries_per_poll"], retry_delay_per_poll_s=policy["retry_delay_per_poll_s"], retry_backoff_per_poll=policy["retry_backoff_per_poll"], ) # Shared mutable holder the local ticker watches so a # SERVER_BUSY/MAINTENANCE retry inside execute_remote can swap # the per-second "Status: Processing" line for the friendly # backpressure copy without restarting the ticker. The ticker # auto-expires the override once ``expires_at`` passes so a # stale "Server busy, retrying in 30s..." doesn't linger if the # post-wait request itself takes longer than the announced # Retry-After window. status_override: dict[str, Any] = {"label": None, "expires_at": None} def _local_rate_limit_label( status: int, body: Any, retry_after_s: float, ) -> str | None: label = rnp_client._rnp_rate_limit_label(status, body, retry_after_s) status_override["label"] = label # may be None for unknown codes status_override["expires_at"] = ( time.monotonic() + max(1.0, retry_after_s) if label else None ) return label ticker = asyncio.create_task( _progress_ticker( ticker_node_id, estimated_duration_s, status_override=status_override, ), ) try: return await rnp_client.execute_remote( server_url, execute_schema_hash, execute_path, payload_inputs, context, timeout_s=timeout_s, extra_headers=auth_headers or None, node_cls=cls, rate_limit_label=_local_rate_limit_label, ) finally: ticker.cancel() with contextlib.suppress(BaseException): await ticker try: body = await _run_with_ticker(schema_hash) except RnpProtocolError as e: # Schema-hash mismatch → re-fetch object_info, find the new # hash for this node, and retry once. This recovers from the # common deploy ordering of "server updated descriptor ⇒ # client still using cached one" without forcing a ComfyUI # restart. if e.code == ErrorCode.SCHEMA_HASH_MISMATCH: new_hash = await _refresh_schema_hash(server_url, node_id) if new_hash and new_hash != schema_hash: log.info( "RNP %s: schema_hash changed (%s → %s); retrying execute", node_id, schema_hash, new_hash, ) try: body = await _run_with_ticker(new_hash) except RnpProtocolError as e2: raise RuntimeError( _to_friendly_message(e2, node_id), ) from e2 else: raise RuntimeError( f"Remote node {node_id}: schema-hash mismatch and no " f"updated descriptor found upstream" ) from e else: log.warning("RNP execute %s failed: %s [%s]", node_id, e, e.code) raise RuntimeError(_to_friendly_message(e, node_id)) from e outputs = body.get("outputs") if not isinstance(outputs, list): raise RuntimeError( f"Remote node {node_id} returned no 'outputs' list" ) if len(outputs) != len(outputs_meta): raise RuntimeError( f"Remote node {node_id} returned {len(outputs)} outputs; " f"schema declared {len(outputs_meta)}" ) decoded = [] for value, meta in zip(outputs, outputs_meta): decoded.append(await _deserialize_output( value, meta, url_fetch_policy=url_fetch_policy, server_url=server_url, auth_headers=auth_headers, node_cls=cls, )) return IO.NodeOutput(*decoded) async def _refresh_schema_hash(server_url: str, node_id: str) -> str | None: """Re-fetch ``/object_info`` and return the freshly published ``remote.schema_hash`` for ``node_id`` (or ``None`` if the node has been removed upstream). """ try: body, _etag, _max_age = await rnp_client.fetch_object_info(server_url) except Exception as e: # noqa: BLE001 log.warning("schema-hash refresh failed for %s: %s", node_id, e) return None if not isinstance(body, dict): return None descriptor = body.get(node_id) if not isinstance(descriptor, dict): return None remote = descriptor.get("remote") or {} return remote.get("schema_hash") if isinstance(remote, dict) else None async def _deserialize_output( value: Any, meta: Any, *, url_fetch_policy: dict[str, Any], server_url: str, auth_headers: dict[str, str], node_cls: type[IO.ComfyNode] | None, ) -> Any: """Convert one output value (envelope or scalar) to its native Python form. Inline envelopes (with ``data``) decode immediately. URI envelopes look up their fetch policy from ``url_fetch_policy`` (keyed by the envelope's ``type``) and fetch the bytes via ``sync_op_raw`` so the GET inherits ``ProcessingInterrupted`` cancellation, retry/backoff on transient errors, and the per-policy timeout. Scalars pass through unchanged.""" if not is_envelope(value): # Scalars pass through unchanged — the schema's output type is # advisory here, the value already matches the native ComfyUI # type for STRING / INT / FLOAT / BOOLEAN. return value encoding = value.get("encoding") env_type = value.get("type", "") if encoding == "png_base64_batch_uri": # Externalised IMAGE batch on the response path (mirror of the # request-side ``serialization.maybe_externalize`` behaviour). # No server emits this shape today, but the decoder owns the # symmetry: any envelope shape the client may externalise on # the way out, the client must also accept on the way in. # Per-frame URIs fetched in parallel via ``asyncio.gather`` so # a 9-frame batch import is one round-trip rather than N — the # GET fan-out is the dominant latency for an out-of-band batch. # Frame ordering is preserved (gather returns positional # results) — critical because the IMAGE tensor's batch axis # carries semantic order. policy = url_fetch_policy.get(env_type) if not isinstance(policy, dict): raise RnpProtocolError( f"Server returned a URL batch envelope for output type " f"{env_type!r} but the descriptor declared no " f"remote.url_fetch policy for that type", code=ErrorCode.INTERNAL, ) uris = value.get("uris") if not isinstance(uris, list) or not uris: raise RnpProtocolError( "png_base64_batch_uri envelope requires a non-empty 'uris' list", code=ErrorCode.INTERNAL, ) import asyncio as _asyncio import base64 as _b64 per_frame_envs = [ {"type": env_type, "encoding": "png_base64", "uri": u} for u in uris ] raws = await _asyncio.gather(*[ _fetch_url_envelope( env, policy, auth_headers=auth_headers, node_cls=node_cls, ) for env in per_frame_envs ]) # Re-stamp as the inline batch shape so the existing # ``_decode_image_batch_envelope`` path handles it uniformly. value = {k: v for k, v in value.items() if k != "uris"} value["encoding"] = "png_base64_batch" value["frames"] = [_b64.b64encode(r).decode("ascii") for r in raws] elif "data" not in value and "uri" in value: policy = url_fetch_policy.get(env_type) if not isinstance(policy, dict): raise RnpProtocolError( f"Server returned a URL envelope for output type {env_type!r} " "but the descriptor declared no remote.url_fetch policy " "for that type", code=ErrorCode.INTERNAL, ) raw = await _fetch_url_envelope( value, policy, auth_headers=auth_headers, node_cls=node_cls, ) # Re-stamp the envelope with inline ``data`` so the existing # decoder dispatch in ``serialization`` sees a homogeneous # input shape regardless of how the bytes arrived. import base64 as _b64 value = {k: v for k, v in value.items() if k != "uri"} value["data"] = _b64.b64encode(raw).decode("ascii") return serialization.decode_envelope(value) async def _fetch_url_envelope( envelope: dict[str, Any], policy: dict[str, Any], *, auth_headers: dict[str, str], node_cls: type[IO.ComfyNode] | None, ) -> bytes: """GET ``envelope["uri"]`` via ``sync_op_raw`` honoring the descriptor's url_fetch policy (retries, backoff, timeout, optional auth). ``url_kind`` drives the auth posture: ``presigned`` / ``public`` → no auth header (the URL embeds its own credentials or needs none); ``comfy_api`` → forward the same Authorization / X-API-KEY the polls use. ``auth_required`` overrides ``url_kind`` when explicitly set to ``True``. """ from comfy_api_nodes.util.client import ApiEndpoint, sync_op_raw uri = envelope["uri"] url_kind = policy.get("url_kind") or "presigned" auth_required = bool(policy.get( "auth_required", url_kind == "comfy_api", )) fetch_auth: dict[str, str] = ( dict(auth_headers) if (auth_required and auth_headers) else {} ) timeout_s = float(policy.get("timeout_s") or 300) max_retries = int(policy.get("max_retries") or 3) retry_delay = float(policy.get("retry_delay_s") or 1.0) retry_backoff = float(policy.get("retry_backoff") or 2.0) raw = await sync_op_raw( node_cls, ApiEndpoint(uri, "GET"), timeout=timeout_s, max_retries=max_retries, retry_delay=retry_delay, retry_backoff=retry_backoff, wait_label="Downloading", as_binary=True, final_label_on_success=None, monitor_progress=False, auth_headers=fetch_auth, ) if not isinstance(raw, (bytes, bytearray)): raise RnpProtocolError( f"url_fetch GET {uri!r} returned non-bytes payload", code=ErrorCode.INTERNAL, ) return bytes(raw) async def _encode_inputs( inputs: dict[str, Any], *, 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. The duck-typing rules: * ``torch.Tensor`` rank 4 (B,H,W,C) → image envelope. * ``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(): _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 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": , "image2": }}`` or ``{"model": {"branch": "wan2.7-r2v", "video_refs": {"video1": }}}``); 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": , "image1": , ...}``). 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[]`` 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": # , ...}}`` for top-level AG; ``{"": branch_key, # "": {"video1": , ...}}`` 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. # 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( "Input %s: unrecognised tensor rank %d — sending as-is", name, rank, ) return value