Comfy-Org-comfy-remote-nodes/proxy_node.py

"""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: <N>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: <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")
        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": <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)``.

    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: <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)
    # 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": <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.
        # 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