diff --git a/comfy_api_nodes/apis/comfy_cloud.py b/comfy_api_nodes/apis/comfy_cloud.py new file mode 100644 index 000000000..a781d3aef --- /dev/null +++ b/comfy_api_nodes/apis/comfy_cloud.py @@ -0,0 +1,120 @@ +from typing import Literal + +from pydantic import BaseModel, Field, field_validator + + +ComfyCloudWorkflow = Literal[ + "text-to-image", + "text-to-video", + "image-to-video", + "image-edit", + "image.ideogram-4-design.v1", + "image.krea-2-creative-image.v1", + "image.mage-flow-image.v1", + "image.flux-2-reference-edit.v1", + "image.qwen-image-edit-2511.v1", + "image.seedvr2-image-upscale.v1", + "video.minimax-h3-text-sound.v1", + "video.minimax-h3-image-sound.v1", + "video.ltx-2-3-image-audio-performance.v1", + "video.ltx-2-3-first-last-frame.v1", + "video.wan-2-2-14b-first-last-frame.v1", + "video.scail-2-character-replacement.v1", + "audio.ace-step-1-5-xl-turbo.v1", + "audio.stable-audio-3-medium.v1", + "audio.chatterbox-multilingual-voice-clone.v1", + "audio.chatterbox-dialogue.v1", + "audio.chatterbox-voice-conversion.v1", + "audio.melbandroformer-stem-separation.v1", + "3d.triposplat-image-to-gaussian-splat.v1", + "3d.hunyuan3d-2-1-image-to-3d.v1", + "3d.hunyuan3d-multiview-to-3d.v1", + "3d.moge-2-photo-to-textured-mesh.v1", + "3d.moge-2-panorama-to-3d-scene.v1", +] + + +class ComfyCloudWorkflowInputs(BaseModel): + prompt: str | None = Field(None) + image_url: str | None = Field(None) + assets: dict[str, "ComfyCloudAssetInput"] | None = Field(None) + audio_url: str | None = Field(None) + first_frame_url: str | None = Field(None) + last_frame_url: str | None = Field(None) + reference_character_url: str | None = Field(None) + driving_video_url: str | None = Field(None) + instruction: str | None = Field(None) + prompt_enhance: bool | None = Field(None) + enhance_prompt: bool | None = Field(None) + negative_prompt: str | None = Field(None) + aspect_ratio: str | None = Field(None) + duration_seconds: float | None = Field(None) + guidance: float | None = Field(None) + quality_mode: str | None = Field(None) + seed: int | None = Field(None, ge=0, le=0xFFFFFFFFFFFFFFFF) + scale: str | None = Field(None) + scene_prompt: str | None = Field(None) + driving_subject: str | None = Field(None) + reference_subject: str | None = Field(None) + style_prompt: str | None = Field(None) + lyrics: str | None = Field(None) + bpm: int | None = Field(None) + time_signature: str | None = Field(None) + language: str | None = Field(None) + key: str | None = Field(None) + expand_prompt: bool | None = Field(None) + category: str | None = Field(None) + text: str | None = Field(None) + exaggeration: float | None = Field(None) + cfg_weight: float | None = Field(None) + temperature: float | None = Field(None) + script: str | None = Field(None) + remove_background: bool | None = Field(None) + gaussian_count: int | None = Field(None) + fov_degrees: float | None = Field(None) + detail: int | None = Field(None) + mesh_decimation: int | None = Field(None) + gap_threshold: float | None = Field(None) + texture: bool | None = Field(None) + split_resolution: int | None = Field(None) + merge_resolution: int | None = Field(None) + + +class ComfyCloudAssetInput(BaseModel): + type: Literal["IMAGE", "VIDEO", "AUDIO"] = Field(...) + url: str = Field(...) + + +class ComfyCloudGenerateRequest(BaseModel): + workflow: ComfyCloudWorkflow = Field(...) + inputs: ComfyCloudWorkflowInputs = Field(...) + + +class ComfyCloudGenerateResponse(BaseModel): + task_id: str = Field(..., min_length=1) + status: str = Field(...) + polling_url: str = Field(...) + cancel_url: str = Field(...) + + @field_validator("task_id") + @classmethod + def task_id_must_not_be_blank(cls, value: str) -> str: + if not value.strip(): + raise ValueError("task_id must not be blank") + return value + + +class ComfyCloudStatusResponse(BaseModel): + task_id: str = Field(..., min_length=1) + status: str = Field(...) + progress: float | None = Field(None, ge=0, le=100) + output_url: str | None = Field(None) + output_urls: dict[str, str] | None = Field(None) + error: str | None = Field(None) + + @field_validator("task_id") + @classmethod + def task_id_must_not_be_blank(cls, value: str) -> str: + if not value.strip(): + raise ValueError("task_id must not be blank") + return value diff --git a/comfy_api_nodes/nodes_comfy_cloud.py b/comfy_api_nodes/nodes_comfy_cloud.py new file mode 100644 index 000000000..188c1f0cc --- /dev/null +++ b/comfy_api_nodes/nodes_comfy_cloud.py @@ -0,0 +1,1189 @@ +import math +import re +from typing import ClassVar +from urllib.parse import quote, urlsplit + +import torch + +from typing_extensions import override + +from comfy_api.latest import IO, ComfyExtension, Input +from comfy_api_nodes.apis.comfy_cloud import ( + ComfyCloudAssetInput, + ComfyCloudGenerateRequest, + ComfyCloudGenerateResponse, + ComfyCloudStatusResponse, + ComfyCloudWorkflow, + ComfyCloudWorkflowInputs, +) +from comfy_api_nodes.util import ( + ApiEndpoint, + download_url_to_audio_input, + download_url_to_file_3d, + download_url_to_image_tensor, + download_url_to_video_output, + get_number_of_images, + poll_op, + sync_op, + upload_audio_to_comfyapi, + upload_image_to_comfyapi, + upload_video_to_comfyapi, + validate_string, +) + + +_GENERATE_ENDPOINT = ApiEndpoint(path="/proxy/comfy-cloud/workflow/generate", method="POST") +_OUTPUT_DOWNLOAD_TIMEOUT = 30 * 60 +_MAX_UPLOAD_IMAGE_PIXELS = 32_000_000 +_MAX_UPLOAD_IMAGE_DIMENSION = 8192 +_MAX_DECODED_AUDIO_BYTES = 256 * 1024 * 1024 +COMFY_CLOUD_GPU_SECOND_USD = 0.001295 +COMFY_CLOUD_CREDITS_PER_USD = 211 +COMFY_CLOUD_GPU_SECOND_CREDITS = COMFY_CLOUD_GPU_SECOND_USD * COMFY_CLOUD_CREDITS_PER_USD +COMFY_CLOUD_GPU_HOUR_USD = COMFY_CLOUD_GPU_SECOND_USD * 3600 +COMFY_CLOUD_GPU_HOUR_CREDITS = COMFY_CLOUD_GPU_SECOND_CREDITS * 3600 +_COMFY_CLOUD_PRICE_BADGE = IO.PriceBadge( + expr=( + f'{{"type":"usd","usd":{COMFY_CLOUD_GPU_SECOND_USD:.6f},' + '"format":{"suffix":"/GPU-second","approximate":true}}' + ) +) +_COMFY_CLOUD_RATE_DESCRIPTION = ( + f" Estimated compute rate: ${COMFY_CLOUD_GPU_SECOND_USD:.6f}/GPU-second " + f"({COMFY_CLOUD_GPU_SECOND_CREDITS:.6f} credits/GPU-second using " + f"{COMFY_CLOUD_CREDITS_PER_USD} credits/USD; ${COMFY_CLOUD_GPU_HOUR_USD:.3f} or " + f"{COMFY_CLOUD_GPU_HOUR_CREDITS:.3f} credits/GPU-hour). " + "Actual final cost depends on GPU runtime." +) +_TEXT_LIMITS = { + "prompt": (1, 4096), + "instruction": (1, 4096), + "negative_prompt": (0, 2048), + "scene_prompt": (1, 4096), + "driving_subject": (1, 256), + "reference_subject": (1, 256), + "style_prompt": (1, 4096), + "lyrics": (0, 20000), + "text": (1, 5000), + "script": (1, 10000), +} + + +def _task_endpoints(task_id: str) -> tuple[ApiEndpoint, ApiEndpoint]: + if not task_id.strip(): + raise ValueError("Comfy Cloud returned an empty task ID.") + task_path = f"/proxy/comfy-cloud/workflow/tasks/{quote(task_id, safe='')}" + return ApiEndpoint(path=task_path), ApiEndpoint(path=f"{task_path}/cancel", method="POST") + + +def _with_input_sockets(inputs: list[IO.Input]) -> list[IO.Input]: + for input_spec in inputs: + if isinstance(input_spec, IO.WidgetInput): + input_spec.socketless = False + return inputs + + +def _validated_output_url(url: str) -> str: + parsed = urlsplit(url) + is_proxy_path = not parsed.scheme and not parsed.netloc and parsed.path.startswith("/proxy/comfy-cloud/") + is_signed_https_url = parsed.scheme == "https" and bool(parsed.netloc) and parsed.username is None + if not is_proxy_path and not is_signed_https_url: + raise RuntimeError("Comfy Cloud returned an invalid output URL.") + return url + + +def _validate_image_upload(image: Input.Image) -> None: + if not isinstance(image, torch.Tensor): + return + if image.ndim not in (3, 4): + raise ValueError("Invalid input image shape.") + height, width = image.shape[-3:-1] + if max(height, width) > _MAX_UPLOAD_IMAGE_DIMENSION or height * width > _MAX_UPLOAD_IMAGE_PIXELS: + raise ValueError("Input image exceeds the 8192px or 32-megapixel Comfy Cloud limit.") + + +def _validate_audio_upload(audio: Input.Audio) -> None: + waveform = audio["waveform"] + if waveform.ndim != 3 or waveform.shape[0] != 1 or waveform.shape[1] not in (1, 2): + raise ValueError("Audio must contain one mono or stereo waveform.") + if waveform.numel() * waveform.element_size() > _MAX_DECODED_AUDIO_BYTES: + raise ValueError("Decoded audio exceeds the 256 MiB Comfy Cloud limit.") + + +def _validate_node_inputs(cls: type[IO.ComfyNode], values: dict) -> dict: + validated = dict(values) + for input_spec in cls.define_schema().inputs: + if not isinstance(input_spec, IO.WidgetInput) or input_spec.id not in values: + continue + value = values[input_spec.id] + io_type = input_spec.get_io_type() + if io_type == "STRING": + value = value.strip() + minimum, maximum = _TEXT_LIMITS.get(input_spec.id, (0, None)) + validate_string( + value, + min_length=minimum, + max_length=maximum, + field_name=input_spec.id, + ) + validated[input_spec.id] = value + elif io_type == "COMBO" and value not in input_spec.options: + raise ValueError(f"Invalid {input_spec.id}: {value!r}.") + elif io_type == "BOOLEAN" and not isinstance(value, bool): + raise ValueError(f"{input_spec.id} must be a boolean.") + elif io_type == "INT": + if isinstance(value, bool) or not isinstance(value, int): + raise ValueError(f"{input_spec.id} must be an integer.") + if input_spec.min is not None and value < input_spec.min: + raise ValueError(f"{input_spec.id} must be at least {input_spec.min}.") + if input_spec.max is not None and value > input_spec.max: + raise ValueError(f"{input_spec.id} must be at most {input_spec.max}.") + elif io_type == "FLOAT": + if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value): + raise ValueError(f"{input_spec.id} must be a finite number.") + if input_spec.min is not None and value < input_spec.min: + raise ValueError(f"{input_spec.id} must be at least {input_spec.min}.") + if input_spec.max is not None and value > input_spec.max: + raise ValueError(f"{input_spec.id} must be at most {input_spec.max}.") + if input_spec.step: + origin = input_spec.min or 0 + steps = (value - origin) / input_spec.step + if not math.isclose(steps, round(steps), abs_tol=1e-7): + raise ValueError(f"{input_spec.id} must use increments of {input_spec.step}.") + return validated + + +class _ComfyCloudWorkflowNode(IO.ComfyNode): + workflow: ClassVar[ComfyCloudWorkflow] + node_id: ClassVar[str] + display_name: ClassVar[str] + category: ClassVar[str] + requires_image: ClassVar[bool] + returns_video: ClassVar[bool] + + @classmethod + def define_schema(cls) -> IO.Schema: + inputs = [ + IO.String.Input( + "prompt", + multiline=True, + default="", + tooltip="Describe the content to generate or the edit to apply.", + ) + ] + if cls.requires_image: + inputs.append(IO.Image.Input("image")) + + output = IO.Video.Output() if cls.returns_video else IO.Image.Output() + return IO.Schema( + node_id=cls.node_id, + display_name=cls.display_name, + category=cls.category, + description="Runs this workflow on Comfy Cloud and returns the generated media." + + _COMFY_CLOUD_RATE_DESCRIPTION, + inputs=_with_input_sockets(inputs), + outputs=[output], + hidden=[ + IO.Hidden.auth_token_comfy_org, + IO.Hidden.api_key_comfy_org, + IO.Hidden.unique_id, + ], + is_api_node=True, + price_badge=_COMFY_CLOUD_PRICE_BADGE, + ) + + @classmethod + async def execute(cls, prompt: str, image: Input.Image | None = None) -> IO.NodeOutput: + prompt = _validate_node_inputs(cls, locals())["prompt"] + + image_url = None + if cls.requires_image: + image_url = await cls._upload_image(image) + + return await cls._run(ComfyCloudWorkflowInputs(prompt=prompt, image_url=image_url)) + + @classmethod + async def _upload_image(cls, image: Input.Image, total_pixels: int | None = 2048 * 2048) -> str: + if get_number_of_images(image) != 1: + raise ValueError("Exactly one input image is required.") + _validate_image_upload(image) + return await upload_image_to_comfyapi(cls, image, total_pixels=total_pixels) + + @classmethod + async def _run(cls, inputs: ComfyCloudWorkflowInputs) -> IO.NodeOutput: + task = await sync_op( + cls, + _GENERATE_ENDPOINT, + response_model=ComfyCloudGenerateResponse, + data=ComfyCloudGenerateRequest( + workflow=cls.workflow, + inputs=inputs, + ), + ) + polling_endpoint, cancel_endpoint = _task_endpoints(task.task_id) + result = await poll_op( + cls, + polling_endpoint, + response_model=ComfyCloudStatusResponse, + status_extractor=lambda response: response.status, + progress_extractor=lambda response: response.progress, + cancel_endpoint=cancel_endpoint, + ) + if not result.output_url: + raise RuntimeError("Comfy Cloud task completed without an output URL.") + + if cls.returns_video: + output = await download_url_to_video_output( + _validated_output_url(result.output_url), + timeout=_OUTPUT_DOWNLOAD_TIMEOUT, + cls=cls, + allow_redirects=False, + ) + else: + output = await download_url_to_image_tensor( + _validated_output_url(result.output_url), + timeout=_OUTPUT_DOWNLOAD_TIMEOUT, + cls=cls, + allow_redirects=False, + ) + return IO.NodeOutput(output) + + +class ComfyCloudTextToImageNode(_ComfyCloudWorkflowNode): + workflow = "text-to-image" + node_id = "ComfyCloudTextToImageNode" + display_name = "Comfy Cloud Text to Image" + category = "partner/image/Comfy Cloud" + requires_image = False + returns_video = False + + +class ComfyCloudTextToVideoNode(_ComfyCloudWorkflowNode): + workflow = "text-to-video" + node_id = "ComfyCloudTextToVideoNode" + display_name = "Comfy Cloud Text to Video" + category = "partner/video/Comfy Cloud" + requires_image = False + returns_video = True + + +class ComfyCloudImageToVideoNode(_ComfyCloudWorkflowNode): + workflow = "image-to-video" + node_id = "ComfyCloudImageToVideoNode" + display_name = "Comfy Cloud Image to Video" + category = "partner/video/Comfy Cloud" + requires_image = True + returns_video = True + + +class ComfyCloudImageEditNode(_ComfyCloudWorkflowNode): + workflow = "image-edit" + node_id = "ComfyCloudImageEditNode" + display_name = "Comfy Cloud Image Edit" + category = "partner/image/Comfy Cloud" + requires_image = True + returns_video = False + + +_ASPECT_RATIOS = ["1:1", "3:4", "2:3", "3:2", "4:3", "16:9", "9:16", "21:9"] +_UINT64_MAX = 0xFFFFFFFFFFFFFFFF + + +def _prompt_input(name: str = "prompt") -> IO.String.Input: + return IO.String.Input(name, multiline=True, default="") + + +def _aspect_ratio_input() -> IO.Combo.Input: + return IO.Combo.Input("aspect_ratio", options=_ASPECT_RATIOS, default="1:1") + + +def _seed_input() -> IO.Int.Input: + return IO.Int.Input("seed", default=0, min=0, max=_UINT64_MAX, control_after_generate=True) + + +def _image_schema(node_id: str, display_name: str, inputs: list[IO.Input]) -> IO.Schema: + return IO.Schema( + node_id=node_id, + display_name=display_name, + category="partner/image/Comfy Cloud", + description="Runs this image workflow on Comfy Cloud and returns the generated image." + + _COMFY_CLOUD_RATE_DESCRIPTION, + inputs=_with_input_sockets(inputs), + outputs=[IO.Image.Output()], + hidden=[ + IO.Hidden.auth_token_comfy_org, + IO.Hidden.api_key_comfy_org, + IO.Hidden.unique_id, + ], + is_api_node=True, + price_badge=_COMFY_CLOUD_PRICE_BADGE, + ) + + +class ComfyCloudIdeogram4DesignNode(_ComfyCloudWorkflowNode): + workflow = "image.ideogram-4-design.v1" + node_id = "ComfyCloudIdeogram4DesignNode" + display_name = "Ideogram 4 Design" + category = "partner/image/Comfy Cloud" + requires_image = False + returns_video = False + + @classmethod + def define_schema(cls) -> IO.Schema: + return _image_schema( + cls.node_id, + cls.display_name, + [ + _prompt_input(), + _aspect_ratio_input(), + IO.Combo.Input( + "quality_mode", options=["quality", "balanced", "fast"], default="balanced" + ), + _seed_input(), + ], + ) + + @classmethod + # pylint: disable=arguments-renamed + async def execute( + cls, prompt: str, aspect_ratio: str = "1:1", quality_mode: str = "balanced", seed: int = 0 + ) -> IO.NodeOutput: + values = _validate_node_inputs(cls, locals()) + prompt = values["prompt"] + return await cls._run( + ComfyCloudWorkflowInputs( + prompt=prompt, aspect_ratio=aspect_ratio, quality_mode=quality_mode, seed=seed + ) + ) + + +class ComfyCloudKrea2CreativeImageNode(_ComfyCloudWorkflowNode): + workflow = "image.krea-2-creative-image.v1" + node_id = "ComfyCloudKrea2CreativeImageNode" + display_name = "Krea 2 Creative Image" + category = "partner/image/Comfy Cloud" + requires_image = False + returns_video = False + + @classmethod + def define_schema(cls) -> IO.Schema: + return _image_schema( + cls.node_id, + cls.display_name, + [ + _prompt_input(), + IO.Boolean.Input("prompt_enhance", default=True), + _aspect_ratio_input(), + _seed_input(), + ], + ) + + @classmethod + # pylint: disable=arguments-renamed + async def execute( + cls, prompt: str, prompt_enhance: bool = True, aspect_ratio: str = "1:1", seed: int = 0 + ) -> IO.NodeOutput: + values = _validate_node_inputs(cls, locals()) + prompt = values["prompt"] + return await cls._run( + ComfyCloudWorkflowInputs( + prompt=prompt, prompt_enhance=prompt_enhance, aspect_ratio=aspect_ratio, seed=seed + ) + ) + + +class ComfyCloudMageFlowImageNode(_ComfyCloudWorkflowNode): + workflow = "image.mage-flow-image.v1" + node_id = "ComfyCloudMageFlowImageNode" + display_name = "Mage-Flow Image" + category = "partner/image/Comfy Cloud" + requires_image = False + returns_video = False + + @classmethod + def define_schema(cls) -> IO.Schema: + return _image_schema( + cls.node_id, + cls.display_name, + [ + _prompt_input(), + IO.String.Input("negative_prompt", multiline=True, default=""), + _aspect_ratio_input(), + _seed_input(), + ], + ) + + @classmethod + # pylint: disable=arguments-renamed + async def execute( + cls, prompt: str, negative_prompt: str = "", aspect_ratio: str = "1:1", seed: int = 0 + ) -> IO.NodeOutput: + values = _validate_node_inputs(cls, locals()) + prompt = values["prompt"] + negative_prompt = values["negative_prompt"] + return await cls._run( + ComfyCloudWorkflowInputs( + prompt=prompt, negative_prompt=negative_prompt, aspect_ratio=aspect_ratio, seed=seed + ) + ) + + +class ComfyCloudFlux2ReferenceEditNode(_ComfyCloudWorkflowNode): + workflow = "image.flux-2-reference-edit.v1" + node_id = "ComfyCloudFlux2ReferenceEditNode" + display_name = "FLUX.2 Reference Edit" + category = "partner/image/Comfy Cloud" + requires_image = True + returns_video = False + + @classmethod + def define_schema(cls) -> IO.Schema: + return _image_schema( + cls.node_id, + cls.display_name, + [ + IO.Image.Input("image"), + _prompt_input("instruction"), + IO.Float.Input("guidance", default=4.0, min=1.0, max=10.0, step=0.1), + IO.Combo.Input("quality_mode", options=["quality", "fast"], default="quality"), + _seed_input(), + ], + ) + + @classmethod + # pylint: disable=arguments-renamed + async def execute( + cls, + image: Input.Image, + instruction: str, + guidance: float = 4.0, + quality_mode: str = "quality", + seed: int = 0, + ) -> IO.NodeOutput: + values = _validate_node_inputs(cls, locals()) + instruction = values["instruction"] + return await cls._run( + ComfyCloudWorkflowInputs( + assets={ + "image": ComfyCloudAssetInput( + type="IMAGE", url=await cls._upload_image(image, total_pixels=None) + ) + }, + instruction=instruction, + guidance=guidance, + quality_mode=quality_mode, + seed=seed, + ) + ) + + +class ComfyCloudQwenImageEdit2511Node(_ComfyCloudWorkflowNode): + workflow = "image.qwen-image-edit-2511.v1" + node_id = "ComfyCloudQwenImageEdit2511Node" + display_name = "Qwen Image Edit 2511" + category = "partner/image/Comfy Cloud" + requires_image = True + returns_video = False + + @classmethod + def define_schema(cls) -> IO.Schema: + return _image_schema( + cls.node_id, + cls.display_name, + [ + IO.Image.Input("image"), + _prompt_input("instruction"), + IO.Combo.Input("quality_mode", options=["quality", "fast"], default="quality"), + _seed_input(), + ], + ) + + @classmethod + # pylint: disable=arguments-renamed + async def execute( + cls, + image: Input.Image, + instruction: str, + quality_mode: str = "quality", + seed: int = 0, + ) -> IO.NodeOutput: + values = _validate_node_inputs(cls, locals()) + instruction = values["instruction"] + return await cls._run( + ComfyCloudWorkflowInputs( + assets={ + "image": ComfyCloudAssetInput( + type="IMAGE", url=await cls._upload_image(image, total_pixels=None) + ) + }, + instruction=instruction, + quality_mode=quality_mode, + seed=seed, + ) + ) + + +class ComfyCloudSeedVR2ImageUpscaleNode(_ComfyCloudWorkflowNode): + workflow = "image.seedvr2-image-upscale.v1" + node_id = "ComfyCloudSeedVR2ImageUpscaleNode" + display_name = "SeedVR2 Image Upscale" + category = "partner/image/Comfy Cloud" + requires_image = True + returns_video = False + + @classmethod + def define_schema(cls) -> IO.Schema: + return _image_schema( + cls.node_id, + cls.display_name, + [IO.Image.Input("image"), IO.Combo.Input("scale", options=["2x", "4x"], default="4x")], + ) + + @classmethod + # pylint: disable=arguments-renamed + async def execute(cls, image: Input.Image, scale: str = "4x") -> IO.NodeOutput: + _validate_node_inputs(cls, locals()) + return await cls._run( + ComfyCloudWorkflowInputs( + assets={ + "image": ComfyCloudAssetInput( + type="IMAGE", url=await cls._upload_image(image, total_pixels=None) + ) + }, + scale=scale, + ) + ) + + +async def _run_video_workflow(cls: type[IO.ComfyNode], workflow: ComfyCloudWorkflow, inputs: ComfyCloudWorkflowInputs) -> IO.NodeOutput: + task = await sync_op(cls, _GENERATE_ENDPOINT, response_model=ComfyCloudGenerateResponse, data=ComfyCloudGenerateRequest(workflow=workflow, inputs=inputs)) + polling_endpoint, cancel_endpoint = _task_endpoints(task.task_id) + result = await poll_op( + cls, + polling_endpoint, + response_model=ComfyCloudStatusResponse, + status_extractor=lambda response: response.status, + progress_extractor=lambda response: response.progress, + cancel_endpoint=cancel_endpoint, + ) + if not result.output_url: + raise RuntimeError("Comfy Cloud task completed without an output URL.") + return IO.NodeOutput( + await download_url_to_video_output( + _validated_output_url(result.output_url), + timeout=_OUTPUT_DOWNLOAD_TIMEOUT, + cls=cls, + allow_redirects=False, + ) + ) + + +def _video_schema(node_id: str, display_name: str, inputs: list[IO.Input]) -> IO.Schema: + return IO.Schema( + node_id=node_id, + display_name=display_name, + category="partner/video/Comfy Cloud", + description="Runs this video workflow on Comfy Cloud and returns the generated video." + + _COMFY_CLOUD_RATE_DESCRIPTION, + inputs=_with_input_sockets(inputs), + outputs=[IO.Video.Output()], + hidden=[IO.Hidden.auth_token_comfy_org, IO.Hidden.api_key_comfy_org, IO.Hidden.unique_id], + is_api_node=True, + price_badge=_COMFY_CLOUD_PRICE_BADGE, + ) + + +def _video_seed_input(default: int) -> IO.Int.Input: + return IO.Int.Input("seed", default=default, min=0, max=_UINT64_MAX, control_after_generate=True) + + +class ComfyCloudMiniMaxH3TextSoundNode(IO.ComfyNode): + @classmethod + def define_schema(cls) -> IO.Schema: + return _video_schema( + "ComfyCloudMiniMaxH3TextSoundNode", + "MiniMax H3 Text + Sound", + [ + _prompt_input(), + IO.Combo.Input("aspect_ratio", options=["1:1", "2:3", "3:2", "3:4", "4:3", "9:16", "16:9", "21:9"], default="1:1"), + IO.Float.Input("duration_seconds", default=5, min=5, max=15, step=0.01), + _video_seed_input(168866841893410), + ], + ) + + @classmethod + async def execute(cls, prompt: str, aspect_ratio: str, duration_seconds: float, seed: int) -> IO.NodeOutput: + values = _validate_node_inputs(cls, locals()) + prompt = values["prompt"] + return await _run_video_workflow(cls, "video.minimax-h3-text-sound.v1", ComfyCloudWorkflowInputs(prompt=prompt, aspect_ratio=aspect_ratio, duration_seconds=duration_seconds, seed=seed)) + + +class ComfyCloudMiniMaxH3ImageSoundNode(IO.ComfyNode): + @classmethod + def define_schema(cls) -> IO.Schema: + return _video_schema( + "ComfyCloudMiniMaxH3ImageSoundNode", + "MiniMax H3 Image + Sound", + [ + IO.Image.Input("image"), + _prompt_input(), + IO.Combo.Input("aspect_ratio", options=["1:1", "2:3", "3:2", "3:4", "4:3", "9:16", "16:9", "21:9"], default="1:1"), + IO.Float.Input("duration_seconds", default=5, min=5, max=15, step=0.01), + _video_seed_input(168866841893410), + ], + ) + + @classmethod + async def execute(cls, image: Input.Image, prompt: str, aspect_ratio: str, duration_seconds: float, seed: int) -> IO.NodeOutput: + values = _validate_node_inputs(cls, locals()) + prompt = values["prompt"] + if get_number_of_images(image) != 1: + raise ValueError("Exactly one input image is required.") + _validate_image_upload(image) + image_url = await upload_image_to_comfyapi(cls, image) + return await _run_video_workflow(cls, "video.minimax-h3-image-sound.v1", ComfyCloudWorkflowInputs(prompt=prompt, image_url=image_url, aspect_ratio=aspect_ratio, duration_seconds=duration_seconds, seed=seed)) + + +class ComfyCloudLTX23ImageAudioPerformanceNode(IO.ComfyNode): + @classmethod + def define_schema(cls) -> IO.Schema: + return _video_schema( + "ComfyCloudLTX23ImageAudioPerformanceNode", + "LTX-2.3 Image + Audio Performance", + [ + IO.Image.Input("image"), IO.Audio.Input("audio"), _prompt_input(), + IO.Boolean.Input("enhance_prompt", default=True), + IO.Float.Input("duration_seconds", default=9, min=1, max=15, step=0.01, tooltip="Must not exceed the input audio duration."), + _video_seed_input(225158785956033), + ], + ) + + @classmethod + async def execute(cls, image: Input.Image, audio: Input.Audio, prompt: str, enhance_prompt: bool, duration_seconds: float, seed: int) -> IO.NodeOutput: + values = _validate_node_inputs(cls, locals()) + prompt = values["prompt"] + if get_number_of_images(image) != 1: + raise ValueError("Exactly one input image is required.") + _validate_image_upload(image) + _validate_audio_upload(audio) + audio_duration = _audio_duration(audio) + if duration_seconds - min(1 / float(audio["sample_rate"]), 1e-3) > audio_duration: + raise ValueError(f"Duration ({duration_seconds:g}s) exceeds input audio duration ({audio_duration:.2f}s).") + image_url = await upload_image_to_comfyapi(cls, image) + audio_url = await upload_audio_to_comfyapi(cls, audio) + return await _run_video_workflow(cls, "video.ltx-2-3-image-audio-performance.v1", ComfyCloudWorkflowInputs(prompt=prompt, image_url=image_url, audio_url=audio_url, enhance_prompt=enhance_prompt, duration_seconds=duration_seconds, seed=seed)) + + +class ComfyCloudLTX23FirstLastFrameNode(IO.ComfyNode): + @classmethod + def define_schema(cls) -> IO.Schema: + return _video_schema( + "ComfyCloudLTX23FirstLastFrameNode", + "LTX-2.3 First & Last Frame", + [IO.Image.Input("first_frame"), IO.Image.Input("last_frame"), _prompt_input(), IO.Int.Input("duration_seconds", default=5, min=2, max=10, step=1, tooltip="25 fps; output frame count is duration × 25 + 1."), _video_seed_input(315253765879496)], + ) + + @classmethod + async def execute(cls, first_frame: Input.Image, last_frame: Input.Image, prompt: str, duration_seconds: int, seed: int) -> IO.NodeOutput: + values = _validate_node_inputs(cls, locals()) + prompt = values["prompt"] + if get_number_of_images(first_frame) != 1 or get_number_of_images(last_frame) != 1: + raise ValueError("Exactly one first frame and one last frame are required.") + _validate_image_upload(first_frame) + _validate_image_upload(last_frame) + first_url = await upload_image_to_comfyapi(cls, first_frame, wait_label="Uploading first frame") + last_url = await upload_image_to_comfyapi(cls, last_frame, wait_label="Uploading last frame") + return await _run_video_workflow(cls, "video.ltx-2-3-first-last-frame.v1", ComfyCloudWorkflowInputs(prompt=prompt, first_frame_url=first_url, last_frame_url=last_url, duration_seconds=duration_seconds, seed=seed)) + + +class ComfyCloudWan22FirstLastFrameNode(IO.ComfyNode): + @classmethod + def define_schema(cls) -> IO.Schema: + return _video_schema( + "ComfyCloudWan22FirstLastFrameNode", + "Wan 2.2 14B First & Last Frame", + [IO.Image.Input("first_frame"), IO.Image.Input("last_frame"), _prompt_input(), IO.String.Input("negative_prompt", multiline=True, default="graph tested Chinese quality negative"), IO.Int.Input("duration_seconds", default=5, min=2, max=8, step=1, tooltip="Graph frame count is floor(duration × 16 + 1)."), _video_seed_input(984937593540091)], + ) + + @classmethod + async def execute(cls, first_frame: Input.Image, last_frame: Input.Image, prompt: str, negative_prompt: str, duration_seconds: int, seed: int) -> IO.NodeOutput: + values = _validate_node_inputs(cls, locals()) + prompt = values["prompt"] + negative_prompt = values["negative_prompt"] + if get_number_of_images(first_frame) != 1 or get_number_of_images(last_frame) != 1: + raise ValueError("Exactly one first frame and one last frame are required.") + _validate_image_upload(first_frame) + _validate_image_upload(last_frame) + first_url = await upload_image_to_comfyapi(cls, first_frame, wait_label="Uploading first frame") + last_url = await upload_image_to_comfyapi(cls, last_frame, wait_label="Uploading last frame") + return await _run_video_workflow(cls, "video.wan-2-2-14b-first-last-frame.v1", ComfyCloudWorkflowInputs(prompt=prompt, negative_prompt=negative_prompt, first_frame_url=first_url, last_frame_url=last_url, duration_seconds=duration_seconds, seed=seed)) + + +class ComfyCloudSCAIL2CharacterReplacementNode(IO.ComfyNode): + @classmethod + def define_schema(cls) -> IO.Schema: + return _video_schema( + "ComfyCloudSCAIL2CharacterReplacementNode", + "SCAIL-2 Character Replacement", + [IO.Image.Input("reference_character"), IO.Video.Input("driving_video", tooltip="Must contain 81–157 decoded frames."), _prompt_input("scene_prompt"), IO.String.Input("driving_subject", default="human"), IO.String.Input("reference_subject", default="human"), _video_seed_input(1)], + ) + + @classmethod + async def execute(cls, reference_character: Input.Image, driving_video: Input.Video, scene_prompt: str, driving_subject: str, reference_subject: str, seed: int) -> IO.NodeOutput: + values = _validate_node_inputs(cls, locals()) + scene_prompt = values["scene_prompt"] + driving_subject = values["driving_subject"] + reference_subject = values["reference_subject"] + if get_number_of_images(reference_character) != 1: + raise ValueError("Exactly one reference character image is required.") + _validate_image_upload(reference_character) + try: + frame_count = driving_video.get_frame_count() + except Exception as error: + raise ValueError("Unable to determine video frame count.") from error + if isinstance(frame_count, bool) or not isinstance(frame_count, int): + raise ValueError("Unable to determine video frame count.") + if not 81 <= frame_count <= 157: + raise ValueError(f"Video frame count must be between 81 and 157, got {frame_count}.") + image_url = await upload_image_to_comfyapi(cls, reference_character) + video_url = await upload_video_to_comfyapi(cls, driving_video) + return await _run_video_workflow(cls, "video.scail-2-character-replacement.v1", ComfyCloudWorkflowInputs(scene_prompt=scene_prompt, driving_subject=driving_subject, reference_subject=reference_subject, reference_character_url=image_url, driving_video_url=video_url, seed=seed)) + + +_UINT32_MAX = 0xFFFFFFFF +_ACE_LANGUAGES = ["ar", "az", "bg", "bn", "ca", "cs", "da", "de", "el", "en", "es", "fa", "fi", "fr", "he", "hi", "hr", "ht", "hu", "id", "is", "it", "ja", "ko", "la", "lt", "ms", "ne", "nl", "no", "pa", "pl", "pt", "ro", "ru", "sa", "sk", "sr", "sv", "sw", "ta", "te", "th", "tl", "tr", "uk", "ur", "vi", "yue", "zh", "unknown"] +_ACE_KEYS = [f"{root} {mode}" for mode in ("major", "minor") for root in ("C", "C#", "Db", "D", "D#", "Eb", "E", "F", "F#", "Gb", "G", "G#", "Ab", "A", "A#", "Bb", "B")] +_CHATTERBOX_LANGUAGES = ["Arabic (ar)", "Danish (da)", "German (de)", "Greek (el)", "English (en)", "Spanish (es)", "Finnish (fi)", "French (fr)", "Hebrew (he)", "Hindi (hi)", "Italian (it)", "Japanese (ja)", "Korean (ko)", "Malay (ms)", "Dutch (nl)", "Norwegian (no)", "Polish (pl)", "Portuguese (pt)", "Russian (ru)", "Swedish (sv)", "Swahili (sw)", "Turkish (tr)", "Chinese (zh)"] + + +def _audio_schema(node_id: str, display_name: str, inputs: list[IO.Input], outputs: list[IO.Output] | None = None) -> IO.Schema: + return IO.Schema( + node_id=node_id, + display_name=display_name, + category="partner/audio/Comfy Cloud", + description="Runs this audio workflow on Comfy Cloud and returns the generated audio." + + _COMFY_CLOUD_RATE_DESCRIPTION, + inputs=_with_input_sockets(inputs), + outputs=outputs or [IO.Audio.Output()], + hidden=[IO.Hidden.auth_token_comfy_org, IO.Hidden.api_key_comfy_org, IO.Hidden.unique_id], + is_api_node=True, + price_badge=_COMFY_CLOUD_PRICE_BADGE, + ) + + +def _audio_duration(audio: Input.Audio) -> float: + sample_rate = float(audio["sample_rate"]) + if not math.isfinite(sample_rate) or sample_rate <= 0: + raise ValueError("Audio sample rate must be a positive number.") + return audio["waveform"].shape[-1] / sample_rate + + +def _validate_audio_duration(name: str, audio: Input.Audio, minimum: float, maximum: float) -> None: + duration = _audio_duration(audio) + tolerance = min(1 / float(audio["sample_rate"]), 1e-3) + if duration + tolerance < minimum or duration - tolerance > maximum: + raise ValueError(f"{name} duration must be between {minimum:g} and {maximum:g} seconds.") + + +def _normalize_dialogue(script: str) -> str: + utterances: list[tuple[str, list[str]]] = [] + for raw_line in script.splitlines(): + line = raw_line.strip() + if not line: + continue + match = re.fullmatch(r"(?:SPEAKER\s+)?([A-Z])\s*:\s*(.*)", line, flags=re.IGNORECASE) + if match: + speaker, text = match.groups() + if speaker.upper() not in ("A", "B"): + raise ValueError("Dialogue supports only speakers A and B.") + if utterances and not utterances[-1][1]: + raise ValueError("Dialogue utterances cannot be blank.") + utterances.append((speaker.upper(), [text.strip()] if text.strip() else [])) + elif re.match(r"(?:SPEAKER\s+[A-Z]|NARRATOR)\s*:", line, flags=re.IGNORECASE): + raise ValueError("Dialogue supports only speakers A and B.") + elif utterances: + utterances[-1][1].append(line) + else: + raise ValueError("Dialogue must start with speaker A or B.") + if not utterances: + raise ValueError("Dialogue must contain at least one utterance.") + if not utterances[-1][1]: + raise ValueError("Dialogue utterances cannot be blank.") + return "\n".join(f"SPEAKER {speaker}: {' '.join(lines)}" for speaker, lines in utterances) + + +async def _audio_asset(cls: type[IO.ComfyNode], name: str, audio: Input.Audio) -> dict[str, ComfyCloudAssetInput]: + _validate_audio_upload(audio) + return {name: ComfyCloudAssetInput(type="AUDIO", url=await upload_audio_to_comfyapi(cls, audio))} + + +async def _run_audio_workflow(cls: type[IO.ComfyNode], workflow: ComfyCloudWorkflow, inputs: ComfyCloudWorkflowInputs, output_names: tuple[str, ...] = ()) -> IO.NodeOutput: + task = await sync_op(cls, _GENERATE_ENDPOINT, response_model=ComfyCloudGenerateResponse, data=ComfyCloudGenerateRequest(workflow=workflow, inputs=inputs)) + polling_endpoint, cancel_endpoint = _task_endpoints(task.task_id) + result = await poll_op( + cls, + polling_endpoint, + response_model=ComfyCloudStatusResponse, + status_extractor=lambda response: response.status, + progress_extractor=lambda response: response.progress, + cancel_endpoint=cancel_endpoint, + ) + if output_names: + if not result.output_urls or any(not result.output_urls.get(name) for name in output_names): + raise RuntimeError("Comfy Cloud task completed without all named output URLs.") + outputs = [ + await download_url_to_audio_input( + _validated_output_url(result.output_urls[name]), + timeout=_OUTPUT_DOWNLOAD_TIMEOUT, + cls=cls, + allow_redirects=False, + ) + for name in output_names + ] + return IO.NodeOutput(*outputs) + if not result.output_url: + raise RuntimeError("Comfy Cloud task completed without an output URL.") + return IO.NodeOutput( + await download_url_to_audio_input( + _validated_output_url(result.output_url), + timeout=_OUTPUT_DOWNLOAD_TIMEOUT, + cls=cls, + allow_redirects=False, + ) + ) + + +class ComfyCloudACEStep15XLTurboNode(IO.ComfyNode): + @classmethod + def define_schema(cls) -> IO.Schema: + return _audio_schema( + "ComfyCloudACEStep15XLTurboNode", + "ACE-Step 1.5 XL Turbo", + [ + _prompt_input("style_prompt"), + IO.String.Input("lyrics", multiline=True, default=""), + IO.Float.Input("duration_seconds", default=120, min=10, max=300, step=0.1), + _seed_input(), + IO.Int.Input("bpm", default=120, min=10, max=300), + IO.Combo.Input("time_signature", options=["2", "3", "4", "6"], default="4"), + IO.Combo.Input("language", options=_ACE_LANGUAGES, default="en"), + IO.Combo.Input("key", options=_ACE_KEYS, default="E minor"), + ], + ) + + @classmethod + async def execute(cls, style_prompt: str, lyrics: str, duration_seconds: float, seed: int, bpm: int, time_signature: str, language: str, key: str) -> IO.NodeOutput: + values = _validate_node_inputs(cls, locals()) + style_prompt = values["style_prompt"] + lyrics = values["lyrics"] + return await _run_audio_workflow(cls, "audio.ace-step-1-5-xl-turbo.v1", ComfyCloudWorkflowInputs(style_prompt=style_prompt, lyrics=lyrics, duration_seconds=duration_seconds, seed=seed, bpm=bpm, time_signature=time_signature, language=language, key=key)) + + +class ComfyCloudStableAudio3MediumNode(IO.ComfyNode): + @classmethod + def define_schema(cls) -> IO.Schema: + return _audio_schema( + "ComfyCloudStableAudio3MediumNode", + "Stable Audio 3 Medium", + [ + _prompt_input(), + IO.Float.Input("duration_seconds", default=30, min=1, max=300, step=0.1), + _seed_input(), + IO.Boolean.Input("expand_prompt", default=True), + IO.Combo.Input("category", options=["Music", "Instrument", "SFX", "One-shot"], default="Music"), + ], + ) + + @classmethod + async def execute(cls, prompt: str, duration_seconds: float, seed: int, expand_prompt: bool, category: str) -> IO.NodeOutput: + values = _validate_node_inputs(cls, locals()) + prompt = values["prompt"] + return await _run_audio_workflow(cls, "audio.stable-audio-3-medium.v1", ComfyCloudWorkflowInputs(prompt=prompt, duration_seconds=duration_seconds, seed=seed, expand_prompt=expand_prompt, category=category)) + + +class ComfyCloudChatterboxMultilingualVoiceCloneNode(IO.ComfyNode): + @classmethod + def define_schema(cls) -> IO.Schema: + return _audio_schema( + "ComfyCloudChatterboxMultilingualVoiceCloneNode", + "Chatterbox Multilingual Voice Clone", + [ + _prompt_input("text"), IO.Audio.Input("voice_reference"), + IO.Combo.Input("language", options=_CHATTERBOX_LANGUAGES, default="English (en)"), + IO.Float.Input("exaggeration", default=0.5, min=0, max=2, step=0.05), + IO.Float.Input("cfg_weight", default=0.5, min=0, max=1, step=0.05), + IO.Float.Input("temperature", default=0.8, min=0.05, max=2, step=0.05), + IO.Int.Input("seed", default=0, min=0, max=_UINT32_MAX, control_after_generate=True), + ], + ) + + @classmethod + async def execute(cls, text: str, voice_reference: Input.Audio, language: str, exaggeration: float, cfg_weight: float, temperature: float, seed: int) -> IO.NodeOutput: + values = _validate_node_inputs(cls, locals()) + text = values["text"] + _validate_audio_duration("Voice reference", voice_reference, 1, 30) + return await _run_audio_workflow(cls, "audio.chatterbox-multilingual-voice-clone.v1", ComfyCloudWorkflowInputs(text=text, assets=await _audio_asset(cls, "voice_reference", voice_reference), language=language, exaggeration=exaggeration, cfg_weight=cfg_weight, temperature=temperature, seed=seed)) + + +class ComfyCloudChatterboxDialogueNode(IO.ComfyNode): + @classmethod + def define_schema(cls) -> IO.Schema: + return _audio_schema( + "ComfyCloudChatterboxDialogueNode", + "Chatterbox Dialogue", + [ + _prompt_input("script"), IO.Audio.Input("speaker_a_reference"), IO.Audio.Input("speaker_b_reference"), + IO.Float.Input("exaggeration", default=0.5, min=0.25, max=2, step=0.05), + IO.Float.Input("cfg_weight", default=0.5, min=0.2, max=1, step=0.05), + IO.Float.Input("temperature", default=0.8, min=0.05, max=5, step=0.05), + IO.Int.Input("seed", default=0, min=0, max=_UINT32_MAX, control_after_generate=True), + ], + ) + + @classmethod + async def execute(cls, script: str, speaker_a_reference: Input.Audio, speaker_b_reference: Input.Audio, exaggeration: float, cfg_weight: float, temperature: float, seed: int) -> IO.NodeOutput: + values = _validate_node_inputs(cls, locals()) + script = values["script"] + script = _normalize_dialogue(script) + validate_string(script, min_length=1, max_length=10000, field_name="script") + _validate_audio_upload(speaker_a_reference) + _validate_audio_upload(speaker_b_reference) + _validate_audio_duration("Speaker A reference", speaker_a_reference, 1, 30) + _validate_audio_duration("Speaker B reference", speaker_b_reference, 1, 30) + assets = { + "speaker_a_reference": ComfyCloudAssetInput(type="AUDIO", url=await upload_audio_to_comfyapi(cls, speaker_a_reference)), + "speaker_b_reference": ComfyCloudAssetInput(type="AUDIO", url=await upload_audio_to_comfyapi(cls, speaker_b_reference)), + } + return await _run_audio_workflow(cls, "audio.chatterbox-dialogue.v1", ComfyCloudWorkflowInputs(script=script, assets=assets, exaggeration=exaggeration, cfg_weight=cfg_weight, temperature=temperature, seed=seed)) + + +class ComfyCloudChatterboxVoiceConversionNode(IO.ComfyNode): + @classmethod + def define_schema(cls) -> IO.Schema: + return _audio_schema( + "ComfyCloudChatterboxVoiceConversionNode", + "Chatterbox Voice Conversion", + [IO.Audio.Input("source_audio"), IO.Audio.Input("target_voice_reference"), IO.Int.Input("seed", default=0, min=0, max=_UINT32_MAX, control_after_generate=True)], + ) + + @classmethod + async def execute(cls, source_audio: Input.Audio, target_voice_reference: Input.Audio, seed: int) -> IO.NodeOutput: + _validate_node_inputs(cls, locals()) + _validate_audio_upload(source_audio) + _validate_audio_upload(target_voice_reference) + _validate_audio_duration("Source audio", source_audio, 0.5, 300) + _validate_audio_duration("Target voice reference", target_voice_reference, 1, 30) + assets = { + "source_audio": ComfyCloudAssetInput(type="AUDIO", url=await upload_audio_to_comfyapi(cls, source_audio)), + "target_voice_reference": ComfyCloudAssetInput(type="AUDIO", url=await upload_audio_to_comfyapi(cls, target_voice_reference)), + } + return await _run_audio_workflow(cls, "audio.chatterbox-voice-conversion.v1", ComfyCloudWorkflowInputs(assets=assets, seed=seed)) + + +class ComfyCloudMelBandRoFormerStemSeparationNode(IO.ComfyNode): + @classmethod + def define_schema(cls) -> IO.Schema: + return _audio_schema( + "ComfyCloudMelBandRoFormerStemSeparationNode", + "MelBandRoFormer Stem Separation", + [IO.Audio.Input("audio")], + [IO.Audio.Output("vocals"), IO.Audio.Output("instruments")], + ) + + @classmethod + async def execute(cls, audio: Input.Audio) -> IO.NodeOutput: + _validate_audio_duration("Audio", audio, 0.5, 600) + return await _run_audio_workflow(cls, "audio.melbandroformer-stem-separation.v1", ComfyCloudWorkflowInputs(assets=await _audio_asset(cls, "audio", audio)), ("vocals", "instruments")) + + +async def _run_3d_workflow(cls: type[IO.ComfyNode], workflow: ComfyCloudWorkflow, inputs: ComfyCloudWorkflowInputs, file_format: str) -> IO.NodeOutput: + task = await sync_op(cls, _GENERATE_ENDPOINT, response_model=ComfyCloudGenerateResponse, data=ComfyCloudGenerateRequest(workflow=workflow, inputs=inputs)) + polling_endpoint, cancel_endpoint = _task_endpoints(task.task_id) + result = await poll_op( + cls, + polling_endpoint, + response_model=ComfyCloudStatusResponse, + status_extractor=lambda response: response.status, + progress_extractor=lambda response: response.progress, + cancel_endpoint=cancel_endpoint, + ) + if not result.output_url: + raise RuntimeError("Comfy Cloud task completed without an output URL.") + return IO.NodeOutput( + await download_url_to_file_3d( + _validated_output_url(result.output_url), + file_format, + timeout=_OUTPUT_DOWNLOAD_TIMEOUT, + cls=cls, + allow_redirects=False, + ) + ) + + +def _3d_schema(node_id: str, display_name: str, inputs: list[IO.Input], output: IO.Output) -> IO.Schema: + return IO.Schema( + node_id=node_id, + display_name=display_name, + category="partner/3d/Comfy Cloud", + description="Runs this 3D workflow on Comfy Cloud and returns the generated 3D file." + + _COMFY_CLOUD_RATE_DESCRIPTION, + inputs=_with_input_sockets(inputs), + outputs=[output], + hidden=[IO.Hidden.auth_token_comfy_org, IO.Hidden.api_key_comfy_org, IO.Hidden.unique_id], + is_api_node=True, + price_badge=_COMFY_CLOUD_PRICE_BADGE, + ) + + +async def _image_asset(cls: type[IO.ComfyNode], name: str, image: Input.Image, wait_label: str | None = None) -> dict[str, ComfyCloudAssetInput]: + if get_number_of_images(image) != 1: + raise ValueError(f"Exactly one {name.replace('_', ' ')} is required.") + _validate_image_upload(image) + kwargs = {"total_pixels": None} + if wait_label is not None: + kwargs["wait_label"] = wait_label + return {name: ComfyCloudAssetInput(type="IMAGE", url=await upload_image_to_comfyapi(cls, image, **kwargs))} + + +class ComfyCloudTripoSplatImageToGaussianSplatNode(IO.ComfyNode): + @classmethod + def define_schema(cls) -> IO.Schema: + return _3d_schema( + "ComfyCloudTripoSplatImageToGaussianSplatNode", + "TripoSplat Image to Gaussian Splat", + [ + IO.Image.Input("image"), + IO.Boolean.Input("remove_background", default=True), + IO.Int.Input("seed", default=46, min=0, max=_UINT64_MAX, control_after_generate=True), + IO.Int.Input("gaussian_count", default=262144, min=32768, max=262144), + ], + IO.File3DSPZ.Output(tooltip="SPZ Gaussian splat (.spz; POC MIME application/octet-stream)."), + ) + + @classmethod + async def execute(cls, image: Input.Image, remove_background: bool, seed: int, gaussian_count: int) -> IO.NodeOutput: + _validate_node_inputs(cls, locals()) + return await _run_3d_workflow(cls, "3d.triposplat-image-to-gaussian-splat.v1", ComfyCloudWorkflowInputs(assets=await _image_asset(cls, "image", image), remove_background=remove_background, seed=seed, gaussian_count=gaussian_count), "spz") + + +class ComfyCloudHunyuan3D21ImageTo3DNode(IO.ComfyNode): + @classmethod + def define_schema(cls) -> IO.Schema: + return _3d_schema( + "ComfyCloudHunyuan3D21ImageTo3DNode", + "Hunyuan3D 2.1 Image to 3D", + [IO.Image.Input("image"), IO.Int.Input("seed", default=952805179515179, min=0, max=_UINT64_MAX, control_after_generate=True)], + IO.File3DGLB.Output(), + ) + + @classmethod + async def execute(cls, image: Input.Image, seed: int) -> IO.NodeOutput: + _validate_node_inputs(cls, locals()) + return await _run_3d_workflow(cls, "3d.hunyuan3d-2-1-image-to-3d.v1", ComfyCloudWorkflowInputs(assets=await _image_asset(cls, "image", image), seed=seed), "glb") + + +class ComfyCloudHunyuan3DMultiViewTo3DNode(IO.ComfyNode): + @classmethod + def define_schema(cls) -> IO.Schema: + return _3d_schema( + "ComfyCloudHunyuan3DMultiViewTo3DNode", + "Hunyuan3D Multi-View to 3D", + [IO.Image.Input("front_image"), IO.Image.Input("back_image"), IO.Int.Input("seed", default=502126049100058, min=0, max=_UINT64_MAX, control_after_generate=True)], + IO.File3DGLB.Output(), + ) + + @classmethod + async def execute(cls, front_image: Input.Image, back_image: Input.Image, seed: int) -> IO.NodeOutput: + _validate_node_inputs(cls, locals()) + if get_number_of_images(front_image) != 1 or get_number_of_images(back_image) != 1: + raise ValueError("Exactly one front image and one back image are required.") + assets = await _image_asset(cls, "front_image", front_image, "Uploading front image") + assets.update(await _image_asset(cls, "back_image", back_image, "Uploading back image")) + return await _run_3d_workflow(cls, "3d.hunyuan3d-multiview-to-3d.v1", ComfyCloudWorkflowInputs(assets=assets, seed=seed), "glb") + + +class ComfyCloudMoGe2PhotoToTexturedMeshNode(IO.ComfyNode): + @classmethod + def define_schema(cls) -> IO.Schema: + return _3d_schema( + "ComfyCloudMoGe2PhotoToTexturedMeshNode", + "MoGe 2 Photo to Textured Mesh", + [ + IO.Image.Input("image"), + IO.Float.Input("fov_degrees", default=0, min=0, max=170, step=0.1, tooltip="0 selects automatic field-of-view estimation."), + IO.Int.Input("detail", default=9, min=0, max=9), + IO.Int.Input("mesh_decimation", default=1, min=1, max=8), + IO.Float.Input("gap_threshold", default=0.04, min=0, max=1, step=0.01), + IO.Boolean.Input("texture", default=True), + ], + IO.File3DGLB.Output(), + ) + + @classmethod + async def execute(cls, image: Input.Image, fov_degrees: float, detail: int, mesh_decimation: int, gap_threshold: float, texture: bool) -> IO.NodeOutput: + _validate_node_inputs(cls, locals()) + inputs = ComfyCloudWorkflowInputs(assets=await _image_asset(cls, "image", image), fov_degrees=fov_degrees, detail=detail, mesh_decimation=mesh_decimation, gap_threshold=gap_threshold, texture=texture) + return await _run_3d_workflow(cls, "3d.moge-2-photo-to-textured-mesh.v1", inputs, "glb") + + +class ComfyCloudMoGe2PanoramaTo3DSceneNode(IO.ComfyNode): + @classmethod + def define_schema(cls) -> IO.Schema: + return _3d_schema( + "ComfyCloudMoGe2PanoramaTo3DSceneNode", + "MoGe 2 Panorama to 3D Scene", + [ + IO.Image.Input("panorama", tooltip="Equirectangular panorama."), + IO.Int.Input("detail", default=5, min=0, max=9), + IO.Int.Input("split_resolution", default=512, min=256, max=1024), + IO.Int.Input("merge_resolution", default=1024, min=256, max=8192), + IO.Int.Input("mesh_decimation", default=1, min=1, max=8), + IO.Float.Input("gap_threshold", default=0.04, min=0, max=1, step=0.01), + IO.Boolean.Input("texture", default=True), + ], + IO.File3DGLB.Output(), + ) + + @classmethod + async def execute(cls, panorama: Input.Image, detail: int, split_resolution: int, merge_resolution: int, mesh_decimation: int, gap_threshold: float, texture: bool) -> IO.NodeOutput: + _validate_node_inputs(cls, locals()) + inputs = ComfyCloudWorkflowInputs(assets=await _image_asset(cls, "panorama", panorama), detail=detail, split_resolution=split_resolution, merge_resolution=merge_resolution, mesh_decimation=mesh_decimation, gap_threshold=gap_threshold, texture=texture) + return await _run_3d_workflow(cls, "3d.moge-2-panorama-to-3d-scene.v1", inputs, "glb") + + +class ComfyCloudExtension(ComfyExtension): + @override + async def get_node_list(self) -> list[type[IO.ComfyNode]]: + return [ + ComfyCloudTextToImageNode, + ComfyCloudTextToVideoNode, + ComfyCloudImageToVideoNode, + ComfyCloudImageEditNode, + ComfyCloudIdeogram4DesignNode, + ComfyCloudKrea2CreativeImageNode, + ComfyCloudMageFlowImageNode, + ComfyCloudFlux2ReferenceEditNode, + ComfyCloudQwenImageEdit2511Node, + ComfyCloudSeedVR2ImageUpscaleNode, + ComfyCloudMiniMaxH3TextSoundNode, + ComfyCloudMiniMaxH3ImageSoundNode, + ComfyCloudLTX23ImageAudioPerformanceNode, + ComfyCloudLTX23FirstLastFrameNode, + ComfyCloudWan22FirstLastFrameNode, + ComfyCloudSCAIL2CharacterReplacementNode, + ComfyCloudACEStep15XLTurboNode, + ComfyCloudStableAudio3MediumNode, + ComfyCloudChatterboxMultilingualVoiceCloneNode, + ComfyCloudChatterboxDialogueNode, + ComfyCloudChatterboxVoiceConversionNode, + ComfyCloudMelBandRoFormerStemSeparationNode, + ComfyCloudTripoSplatImageToGaussianSplatNode, + ComfyCloudHunyuan3D21ImageTo3DNode, + ComfyCloudHunyuan3DMultiViewTo3DNode, + ComfyCloudMoGe2PhotoToTexturedMeshNode, + ComfyCloudMoGe2PanoramaTo3DSceneNode, + ] + + +async def comfy_entrypoint() -> ComfyCloudExtension: + return ComfyCloudExtension() diff --git a/comfy_api_nodes/util/__init__.py b/comfy_api_nodes/util/__init__.py index 1fb6b96cf..954bb789f 100644 --- a/comfy_api_nodes/util/__init__.py +++ b/comfy_api_nodes/util/__init__.py @@ -32,6 +32,7 @@ from .conversions import ( ) from .download_helpers import ( download_url_as_bytesio, + download_url_to_audio_input, download_url_to_bytesio, download_url_to_file_3d, download_url_to_image_tensor, @@ -76,6 +77,7 @@ __all__ = [ "upload_video_to_comfyapi", # Download helpers "download_url_as_bytesio", + "download_url_to_audio_input", "download_url_to_bytesio", "download_url_to_file_3d", "download_url_to_image_tensor", diff --git a/comfy_api_nodes/util/download_helpers.py b/comfy_api_nodes/util/download_helpers.py index 0ec3c6e66..3edf66fad 100644 --- a/comfy_api_nodes/util/download_helpers.py +++ b/comfy_api_nodes/util/download_helpers.py @@ -11,7 +11,7 @@ import torch from aiohttp.client_exceptions import ClientError, ContentTypeError from comfy_api.latest import IO as COMFY_IO -from comfy_api.latest import InputImpl, Types +from comfy_api.latest import Input, InputImpl, Types from folder_paths import get_output_directory from . import request_logger @@ -24,9 +24,10 @@ from ._helpers import ( ) from .client import _diagnose_connectivity from .common_exceptions import ApiServerError, LocalNetworkError, ProcessingInterrupted -from .conversions import bytesio_to_image_tensor +from .conversions import audio_bytes_to_audio_input, bytesio_to_image_tensor _RETRY_STATUS = {408, 429, 500, 502, 503, 504} +_MAX_IN_MEMORY_DOWNLOAD_BYTES = 512 * 1024 * 1024 async def download_url_to_bytesio( @@ -38,6 +39,7 @@ async def download_url_to_bytesio( retry_delay: float = 1.0, retry_backoff: float = 2.0, cls: type[COMFY_IO.ComfyNode] = None, + allow_redirects: bool = True, ) -> None: """Stream-download a URL to `dest`. @@ -58,6 +60,8 @@ async def download_url_to_bytesio( attempt = 0 delay = retry_delay headers: dict[str, str] = {} + is_path_sink = isinstance(dest, (str, Path)) + can_reset_sink = is_path_sink or (callable(getattr(dest, "seek", None)) and callable(getattr(dest, "truncate", None))) parsed_url = urlparse(url) if not parsed_url.scheme and not parsed_url.netloc: # is URL relative? @@ -68,10 +72,12 @@ async def download_url_to_bytesio( while True: attempt += 1 + if not is_path_sink and can_reset_sink: + dest.seek(0) + dest.truncate(0) op_id = _generate_operation_id("GET", url, attempt) timeout_cfg = aiohttp.ClientTimeout(total=timeout) - is_path_sink = isinstance(dest, (str, Path)) fhandle = None session: aiohttp.ClientSession | None = None stop_evt: asyncio.Event | None = None @@ -96,7 +102,9 @@ async def download_url_to_bytesio( monitor_task = asyncio.create_task(_monitor()) - req_task = asyncio.create_task(session.get(to_aiohttp_url(url), headers=headers)) + req_task = asyncio.create_task( + session.get(to_aiohttp_url(url), headers=headers, allow_redirects=allow_redirects) + ) done, pending = await asyncio.wait({req_task, monitor_task}, return_when=asyncio.FIRST_COMPLETED) if monitor_task in done and req_task in pending: @@ -111,7 +119,7 @@ async def download_url_to_bytesio( raise ProcessingInterrupted("Task cancelled") from None async with resp: - if resp.status >= 400: + if resp.status >= 300: with contextlib.suppress(Exception): try: body = await resp.json() @@ -129,11 +137,17 @@ async def download_url_to_bytesio( ) if resp.status in _RETRY_STATUS and attempt <= max_retries: + if not can_reset_sink: + raise Exception(f"Failed to download (HTTP {resp.status}); destination cannot be reset for retry.") await sleep_with_interrupt(delay, cls, None, None, None) delay *= retry_backoff continue raise Exception(f"Failed to download (HTTP {resp.status}).") + max_bytes = None if is_path_sink else _MAX_IN_MEMORY_DOWNLOAD_BYTES + if max_bytes is not None and resp.content_length is not None and resp.content_length > max_bytes: + raise ValueError(f"Download exceeds the {max_bytes}-byte in-memory limit.") + if is_path_sink: p = Path(str(dest)) with contextlib.suppress(Exception): @@ -160,10 +174,12 @@ async def download_url_to_bytesio( break continue - sink.write(chunk) written += len(chunk) + if max_bytes is not None and written > max_bytes: + raise ValueError(f"Download exceeds the {max_bytes}-byte in-memory limit.") + sink.write(chunk) - if isinstance(dest, BytesIO): + if not is_path_sink and hasattr(dest, "seek"): with contextlib.suppress(Exception): dest.seek(0) @@ -180,6 +196,8 @@ async def download_url_to_bytesio( raise ProcessingInterrupted("Task cancelled") from None except (ClientError, OSError) as e: if attempt <= max_retries: + if not can_reset_sink: + raise ApiServerError("The download failed and its destination cannot be reset for retry.") from e request_logger.log_request_response( operation_id=op_id, request_method="GET", @@ -221,10 +239,11 @@ async def download_url_to_image_tensor( *, timeout: float = None, cls: type[COMFY_IO.ComfyNode] = None, + allow_redirects: bool = True, ) -> torch.Tensor: """Downloads an image from a URL and returns a [B, H, W, C] tensor.""" result = BytesIO() - await download_url_to_bytesio(url, result, timeout=timeout, cls=cls) + await download_url_to_bytesio(url, result, timeout=timeout, cls=cls, allow_redirects=allow_redirects) return bytesio_to_image_tensor(result) @@ -234,13 +253,42 @@ async def download_url_to_video_output( timeout: float = None, max_retries: int = 5, cls: type[COMFY_IO.ComfyNode] = None, + allow_redirects: bool = True, ) -> InputImpl.VideoFromFile: """Downloads a video from a URL and returns a `VIDEO` output.""" result = BytesIO() - await download_url_to_bytesio(video_url, result, timeout=timeout, max_retries=max_retries, cls=cls) + await download_url_to_bytesio( + video_url, + result, + timeout=timeout, + max_retries=max_retries, + cls=cls, + allow_redirects=allow_redirects, + ) return InputImpl.VideoFromFile(result) +async def download_url_to_audio_input( + audio_url: str, + *, + timeout: float = None, + max_retries: int = 5, + cls: type[COMFY_IO.ComfyNode] = None, + allow_redirects: bool = True, +) -> Input.Audio: + """Downloads audio from a URL and decodes it into a Comfy AUDIO input.""" + result = BytesIO() + await download_url_to_bytesio( + audio_url, + result, + timeout=timeout, + max_retries=max_retries, + cls=cls, + allow_redirects=allow_redirects, + ) + return audio_bytes_to_audio_input(result.getvalue()) + + async def download_url_as_bytesio( url: str, *, @@ -270,6 +318,7 @@ async def download_url_to_file_3d( timeout: float | None = None, max_retries: int = 5, cls: type[COMFY_IO.ComfyNode] = None, + allow_redirects: bool = True, ) -> Types.File3D: """Downloads a 3D model file from a URL into memory as BytesIO. @@ -284,6 +333,7 @@ async def download_url_to_file_3d( timeout=timeout, max_retries=max_retries, cls=cls, + allow_redirects=allow_redirects, ) if task_id is not None: diff --git a/tests-unit/comfy_api_nodes_test/comfy_cloud_test.py b/tests-unit/comfy_api_nodes_test/comfy_cloud_test.py new file mode 100644 index 000000000..4a8efb625 --- /dev/null +++ b/tests-unit/comfy_api_nodes_test/comfy_cloud_test.py @@ -0,0 +1,1020 @@ +import asyncio +from io import BytesIO +from typing import get_args +from unittest.mock import AsyncMock, Mock + +import aiohttp +import pytest +import torch + +from comfy.cli_args import args + +if not torch.cuda.is_available(): + args.cpu = True + +from comfy_api_nodes.apis.comfy_cloud import ( + ComfyCloudGenerateRequest, + ComfyCloudGenerateResponse, + ComfyCloudStatusResponse, + ComfyCloudWorkflow, + ComfyCloudWorkflowInputs, +) +from comfy_api_nodes import nodes_comfy_cloud +from comfy_api_nodes.util import download_helpers + + +@pytest.mark.parametrize( + ("node", "workflow", "returns_video", "requires_image"), + [ + (nodes_comfy_cloud.ComfyCloudTextToImageNode, "text-to-image", False, False), + (nodes_comfy_cloud.ComfyCloudTextToVideoNode, "text-to-video", True, False), + (nodes_comfy_cloud.ComfyCloudImageToVideoNode, "image-to-video", True, True), + (nodes_comfy_cloud.ComfyCloudImageEditNode, "image-edit", False, True), + ], +) +def test_workflow_submission_polling_and_download(monkeypatch, node, workflow, returns_video, requires_image): + sync = AsyncMock( + return_value=ComfyCloudGenerateResponse( + task_id="task-1", + status="queued", + polling_url="/proxy/comfy-cloud/workflow/tasks/task-1", + cancel_url="/proxy/comfy-cloud/workflow/tasks/task-1/cancel", + ) + ) + poll = AsyncMock( + return_value=ComfyCloudStatusResponse( + task_id="task-1", + status="completed", + progress=100, + output_url="/proxy/comfy-cloud/results/task-1/output", + ) + ) + upload = AsyncMock(return_value="https://example.com/input.png") + image_download = AsyncMock(return_value="image-output") + video_download = AsyncMock(return_value="video-output") + monkeypatch.setattr(nodes_comfy_cloud, "sync_op", sync) + monkeypatch.setattr(nodes_comfy_cloud, "poll_op", poll) + monkeypatch.setattr(nodes_comfy_cloud, "upload_image_to_comfyapi", upload) + monkeypatch.setattr(nodes_comfy_cloud, "download_url_to_image_tensor", image_download) + monkeypatch.setattr(nodes_comfy_cloud, "download_url_to_video_output", video_download) + monkeypatch.setattr(nodes_comfy_cloud, "get_number_of_images", lambda image: 1) + + image = object() if requires_image else None + output = asyncio.run(node.execute("A tiny fennec fox", image)) + + endpoint = sync.call_args.args[1] + request = sync.call_args.kwargs["data"] + assert endpoint.path == "/proxy/comfy-cloud/workflow/generate" + assert endpoint.method == "POST" + assert request == ComfyCloudGenerateRequest( + workflow=workflow, + inputs=ComfyCloudWorkflowInputs( + prompt="A tiny fennec fox", + image_url="https://example.com/input.png" if requires_image else None, + ), + ) + assert upload.await_count == int(requires_image) + + poll_endpoint = poll.call_args.args[1] + cancel_endpoint = poll.call_args.kwargs["cancel_endpoint"] + assert poll_endpoint.path == "/proxy/comfy-cloud/workflow/tasks/task-1" + assert cancel_endpoint.path == "/proxy/comfy-cloud/workflow/tasks/task-1/cancel" + assert cancel_endpoint.method == "POST" + assert output[0] == ("video-output" if returns_video else "image-output") + + +@pytest.mark.parametrize( + "node", + [nodes_comfy_cloud.ComfyCloudImageToVideoNode, nodes_comfy_cloud.ComfyCloudImageEditNode], +) +def test_image_workflows_reject_batches(monkeypatch, node): + upload = AsyncMock() + monkeypatch.setattr(nodes_comfy_cloud, "get_number_of_images", lambda image: 2) + monkeypatch.setattr(nodes_comfy_cloud, "upload_image_to_comfyapi", upload) + + with pytest.raises(ValueError, match="Exactly one input image"): + asyncio.run(node.execute("Animate this", object())) + upload.assert_not_awaited() + + +def test_contract_omits_optional_status_fields(): + request = ComfyCloudGenerateRequest( + workflow="text-to-image", + inputs=ComfyCloudWorkflowInputs(prompt="A lighthouse"), + ) + status = ComfyCloudStatusResponse(task_id="task-1", status="queued") + + assert request.model_dump(exclude_none=True) == { + "workflow": "text-to-image", + "inputs": {"prompt": "A lighthouse"}, + } + assert status.model_dump(exclude_none=True) == {"task_id": "task-1", "status": "queued"} + + +@pytest.mark.parametrize("response_model", [ComfyCloudGenerateResponse, ComfyCloudStatusResponse]) +@pytest.mark.parametrize("task_id", ["", " "]) +def test_contract_rejects_empty_task_ids(response_model, task_id): + values = {"task_id": task_id, "status": "queued"} + if response_model is ComfyCloudGenerateResponse: + values.update(polling_url="/poll", cancel_url="/cancel") + + with pytest.raises(ValueError, match="task_id"): + response_model(**values) + + +@pytest.mark.parametrize( + "url", + [ + "http://example.com/output.png", + "http://127.0.0.1/output.png", + "//169.254.169.254/latest/meta-data", + "/unrelated/path/output.png", + "https://user@example.com/output.png", + ], +) +def test_cloud_workflows_reject_untrusted_output_urls(monkeypatch, url): + sync = AsyncMock( + return_value=ComfyCloudGenerateResponse( + task_id="task-1", + status="queued", + polling_url="/poll", + cancel_url="/cancel", + ) + ) + poll = AsyncMock( + return_value=ComfyCloudStatusResponse(task_id="task-1", status="completed", output_url=url) + ) + download = AsyncMock() + monkeypatch.setattr(nodes_comfy_cloud, "sync_op", sync) + monkeypatch.setattr(nodes_comfy_cloud, "poll_op", poll) + monkeypatch.setattr(nodes_comfy_cloud, "download_url_to_image_tensor", download) + + with pytest.raises(RuntimeError, match="invalid output URL"): + asyncio.run(nodes_comfy_cloud.ComfyCloudTextToImageNode.execute("prompt")) + download.assert_not_awaited() + + +def test_cloud_workflows_accept_signed_https_output_urls(monkeypatch): + sync = AsyncMock( + return_value=ComfyCloudGenerateResponse( + task_id="task-1", + status="queued", + polling_url="/poll", + cancel_url="/cancel", + ) + ) + poll = AsyncMock( + return_value=ComfyCloudStatusResponse( + task_id="task-1", + status="completed", + output_url="https://storage.googleapis.com/comfy-cloud/output.png?signature=example", + ) + ) + download = AsyncMock(return_value="image-output") + monkeypatch.setattr(nodes_comfy_cloud, "sync_op", sync) + monkeypatch.setattr(nodes_comfy_cloud, "poll_op", poll) + monkeypatch.setattr(nodes_comfy_cloud, "download_url_to_image_tensor", download) + + output = asyncio.run(nodes_comfy_cloud.ComfyCloudTextToImageNode.execute("prompt")) + + assert output[0] == "image-output" + download.assert_awaited_once_with( + "https://storage.googleapis.com/comfy-cloud/output.png?signature=example", + timeout=nodes_comfy_cloud._OUTPUT_DOWNLOAD_TIMEOUT, + cls=nodes_comfy_cloud.ComfyCloudTextToImageNode, + allow_redirects=False, + ) + + +@pytest.mark.parametrize( + "node", + [ + nodes_comfy_cloud.ComfyCloudTextToImageNode, + nodes_comfy_cloud.ComfyCloudTextToVideoNode, + nodes_comfy_cloud.ComfyCloudImageToVideoNode, + nodes_comfy_cloud.ComfyCloudImageEditNode, + ], +) +def test_legacy_nodes_reject_oversized_prompts(monkeypatch, node): + sync = AsyncMock() + monkeypatch.setattr(nodes_comfy_cloud, "sync_op", sync) + + with pytest.raises(Exception, match="4096"): + asyncio.run(node.execute("x" * 4097, object())) + sync.assert_not_awaited() + + +def test_legacy_nodes_strip_prompts_before_submission(monkeypatch): + run = AsyncMock(return_value=("output",)) + monkeypatch.setattr(nodes_comfy_cloud.ComfyCloudTextToImageNode, "_run", run) + + asyncio.run(nodes_comfy_cloud.ComfyCloudTextToImageNode.execute(" prompt ")) + + assert run.call_args.args[0].prompt == "prompt" + + +def test_poc_nodes_strip_prompt_fields_before_submission(monkeypatch): + run = AsyncMock(return_value=("output",)) + monkeypatch.setattr(nodes_comfy_cloud.ComfyCloudMageFlowImageNode, "_run", run) + + asyncio.run( + nodes_comfy_cloud.ComfyCloudMageFlowImageNode.execute( + " prompt ", " avoid this ", "1:1", 0 + ) + ) + + inputs = run.call_args.args[0] + assert inputs.prompt == "prompt" + assert inputs.negative_prompt == "avoid this" + + +def test_task_routes_ignore_response_urls_and_errors_hide_task_token(monkeypatch): + sync = AsyncMock( + return_value=ComfyCloudGenerateResponse( + task_id="secret/task-token", + status="queued", + polling_url="https://attacker.example/poll", + cancel_url="https://attacker.example/cancel", + ) + ) + poll = AsyncMock( + return_value=ComfyCloudStatusResponse( + task_id="secret/task-token", + status="completed", + error="provider details with secret/task-token", + ) + ) + monkeypatch.setattr(nodes_comfy_cloud, "sync_op", sync) + monkeypatch.setattr(nodes_comfy_cloud, "poll_op", poll) + + with pytest.raises(RuntimeError) as error: + asyncio.run(nodes_comfy_cloud.ComfyCloudTextToVideoNode.execute("A prompt")) + + assert poll.call_args.args[1].path == "/proxy/comfy-cloud/workflow/tasks/secret%2Ftask-token" + assert poll.call_args.kwargs["cancel_endpoint"].path == "/proxy/comfy-cloud/workflow/tasks/secret%2Ftask-token/cancel" + assert "task-token" not in str(error.value) + assert "provider details" not in str(error.value) + + +IMAGE_POC_NODES = [ + ( + nodes_comfy_cloud.ComfyCloudIdeogram4DesignNode, + "image.ideogram-4-design.v1", + ["prompt", "aspect_ratio", "quality_mode", "seed"], + { + "prompt": "A geometric fox logo", + "aspect_ratio": "21:9", + "quality_mode": "fast", + "seed": 11, + }, + ), + ( + nodes_comfy_cloud.ComfyCloudKrea2CreativeImageNode, + "image.krea-2-creative-image.v1", + ["prompt", "prompt_enhance", "aspect_ratio", "seed"], + {"prompt": "A glass forest", "prompt_enhance": False, "aspect_ratio": "16:9", "seed": 12}, + ), + ( + nodes_comfy_cloud.ComfyCloudMageFlowImageNode, + "image.mage-flow-image.v1", + ["prompt", "negative_prompt", "aspect_ratio", "seed"], + {"prompt": "A moonlit lake", "negative_prompt": "fog", "aspect_ratio": "3:2", "seed": 13}, + ), + ( + nodes_comfy_cloud.ComfyCloudFlux2ReferenceEditNode, + "image.flux-2-reference-edit.v1", + ["image", "instruction", "guidance", "quality_mode", "seed"], + {"image": object(), "instruction": "Make it winter", "guidance": 5.5, "quality_mode": "fast", "seed": 14}, + ), + ( + nodes_comfy_cloud.ComfyCloudQwenImageEdit2511Node, + "image.qwen-image-edit-2511.v1", + ["image", "instruction", "quality_mode", "seed"], + {"image": object(), "instruction": "Remove the sign", "quality_mode": "fast", "seed": 15}, + ), + ( + nodes_comfy_cloud.ComfyCloudSeedVR2ImageUpscaleNode, + "image.seedvr2-image-upscale.v1", + ["image", "scale"], + {"image": object(), "scale": "2x"}, + ), +] + + +@pytest.mark.parametrize(("node", "workflow", "input_names", "arguments"), IMAGE_POC_NODES) +def test_image_poc_node_schema_and_request_mapping(monkeypatch, node, workflow, input_names, arguments): + sync = AsyncMock( + return_value=ComfyCloudGenerateResponse( + task_id="task-poc", + status="queued", + polling_url="/tasks/task-poc", + cancel_url="/tasks/task-poc/cancel", + ) + ) + poll = AsyncMock( + return_value=ComfyCloudStatusResponse( + task_id="task-poc", + status="completed", + output_url="/proxy/comfy-cloud/results/task-poc/image.png", + ) + ) + upload = AsyncMock(return_value="/uploads/input.png") + download = AsyncMock(return_value="image-output") + monkeypatch.setattr(nodes_comfy_cloud, "sync_op", sync) + monkeypatch.setattr(nodes_comfy_cloud, "poll_op", poll) + monkeypatch.setattr(nodes_comfy_cloud, "upload_image_to_comfyapi", upload) + monkeypatch.setattr(nodes_comfy_cloud, "download_url_to_image_tensor", download) + monkeypatch.setattr(nodes_comfy_cloud, "get_number_of_images", lambda image: 1) + + schema = node.define_schema() + assert schema.node_id == node.node_id + assert schema.display_name == node.display_name + assert schema.is_api_node is True + assert [input.id for input in schema.inputs] == input_names + assert len(schema.outputs) == 1 + assert schema.outputs[0].get_io_type() == "IMAGE" + + output = asyncio.run(node.execute(**arguments)) + request = sync.call_args.kwargs["data"] + expected_inputs = {key: value for key, value in arguments.items() if key != "image"} + if "image" in arguments: + expected_inputs["assets"] = {"image": {"type": "IMAGE", "url": "/uploads/input.png"}} + + assert request.workflow == workflow + assert request.inputs.model_dump(exclude_none=True) == expected_inputs + assert "asset_id" not in request.model_dump_json() + assert '"id"' not in request.model_dump_json() + assert upload.await_count == int("image" in arguments) + if "image" in arguments: + assert upload.call_args.kwargs == {"total_pixels": None} + download.assert_awaited_once_with( + "/proxy/comfy-cloud/results/task-poc/image.png", + timeout=30 * 60, + cls=node, + allow_redirects=False, + ) + assert output[0] == "image-output" + + +def test_image_poc_schema_defaults_ranges_and_enums(): + schemas = { + node.workflow: {input.id: input for input in node.define_schema().inputs} + for node, _, _, _ in IMAGE_POC_NODES + } + aspect_ratios = ["1:1", "3:4", "2:3", "3:2", "4:3", "16:9", "9:16", "21:9"] + + for workflow in [ + "image.ideogram-4-design.v1", + "image.krea-2-creative-image.v1", + "image.mage-flow-image.v1", + ]: + assert schemas[workflow]["aspect_ratio"].options == aspect_ratios + assert schemas[workflow]["aspect_ratio"].default == "1:1" + seed = schemas[workflow]["seed"] + assert (seed.default, seed.min, seed.max) == (0, 0, 0xFFFFFFFFFFFFFFFF) + + assert schemas["image.ideogram-4-design.v1"]["quality_mode"].options == ["quality", "balanced", "fast"] + assert schemas["image.ideogram-4-design.v1"]["quality_mode"].default == "balanced" + assert schemas["image.krea-2-creative-image.v1"]["prompt_enhance"].default is True + assert schemas["image.mage-flow-image.v1"]["negative_prompt"].default == "" + + guidance = schemas["image.flux-2-reference-edit.v1"]["guidance"] + assert (guidance.default, guidance.min, guidance.max, guidance.step) == (4.0, 1.0, 10.0, 0.1) + for workflow in ["image.flux-2-reference-edit.v1", "image.qwen-image-edit-2511.v1"]: + assert schemas[workflow]["quality_mode"].options == ["quality", "fast"] + assert schemas[workflow]["quality_mode"].default == "quality" + seed = schemas[workflow]["seed"] + assert (seed.default, seed.min, seed.max) == (0, 0, 0xFFFFFFFFFFFFFFFF) + + scale = schemas["image.seedvr2-image-upscale.v1"]["scale"] + assert scale.options == ["2x", "4x"] + assert scale.default == "4x" + + +def test_image_poc_api_declarations_and_extension_registration(): + workflows = {workflow for _, workflow, _, _ in IMAGE_POC_NODES} + registered = set(asyncio.run(nodes_comfy_cloud.ComfyCloudExtension().get_node_list())) + + assert workflows <= set(get_args(ComfyCloudWorkflow)) + assert {node for node, _, _, _ in IMAGE_POC_NODES} <= registered + + +def test_cloud_workflow_controls_have_connection_sockets(): + nodes = asyncio.run(nodes_comfy_cloud.ComfyCloudExtension().get_node_list()) + + for node in nodes: + for input_spec in node.define_schema().inputs: + if isinstance(input_spec, nodes_comfy_cloud.IO.WidgetInput): + assert input_spec.socketless is False, f"{node.__name__}.{input_spec.id}" + + +def test_cloud_workflow_schemas_have_descriptions(): + nodes = asyncio.run(nodes_comfy_cloud.ComfyCloudExtension().get_node_list()) + + for node in nodes: + assert node.define_schema().description.strip(), node.__name__ + + +def test_cloud_workflow_schemas_share_exact_estimated_rate_metadata(): + nodes = asyncio.run(nodes_comfy_cloud.ComfyCloudExtension().get_node_list()) + + assert nodes_comfy_cloud.COMFY_CLOUD_GPU_SECOND_USD == 0.001295 + assert nodes_comfy_cloud.COMFY_CLOUD_CREDITS_PER_USD == 211 + assert nodes_comfy_cloud.COMFY_CLOUD_GPU_SECOND_CREDITS == pytest.approx(0.273245) + assert nodes_comfy_cloud.COMFY_CLOUD_GPU_HOUR_USD == pytest.approx(4.662) + assert nodes_comfy_cloud.COMFY_CLOUD_GPU_HOUR_CREDITS == pytest.approx(983.682) + for node in nodes: + schema = node.define_schema() + badge = schema.price_badge.as_dict(schema.inputs) + assert badge["expr"] == ( + '{"type":"usd","usd":0.001295,"format":{"suffix":"/GPU-second","approximate":true}}' + ) + assert "Estimated compute rate" in schema.description + assert "Actual final cost depends on GPU runtime" in schema.description + + +def test_all_linkable_widget_constraints_are_validated(): + nodes = asyncio.run(nodes_comfy_cloud.ComfyCloudExtension().get_node_list()) + + for node in nodes: + for input_spec in node.define_schema().inputs: + if not isinstance(input_spec, nodes_comfy_cloud.IO.WidgetInput): + continue + io_type = input_spec.get_io_type() + if io_type == "COMBO": + invalid = "not-an-option" + elif io_type == "BOOLEAN": + invalid = "true" + elif io_type == "INT": + invalid = (input_spec.min - 1) if input_spec.min is not None else 1.5 + elif io_type == "FLOAT": + invalid = float("nan") + elif io_type == "STRING" and nodes_comfy_cloud._TEXT_LIMITS.get(input_spec.id, (0,))[0]: + invalid = " " + else: + continue + with pytest.raises((ValueError, Exception), match=input_spec.id): + nodes_comfy_cloud._validate_node_inputs(node, {input_spec.id: invalid}) + + +def test_linked_values_are_validated_before_upload(monkeypatch): + upload = AsyncMock() + monkeypatch.setattr(nodes_comfy_cloud, "upload_image_to_comfyapi", upload) + monkeypatch.setattr(nodes_comfy_cloud, "get_number_of_images", lambda image: 1) + + with pytest.raises(ValueError, match="guidance"): + asyncio.run( + nodes_comfy_cloud.ComfyCloudFlux2ReferenceEditNode.execute( + object(), "instruction", float("nan"), "quality", 0 + ) + ) + upload.assert_not_awaited() + + +def test_upload_inputs_have_decoded_resource_limits(): + oversized_image = torch.empty((1, 8193, 1, 3), device="meta") + oversized_audio = { + "waveform": torch.empty((1, 2, nodes_comfy_cloud._MAX_DECODED_AUDIO_BYTES // 8 + 1), device="meta"), + "sample_rate": 48000, + } + + with pytest.raises(ValueError, match="32-megapixel"): + nodes_comfy_cloud._validate_image_upload(oversized_image) + with pytest.raises(ValueError, match="256 MiB"): + nodes_comfy_cloud._validate_audio_upload(oversized_audio) + + +@pytest.mark.parametrize( + ("node", "input_names"), + [ + (nodes_comfy_cloud.ComfyCloudMiniMaxH3TextSoundNode, ["prompt", "aspect_ratio", "duration_seconds", "seed"]), + (nodes_comfy_cloud.ComfyCloudMiniMaxH3ImageSoundNode, ["image", "prompt", "aspect_ratio", "duration_seconds", "seed"]), + (nodes_comfy_cloud.ComfyCloudLTX23ImageAudioPerformanceNode, ["image", "audio", "prompt", "enhance_prompt", "duration_seconds", "seed"]), + (nodes_comfy_cloud.ComfyCloudLTX23FirstLastFrameNode, ["first_frame", "last_frame", "prompt", "duration_seconds", "seed"]), + (nodes_comfy_cloud.ComfyCloudWan22FirstLastFrameNode, ["first_frame", "last_frame", "prompt", "negative_prompt", "duration_seconds", "seed"]), + (nodes_comfy_cloud.ComfyCloudSCAIL2CharacterReplacementNode, ["reference_character", "driving_video", "scene_prompt", "driving_subject", "reference_subject", "seed"]), + ], +) +def test_video_node_schemas_expose_only_manifest_inputs(node, input_names): + schema = node.define_schema() + assert schema.is_api_node + assert [input.id for input in schema.inputs] == input_names + assert len(schema.outputs) == 1 + assert schema.outputs[0].get_io_type() == "VIDEO" + + +def test_ltx_performance_stages_image_and_audio(monkeypatch): + run = AsyncMock(return_value=("video-output",)) + image_upload = AsyncMock(return_value="https://example.com/image.png") + audio_upload = AsyncMock(return_value="https://example.com/audio.mp4") + monkeypatch.setattr(nodes_comfy_cloud, "_run_video_workflow", run) + monkeypatch.setattr(nodes_comfy_cloud, "upload_image_to_comfyapi", image_upload) + monkeypatch.setattr(nodes_comfy_cloud, "upload_audio_to_comfyapi", audio_upload) + monkeypatch.setattr(nodes_comfy_cloud, "get_number_of_images", lambda image: 1) + audio = {"waveform": torch.zeros(1, 1, 480000), "sample_rate": 48000} + + asyncio.run(nodes_comfy_cloud.ComfyCloudLTX23ImageAudioPerformanceNode.execute(object(), audio, "sing", True, 9, 7)) + + inputs = run.call_args.args[2] + assert inputs.image_url == "https://example.com/image.png" + assert inputs.audio_url == "https://example.com/audio.mp4" + assert inputs.duration_seconds == 9 + + +def test_scail_stages_reference_image_and_driving_video(monkeypatch): + run = AsyncMock(return_value=("video-output",)) + image_upload = AsyncMock(return_value="https://example.com/character.png") + video_upload = AsyncMock(return_value="https://example.com/driving.mp4") + video = Mock() + video.get_frame_count.return_value = 100 + monkeypatch.setattr(nodes_comfy_cloud, "_run_video_workflow", run) + monkeypatch.setattr(nodes_comfy_cloud, "upload_image_to_comfyapi", image_upload) + monkeypatch.setattr(nodes_comfy_cloud, "upload_video_to_comfyapi", video_upload) + monkeypatch.setattr(nodes_comfy_cloud, "get_number_of_images", lambda image: 1) + + asyncio.run(nodes_comfy_cloud.ComfyCloudSCAIL2CharacterReplacementNode.execute(object(), video, "park", "woman", "human", 1)) + + inputs = run.call_args.args[2] + assert inputs.reference_character_url == "https://example.com/character.png" + assert inputs.driving_video_url == "https://example.com/driving.mp4" + video.get_frame_count.assert_called_once() + + +def test_scail_defaults_and_frame_count_fail_closed_before_upload(monkeypatch): + schema = {input.id: input for input in nodes_comfy_cloud.ComfyCloudSCAIL2CharacterReplacementNode.define_schema().inputs} + image_upload = AsyncMock() + video_upload = AsyncMock() + video = Mock() + video.get_frame_count.side_effect = RuntimeError("decode failed") + monkeypatch.setattr(nodes_comfy_cloud, "upload_image_to_comfyapi", image_upload) + monkeypatch.setattr(nodes_comfy_cloud, "upload_video_to_comfyapi", video_upload) + monkeypatch.setattr(nodes_comfy_cloud, "get_number_of_images", lambda image: 1) + + assert schema["driving_subject"].default == "human" + with pytest.raises(ValueError, match="Unable to determine video frame count"): + asyncio.run(nodes_comfy_cloud.ComfyCloudSCAIL2CharacterReplacementNode.execute(object(), video, "park", "human", "human", 1)) + image_upload.assert_not_awaited() + video_upload.assert_not_awaited() + + +def test_scail_rejects_missing_frame_count_metadata_before_upload(monkeypatch): + upload = AsyncMock() + video = Mock() + video.get_frame_count.return_value = None + monkeypatch.setattr(nodes_comfy_cloud, "upload_image_to_comfyapi", upload) + monkeypatch.setattr(nodes_comfy_cloud, "upload_video_to_comfyapi", upload) + monkeypatch.setattr(nodes_comfy_cloud, "get_number_of_images", lambda image: 1) + + with pytest.raises(ValueError, match="Unable to determine video frame count"): + asyncio.run( + nodes_comfy_cloud.ComfyCloudSCAIL2CharacterReplacementNode.execute( + object(), video, "park", "human", "human", 1 + ) + ) + upload.assert_not_awaited() + + +@pytest.mark.parametrize("frame_count", [80, 158]) +def test_scail_rejects_out_of_range_frames_before_upload(monkeypatch, frame_count): + upload = AsyncMock() + video = Mock() + video.get_frame_count.return_value = frame_count + monkeypatch.setattr(nodes_comfy_cloud, "upload_image_to_comfyapi", upload) + monkeypatch.setattr(nodes_comfy_cloud, "upload_video_to_comfyapi", upload) + monkeypatch.setattr(nodes_comfy_cloud, "get_number_of_images", lambda image: 1) + + with pytest.raises(ValueError, match="frame count"): + asyncio.run(nodes_comfy_cloud.ComfyCloudSCAIL2CharacterReplacementNode.execute(object(), video, "park", "human", "human", 1)) + upload.assert_not_awaited() + + +def test_in_memory_download_resets_retry_and_enforces_stream_limit(monkeypatch): + class Content: + def __init__(self, chunks): + self.chunks = iter(chunks) + self.finished = False + + async def read(self, size): + chunk = next(self.chunks) + if isinstance(chunk, Exception): + raise chunk + if not chunk: + self.finished = True + return chunk + + def at_eof(self): + return self.finished + + class Response: + status = 200 + headers = {} + + def __init__(self, chunks, content_length=None): + self.content = Content(chunks) + self.content_length = content_length + + async def __aenter__(self): + return self + + async def __aexit__(self, *args): + return False + + responses = [Response([b"partial", aiohttp.ClientPayloadError("retry")]), Response([b"final", b""])] + + class Session: + def __init__(self, timeout): + pass + + async def get(self, url, headers, allow_redirects=True): + return responses.pop(0) + + async def close(self): + pass + + monkeypatch.setattr(download_helpers.aiohttp, "ClientSession", Session) + monkeypatch.setattr(download_helpers, "sleep_with_interrupt", AsyncMock()) + destination = BytesIO() + + asyncio.run(download_helpers.download_url_to_bytesio("https://example.com/result", destination)) + assert destination.read() == b"final" + + monkeypatch.setattr(download_helpers, "_MAX_IN_MEMORY_DOWNLOAD_BYTES", 4) + responses.append(Response([b"12345", b""])) + with pytest.raises(ValueError, match="in-memory limit"): + asyncio.run(download_helpers.download_url_to_bytesio("https://example.com/result", BytesIO())) + + responses.append(Response([], content_length=5)) + with pytest.raises(ValueError, match="in-memory limit"): + asyncio.run(download_helpers.download_url_to_bytesio("https://example.com/result", BytesIO())) + + +def test_file_object_download_resets_retry_and_enforces_stream_limit(monkeypatch, tmp_path): + destination = (tmp_path / "result.bin").open("w+b") + destination.write(b"stale") + + class Content: + def __init__(self, chunks): + self.chunks = iter(chunks) + self.finished = False + + async def read(self, size): + chunk = next(self.chunks) + if isinstance(chunk, Exception): + raise chunk + if not chunk: + self.finished = True + return chunk + + def at_eof(self): + return self.finished + + class Response: + status = 200 + headers = {} + content_length = None + + def __init__(self, chunks): + self.content = Content(chunks) + + async def __aenter__(self): + return self + + async def __aexit__(self, *args): + return False + + responses = [Response([b"partial", aiohttp.ClientPayloadError("retry")]), Response([b"done", b""])] + + class Session: + def __init__(self, timeout): + pass + + async def get(self, url, headers, allow_redirects=True): + return responses.pop(0) + + async def close(self): + pass + + monkeypatch.setattr(download_helpers.aiohttp, "ClientSession", Session) + monkeypatch.setattr(download_helpers, "sleep_with_interrupt", AsyncMock()) + monkeypatch.setattr(download_helpers, "_MAX_IN_MEMORY_DOWNLOAD_BYTES", 10) + + asyncio.run(download_helpers.download_url_to_bytesio("https://example.com/result", destination)) + assert destination.read() == b"done" + + monkeypatch.setattr(download_helpers, "_MAX_IN_MEMORY_DOWNLOAD_BYTES", 4) + responses.append(Response([b"12345", b""])) + with pytest.raises(ValueError, match="in-memory limit"): + asyncio.run(download_helpers.download_url_to_bytesio("https://example.com/result", destination)) + destination.close() + + +def test_download_cloud_audio_url_to_audio_input(monkeypatch): + node = nodes_comfy_cloud.ComfyCloudTextToImageNode + downloaded = b"encoded audio" + expected = {"waveform": torch.ones(1, 2, 3), "sample_rate": 48000} + download_call = Mock() + + async def download(url, dest, **kwargs): + download_call(url=url, dest=dest, **kwargs) + dest.write(downloaded) + dest.seek(0) + + audio_decode = Mock(return_value=expected) + monkeypatch.setattr(download_helpers, "download_url_to_bytesio", download) + monkeypatch.setattr(download_helpers, "audio_bytes_to_audio_input", audio_decode) + + output = asyncio.run( + download_helpers.download_url_to_audio_input( + "/proxy/comfy-cloud/results/task-1/audio.flac", + timeout=30, + max_retries=2, + cls=node, + ) + ) + + assert output is expected + download_call.assert_called_once() + assert download_call.call_args.kwargs["url"] == "/proxy/comfy-cloud/results/task-1/audio.flac" + assert isinstance(download_call.call_args.kwargs["dest"], BytesIO) + assert download_call.call_args.kwargs["timeout"] == 30 + assert download_call.call_args.kwargs["max_retries"] == 2 + assert download_call.call_args.kwargs["cls"] is node + assert download_call.call_args.kwargs["allow_redirects"] is True + audio_decode.assert_called_once_with(downloaded) + + +AUDIO_POC_NODES = [ + (nodes_comfy_cloud.ComfyCloudACEStep15XLTurboNode, "audio.ace-step-1-5-xl-turbo.v1", ["style_prompt", "lyrics", "duration_seconds", "seed", "bpm", "time_signature", "language", "key"]), + (nodes_comfy_cloud.ComfyCloudStableAudio3MediumNode, "audio.stable-audio-3-medium.v1", ["prompt", "duration_seconds", "seed", "expand_prompt", "category"]), + (nodes_comfy_cloud.ComfyCloudChatterboxMultilingualVoiceCloneNode, "audio.chatterbox-multilingual-voice-clone.v1", ["text", "voice_reference", "language", "exaggeration", "cfg_weight", "temperature", "seed"]), + (nodes_comfy_cloud.ComfyCloudChatterboxDialogueNode, "audio.chatterbox-dialogue.v1", ["script", "speaker_a_reference", "speaker_b_reference", "exaggeration", "cfg_weight", "temperature", "seed"]), + (nodes_comfy_cloud.ComfyCloudChatterboxVoiceConversionNode, "audio.chatterbox-voice-conversion.v1", ["source_audio", "target_voice_reference", "seed"]), + (nodes_comfy_cloud.ComfyCloudMelBandRoFormerStemSeparationNode, "audio.melbandroformer-stem-separation.v1", ["audio"]), +] + + +@pytest.mark.parametrize(("node", "workflow", "input_names"), AUDIO_POC_NODES) +def test_audio_poc_schemas_and_registration(node, workflow, input_names): + schema = node.define_schema() + registered = asyncio.run(nodes_comfy_cloud.ComfyCloudExtension().get_node_list()) + + assert schema.is_api_node + assert schema.category == "partner/audio/Comfy Cloud" + assert [input.id for input in schema.inputs] == input_names + assert all(output.get_io_type() == "AUDIO" for output in schema.outputs) + assert workflow in get_args(ComfyCloudWorkflow) + assert node in registered + + +def test_audio_poc_schema_defaults_ranges_and_enums(): + schemas = {workflow: {input.id: input for input in node.define_schema().inputs} for node, workflow, _ in AUDIO_POC_NODES} + ace = schemas["audio.ace-step-1-5-xl-turbo.v1"] + assert (ace["duration_seconds"].default, ace["duration_seconds"].min, ace["duration_seconds"].max, ace["duration_seconds"].step) == (120, 10, 300, 0.1) + assert (ace["bpm"].default, ace["bpm"].min, ace["bpm"].max) == (120, 10, 300) + assert ace["time_signature"].options == ["2", "3", "4", "6"] + assert ace["language"].default == "en" + assert ace["key"].default == "E minor" + + stable = schemas["audio.stable-audio-3-medium.v1"] + assert stable["category"].options == ["Music", "Instrument", "SFX", "One-shot"] + assert stable["expand_prompt"].default is True + for workflow in [ + "audio.chatterbox-multilingual-voice-clone.v1", + "audio.chatterbox-dialogue.v1", + "audio.chatterbox-voice-conversion.v1", + ]: + assert schemas[workflow]["seed"].max == 0xFFFFFFFF + + mel_schema = nodes_comfy_cloud.ComfyCloudMelBandRoFormerStemSeparationNode.define_schema() + assert [output.id for output in mel_schema.outputs] == ["vocals", "instruments"] + + +def test_audio_poc_request_mapping_and_named_result_decoding(monkeypatch): + sync = AsyncMock(return_value=ComfyCloudGenerateResponse(task_id="task-audio", status="queued", polling_url="/tasks/task-audio", cancel_url="/tasks/task-audio/cancel")) + poll = AsyncMock(return_value=ComfyCloudStatusResponse(task_id="task-audio", status="completed", output_urls={"vocals": "/proxy/comfy-cloud/results/vocals.mp3", "instruments": "/proxy/comfy-cloud/results/instruments.mp3"})) + upload = AsyncMock(return_value="/uploads/song.m4a") + download = AsyncMock(side_effect=["vocals-audio", "instruments-audio"]) + monkeypatch.setattr(nodes_comfy_cloud, "sync_op", sync) + monkeypatch.setattr(nodes_comfy_cloud, "poll_op", poll) + monkeypatch.setattr(nodes_comfy_cloud, "upload_audio_to_comfyapi", upload) + monkeypatch.setattr(nodes_comfy_cloud, "download_url_to_audio_input", download) + audio = {"waveform": torch.zeros(1, 2, 48000), "sample_rate": 48000} + + output = asyncio.run(nodes_comfy_cloud.ComfyCloudMelBandRoFormerStemSeparationNode.execute(audio)) + + request = sync.call_args.kwargs["data"] + assert request.workflow == "audio.melbandroformer-stem-separation.v1" + assert request.inputs.model_dump(exclude_none=True) == {"assets": {"audio": {"type": "AUDIO", "url": "/uploads/song.m4a"}}} + assert [call.args[0] for call in download.await_args_list] == [ + "/proxy/comfy-cloud/results/vocals.mp3", + "/proxy/comfy-cloud/results/instruments.mp3", + ] + assert all(call.kwargs["timeout"] == 30 * 60 for call in download.await_args_list) + assert all(call.kwargs["allow_redirects"] is False for call in download.await_args_list) + assert poll.call_args.kwargs["cancel_endpoint"].path == "/proxy/comfy-cloud/workflow/tasks/task-audio/cancel" + assert tuple(output) == ("vocals-audio", "instruments-audio") + + +def test_chatterbox_audio_inputs_use_named_staged_assets(monkeypatch): + run = AsyncMock(return_value=("audio-output",)) + upload = AsyncMock(side_effect=["/uploads/source.m4a", "/uploads/target.m4a"]) + monkeypatch.setattr(nodes_comfy_cloud, "_run_audio_workflow", run) + monkeypatch.setattr(nodes_comfy_cloud, "upload_audio_to_comfyapi", upload) + source = {"waveform": torch.zeros(1, 1, 48000), "sample_rate": 48000} + target = {"waveform": torch.zeros(1, 1, 96000), "sample_rate": 48000} + + asyncio.run(nodes_comfy_cloud.ComfyCloudChatterboxVoiceConversionNode.execute(source, target, 7)) + + inputs = run.call_args.args[2] + assert inputs.model_dump(exclude_none=True) == { + "assets": { + "source_audio": {"type": "AUDIO", "url": "/uploads/source.m4a"}, + "target_voice_reference": {"type": "AUDIO", "url": "/uploads/target.m4a"}, + }, + "seed": 7, + } + assert "audio_url" not in inputs.model_dump(exclude_none=True) + + +def test_chatterbox_dialogue_rejects_invalid_speaker_labels(monkeypatch): + upload = AsyncMock() + monkeypatch.setattr(nodes_comfy_cloud, "upload_audio_to_comfyapi", upload) + audio = {"waveform": torch.zeros(1, 1, 48000), "sample_rate": 48000} + + with pytest.raises(ValueError, match="only speakers A and B"): + asyncio.run(nodes_comfy_cloud.ComfyCloudChatterboxDialogueNode.execute("NARRATOR: Hello", audio, audio, 0.5, 0.5, 0.8, 0)) + upload.assert_not_awaited() + + +def test_chatterbox_dialogue_normalizes_labels_and_continuations(monkeypatch): + run = AsyncMock(return_value=("audio-output",)) + monkeypatch.setattr(nodes_comfy_cloud, "_run_audio_workflow", run) + monkeypatch.setattr(nodes_comfy_cloud, "upload_audio_to_comfyapi", AsyncMock(side_effect=["/a", "/b"])) + audio = {"waveform": torch.zeros(1, 1, 48000), "sample_rate": 48000} + + asyncio.run(nodes_comfy_cloud.ComfyCloudChatterboxDialogueNode.execute("a: Hello\ncontinued\n\nSpeaker B: Hi", audio, audio, 0.5, 0.5, 0.8, 0)) + + assert run.call_args.args[2].script == "SPEAKER A: Hello continued\nSPEAKER B: Hi" + + +def test_chatterbox_dialogue_accepts_colons_and_label_only_lines(): + assert nodes_comfy_cloud._normalize_dialogue("SPEAKER A :\nMeet at 10:30\nhttps://example.com\nB: Done") == ( + "SPEAKER A: Meet at 10:30 https://example.com\nSPEAKER B: Done" + ) + + +@pytest.mark.parametrize("script", ["SPEAKER A:", "just a continuation", " "]) +def test_chatterbox_dialogue_rejects_blank_or_unattributed_text(monkeypatch, script): + upload = AsyncMock() + monkeypatch.setattr(nodes_comfy_cloud, "upload_audio_to_comfyapi", upload) + audio = {"waveform": torch.zeros(1, 1, 48000), "sample_rate": 48000} + + with pytest.raises(Exception): + asyncio.run(nodes_comfy_cloud.ComfyCloudChatterboxDialogueNode.execute(script, audio, audio, 0.5, 0.5, 0.8, 0)) + upload.assert_not_awaited() + + +def test_audio_duration_tolerates_one_sample_and_rejects_invalid_sample_rate(): + one_sample_over = {"waveform": torch.zeros(1, 1, 48001), "sample_rate": 48000} + nodes_comfy_cloud._validate_audio_duration("Audio", one_sample_over, 0.5, 1) + + with pytest.raises(ValueError, match="sample rate"): + nodes_comfy_cloud._validate_audio_duration("Audio", {"waveform": torch.zeros(1, 1, 1), "sample_rate": 0}, 0.5, 1) + + with pytest.raises(ValueError, match="between"): + nodes_comfy_cloud._validate_audio_duration("Audio", {"waveform": torch.zeros(1, 1, 31), "sample_rate": 1}, 1, 30) + + +@pytest.mark.parametrize(("file_format", "expected_format"), [(".GLB", "glb"), ("SPZ", "spz")]) +def test_download_cloud_3d_url_to_file_3d(monkeypatch, file_format, expected_format): + node = nodes_comfy_cloud.ComfyCloudTextToImageNode + downloaded = b"3d result" + calls = [] + + async def download(url, dest, **kwargs): + calls.append((url, dest, kwargs)) + dest.write(downloaded) + dest.seek(0) + + monkeypatch.setattr(download_helpers, "download_url_to_bytesio", download) + + output = asyncio.run( + download_helpers.download_url_to_file_3d( + f"/proxy/comfy-cloud/results/task-1/model.{expected_format}", + file_format, + timeout=45, + max_retries=3, + cls=node, + ) + ) + + assert output.format == expected_format + assert output.get_bytes() == downloaded + assert calls[0][0] == f"/proxy/comfy-cloud/results/task-1/model.{expected_format}" + assert isinstance(calls[0][1], BytesIO) + assert calls[0][2] == {"timeout": 45, "max_retries": 3, "cls": node, "allow_redirects": True} + + +THREE_D_POC_NODES = [ + (nodes_comfy_cloud.ComfyCloudTripoSplatImageToGaussianSplatNode, "3d.triposplat-image-to-gaussian-splat.v1", ["image", "remove_background", "seed", "gaussian_count"], {"image": object(), "remove_background": False, "seed": 7, "gaussian_count": 32768}, "FILE_3D_SPZ", "spz"), + (nodes_comfy_cloud.ComfyCloudHunyuan3D21ImageTo3DNode, "3d.hunyuan3d-2-1-image-to-3d.v1", ["image", "seed"], {"image": object(), "seed": 8}, "FILE_3D_GLB", "glb"), + (nodes_comfy_cloud.ComfyCloudHunyuan3DMultiViewTo3DNode, "3d.hunyuan3d-multiview-to-3d.v1", ["front_image", "back_image", "seed"], {"front_image": object(), "back_image": object(), "seed": 9}, "FILE_3D_GLB", "glb"), + (nodes_comfy_cloud.ComfyCloudMoGe2PhotoToTexturedMeshNode, "3d.moge-2-photo-to-textured-mesh.v1", ["image", "fov_degrees", "detail", "mesh_decimation", "gap_threshold", "texture"], {"image": object(), "fov_degrees": 45.5, "detail": 8, "mesh_decimation": 2, "gap_threshold": 0.05, "texture": False}, "FILE_3D_GLB", "glb"), + (nodes_comfy_cloud.ComfyCloudMoGe2PanoramaTo3DSceneNode, "3d.moge-2-panorama-to-3d-scene.v1", ["panorama", "detail", "split_resolution", "merge_resolution", "mesh_decimation", "gap_threshold", "texture"], {"panorama": object(), "detail": 6, "split_resolution": 768, "merge_resolution": 2048, "mesh_decimation": 3, "gap_threshold": 0.06, "texture": False}, "FILE_3D_GLB", "glb"), +] + + +@pytest.mark.parametrize(("node", "workflow", "input_names", "arguments", "output_type", "file_format"), THREE_D_POC_NODES) +def test_3d_poc_node_schema_request_mapping_and_registration(monkeypatch, node, workflow, input_names, arguments, output_type, file_format): + run = AsyncMock(return_value=("3d-output",)) + upload = AsyncMock(side_effect=["/uploads/front.png", "/uploads/back.png"]) + monkeypatch.setattr(nodes_comfy_cloud, "_run_3d_workflow", run) + monkeypatch.setattr(nodes_comfy_cloud, "upload_image_to_comfyapi", upload) + monkeypatch.setattr(nodes_comfy_cloud, "get_number_of_images", lambda image: 1) + + schema = node.define_schema() + assert schema.is_api_node + assert schema.category == "partner/3d/Comfy Cloud" + assert [input.id for input in schema.inputs] == input_names + assert schema.outputs[0].get_io_type() == output_type + assert workflow in get_args(ComfyCloudWorkflow) + assert node in asyncio.run(nodes_comfy_cloud.ComfyCloudExtension().get_node_list()) + + output = asyncio.run(node.execute(**arguments)) + assert output[0] == "3d-output" + assert run.call_args.args[1] == workflow + assert run.call_args.args[3] == file_format + request_inputs = run.call_args.args[2].model_dump(exclude_none=True) + image_names = [name for name in ("image", "front_image", "back_image", "panorama") if name in arguments] + expected_inputs = {name: value for name, value in arguments.items() if name not in image_names} + expected_inputs["assets"] = { + name: {"type": "IMAGE", "url": f"/uploads/{'front' if index == 0 else 'back'}.png"} + for index, name in enumerate(image_names) + } + assert request_inputs == expected_inputs + assert '"id"' not in run.call_args.args[2].model_dump_json() + + +def test_3d_poc_schema_defaults_and_ranges(): + schemas = {workflow: {input.id: input for input in node.define_schema().inputs} for node, workflow, _, _, _, _ in THREE_D_POC_NODES} + tripo = schemas["3d.triposplat-image-to-gaussian-splat.v1"] + assert tripo["remove_background"].default is True + assert (tripo["seed"].default, tripo["seed"].min, tripo["seed"].max) == (46, 0, 0xFFFFFFFFFFFFFFFF) + assert (tripo["gaussian_count"].default, tripo["gaussian_count"].min, tripo["gaussian_count"].max) == (262144, 32768, 262144) + assert "application/octet-stream" in nodes_comfy_cloud.ComfyCloudTripoSplatImageToGaussianSplatNode.define_schema().outputs[0].tooltip + assert schemas["3d.hunyuan3d-2-1-image-to-3d.v1"]["seed"].default == 952805179515179 + assert schemas["3d.hunyuan3d-multiview-to-3d.v1"]["seed"].default == 502126049100058 + photo = schemas["3d.moge-2-photo-to-textured-mesh.v1"] + assert (photo["fov_degrees"].default, photo["fov_degrees"].min, photo["fov_degrees"].max, photo["fov_degrees"].step) == (0, 0, 170, 0.1) + assert (photo["detail"].default, photo["detail"].min, photo["detail"].max) == (9, 0, 9) + panorama = schemas["3d.moge-2-panorama-to-3d-scene.v1"] + assert (panorama["split_resolution"].default, panorama["split_resolution"].min, panorama["split_resolution"].max) == (512, 256, 1024) + assert (panorama["merge_resolution"].default, panorama["merge_resolution"].min, panorama["merge_resolution"].max) == (1024, 256, 8192) + + +def test_hunyuan_multiview_validates_both_images_before_upload(monkeypatch): + upload = AsyncMock() + monkeypatch.setattr(nodes_comfy_cloud, "upload_image_to_comfyapi", upload) + monkeypatch.setattr(nodes_comfy_cloud, "get_number_of_images", lambda image: image) + + with pytest.raises(ValueError, match="Exactly one front image and one back image"): + asyncio.run(nodes_comfy_cloud.ComfyCloudHunyuan3DMultiViewTo3DNode.execute(1, 2, 9)) + upload.assert_not_awaited() + + +def test_extension_preserves_all_23_poc_node_registrations(): + legacy_nodes = { + nodes_comfy_cloud.ComfyCloudTextToImageNode, + nodes_comfy_cloud.ComfyCloudTextToVideoNode, + nodes_comfy_cloud.ComfyCloudImageToVideoNode, + nodes_comfy_cloud.ComfyCloudImageEditNode, + } + registered = set(asyncio.run(nodes_comfy_cloud.ComfyCloudExtension().get_node_list())) + + assert len(registered - legacy_nodes) == 23 + + +def test_3d_workflow_submission_polling_cancel_and_download(monkeypatch): + sync = AsyncMock(return_value=ComfyCloudGenerateResponse(task_id="task-3d", status="queued", polling_url="/tasks/task-3d", cancel_url="/tasks/task-3d/cancel")) + poll = AsyncMock(return_value=ComfyCloudStatusResponse(task_id="task-3d", status="completed", output_url="/proxy/comfy-cloud/results/model.spz")) + download = AsyncMock(return_value="spz-output") + monkeypatch.setattr(nodes_comfy_cloud, "sync_op", sync) + monkeypatch.setattr(nodes_comfy_cloud, "poll_op", poll) + monkeypatch.setattr(nodes_comfy_cloud, "download_url_to_file_3d", download) + + output = asyncio.run(nodes_comfy_cloud._run_3d_workflow(nodes_comfy_cloud.ComfyCloudTripoSplatImageToGaussianSplatNode, "3d.triposplat-image-to-gaussian-splat.v1", ComfyCloudWorkflowInputs(seed=46), "spz")) + + request = sync.call_args.kwargs["data"] + assert request.workflow == "3d.triposplat-image-to-gaussian-splat.v1" + assert request.inputs.model_dump(exclude_none=True) == {"seed": 46} + assert poll.call_args.args[1].path == "/proxy/comfy-cloud/workflow/tasks/task-3d" + assert poll.call_args.kwargs["cancel_endpoint"].path == "/proxy/comfy-cloud/workflow/tasks/task-3d/cancel" + assert poll.call_args.kwargs["cancel_endpoint"].method == "POST" + download.assert_awaited_once_with( + "/proxy/comfy-cloud/results/model.spz", + "spz", + timeout=30 * 60, + cls=nodes_comfy_cloud.ComfyCloudTripoSplatImageToGaussianSplatNode, + allow_redirects=False, + ) + assert output[0] == "spz-output"