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 == "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))
        # Only envelope-shaped values can be externalized; scalars pass
        # straight through ``maybe_externalize`` unchanged.
        if serialization.is_envelope(encoded):
            encoded = await serialization.maybe_externalize(
                encoded,
                server_url=server_url,
                max_inline_bytes=max_inline_bytes,
                auth_headers=auth_headers,
            )
        out[name] = encoded
    return out


def _enforce_local_validate(
    name: str,
    value: Any,
    rules: dict[str, Any],
    *,
    node_id: str | None,
) -> None:
    """Apply each ``local_validate.*`` rule to ``value`` and raise
    ``RnpProtocolError(INPUT_INVALID)`` on a violation.

    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:
    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 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