Merge 7cecb6dbf8 into 8f40b43e02

* feat(api-nodes): add Meshy 3D nodes

* rebased, added JSONata price badges

comfy/cli_args.py

@@ -152,6 +152,8 @@ parser.add_argument("--force-non-blocking", action="store_true", help="Force Com
 parser.add_argument("--default-hashing-function", type=str, choices=['md5', 'sha1', 'sha256', 'sha512'], default='sha256', help="Allows you to choose the hash function to use for duplicate filename / contents comparison. Default is sha256.")
 
 parser.add_argument("--disable-smart-memory", action="store_true", help="Force ComfyUI to agressively offload to regular ram instead of keeping models in vram when it can.")
+parser.add_argument("--use-subprocess-workers", action="store_true", help="Execute each prompt in an isolated subprocess with complete GPU/ROCm context reset. Ensures clean state between jobs but adds startup overhead.")
+parser.add_argument("--subprocess-timeout", type=int, default=600, help="Timeout in seconds for subprocess execution (default: 600, only used with --use-subprocess-workers).")
 parser.add_argument("--deterministic", action="store_true", help="Make pytorch use slower deterministic algorithms when it can. Note that this might not make images deterministic in all cases.")
 
 class PerformanceFeature(enum.Enum):

comfy/execution_core.py

@@ -0,0 +1,145 @@
+"""Core execution logic shared between normal and subprocess execution modes."""
+
+import logging
+import time
+
+_active_worker = None
+
+
+def create_worker(server_instance):
+    """Create worker backend. Returns NativeWorker or SubprocessWorker."""
+    global _active_worker
+    from comfy.cli_args import args
+
+    server = WorkerServer(server_instance)
+
+    if args.use_subprocess_workers:
+        from comfy.worker_process import SubprocessWorker
+        worker = SubprocessWorker(server, timeout=args.subprocess_timeout)
+    else:
+        from comfy.worker_native import NativeWorker
+        worker = NativeWorker(server)
+
+    _active_worker = worker
+    return worker
+
+
+async def init_execution_environment():
+    """Load nodes and custom nodes. Returns number of node types loaded."""
+    import nodes
+    from comfy.cli_args import args
+
+    await nodes.init_extra_nodes(
+        init_custom_nodes=(not args.disable_all_custom_nodes) or len(args.whitelist_custom_nodes) > 0,
+        init_api_nodes=not args.disable_api_nodes
+    )
+    return len(nodes.NODE_CLASS_MAPPINGS)
+
+
+def setup_progress_hook(server_instance, interrupt_checker):
+    """Set up global progress hook. interrupt_checker must raise on interrupt."""
+    import comfy.utils
+    from comfy_execution.progress import get_progress_state
+    from comfy_execution.utils import get_executing_context
+
+    def hook(value, total, preview_image, prompt_id=None, node_id=None):
+        ctx = get_executing_context()
+        if ctx:
+            prompt_id = prompt_id or ctx.prompt_id
+            node_id = node_id or ctx.node_id
+
+        interrupt_checker()
+
+        prompt_id = prompt_id or server_instance.last_prompt_id
+        node_id = node_id or server_instance.last_node_id
+
+        get_progress_state().update_progress(node_id, value, total, preview_image)
+        server_instance.send_sync("progress", {"value": value, "max": total, "prompt_id": prompt_id, "node": node_id}, server_instance.client_id)
+
+    comfy.utils.set_progress_bar_global_hook(hook)
+
+
+class WorkerServer:
+    """Protocol boundary: client_id, last_node_id, last_prompt_id, sockets_metadata, send_sync(), queue_updated()"""
+
+    _WRITABLE = {'client_id', 'last_node_id', 'last_prompt_id'}
+
+    def __init__(self, server):
+        object.__setattr__(self, '_server', server)
+
+    def __setattr__(self, name, value):
+        if name in self._WRITABLE:
+            setattr(self._server, name, value)
+        else:
+            raise AttributeError(f"WorkerServer does not accept attribute '{name}'")
+
+    @property
+    def client_id(self):
+        return self._server.client_id
+
+    @property
+    def last_node_id(self):
+        return self._server.last_node_id
+
+    @property
+    def last_prompt_id(self):
+        return self._server.last_prompt_id
+
+    @property
+    def sockets_metadata(self):
+        return self._server.sockets_metadata
+
+    def send_sync(self, event, data, sid=None):
+        self._server.send_sync(event, data, sid or self.client_id)
+
+    def queue_updated(self):
+        self._server.queue_updated()
+
+def interrupt_processing(value=True):
+    _active_worker.interrupt(value)
+
+
+def _strip_sensitive(prompt):
+    return prompt[:5] + prompt[6:]
+
+
+def prompt_worker(q, worker):
+    """Main prompt execution loop."""
+    import execution
+
+    server = worker.server_instance
+
+    while True:
+        queue_item = q.get(timeout=worker.get_gc_timeout())
+        if queue_item is not None:
+            item, item_id = queue_item
+            start_time = time.perf_counter()
+            prompt_id = item[1]
+            server.last_prompt_id = prompt_id
+
+            extra_data = {**item[3], **item[5]}
+
+            result = worker.execute_prompt(item[2], prompt_id, extra_data, item[4], server=server)
+            worker.mark_needs_gc()
+
+            q.task_done(
+                item_id,
+                result['history_result'],
+                status=execution.PromptQueue.ExecutionStatus(
+                    status_str='success' if result['success'] else 'error',
+                    completed=result['success'],
+                    messages=result['status_messages']
+                ),
+                process_item=_strip_sensitive
+            )
+
+            if server.client_id is not None:
+                server.send_sync("executing", {"node": None, "prompt_id": prompt_id}, server.client_id)
+
+            elapsed = time.perf_counter() - start_time
+            if elapsed > 600:
+                logging.info(f"Prompt executed in {time.strftime('%H:%M:%S', time.gmtime(elapsed))}")
+            else:
+                logging.info(f"Prompt executed in {elapsed:.2f} seconds")
+
+        worker.handle_flags(q.get_flags())

comfy/sd.py

@@ -1014,6 +1014,7 @@ class CLIPType(Enum):
     KANDINSKY5 = 22
     KANDINSKY5_IMAGE = 23
     NEWBIE = 24
+    FLUX2 = 25
 
 
 def load_clip(ckpt_paths, embedding_directory=None, clip_type=CLIPType.STABLE_DIFFUSION, model_options={}):
@@ -1046,6 +1047,7 @@ class TEModel(Enum):
     QWEN3_2B = 17
     GEMMA_3_12B = 18
     JINA_CLIP_2 = 19
+    QWEN3_8B = 20
 
 
 def detect_te_model(sd):
@@ -1089,6 +1091,8 @@ def detect_te_model(sd):
                 return TEModel.QWEN3_4B
             elif weight.shape[0] == 2048:
                 return TEModel.QWEN3_2B
+            elif weight.shape[0] == 4096:
+                return TEModel.QWEN3_8B
         if weight.shape[0] == 5120:
             if "model.layers.39.post_attention_layernorm.weight" in sd:
                 return TEModel.MISTRAL3_24B
@@ -1214,11 +1218,18 @@ def load_text_encoder_state_dicts(state_dicts=[], embedding_directory=None, clip
             clip_target.tokenizer = comfy.text_encoders.flux.Flux2Tokenizer
             tokenizer_data["tekken_model"] = clip_data[0].get("tekken_model", None)
         elif te_model == TEModel.QWEN3_4B:
-            clip_target.clip = comfy.text_encoders.z_image.te(**llama_detect(clip_data))
-            clip_target.tokenizer = comfy.text_encoders.z_image.ZImageTokenizer
+            if clip_type == CLIPType.FLUX or clip_type == CLIPType.FLUX2:
+                clip_target.clip = comfy.text_encoders.flux.klein_te(**llama_detect(clip_data), model_type="qwen3_4b")
+                clip_target.tokenizer = comfy.text_encoders.flux.KleinTokenizer
+            else:
+                clip_target.clip = comfy.text_encoders.z_image.te(**llama_detect(clip_data))
+                clip_target.tokenizer = comfy.text_encoders.z_image.ZImageTokenizer
         elif te_model == TEModel.QWEN3_2B:
             clip_target.clip = comfy.text_encoders.ovis.te(**llama_detect(clip_data))
             clip_target.tokenizer = comfy.text_encoders.ovis.OvisTokenizer
+        elif te_model == TEModel.QWEN3_8B:
+            clip_target.clip = comfy.text_encoders.flux.klein_te(**llama_detect(clip_data), model_type="qwen3_8b")
+            clip_target.tokenizer = comfy.text_encoders.flux.KleinTokenizer8B
         elif te_model == TEModel.JINA_CLIP_2:
             clip_target.clip = comfy.text_encoders.jina_clip_2.JinaClip2TextModelWrapper
             clip_target.tokenizer = comfy.text_encoders.jina_clip_2.JinaClip2TokenizerWrapper

comfy/text_encoders/flux.py

@@ -3,7 +3,7 @@ import comfy.text_encoders.t5
 import comfy.text_encoders.sd3_clip
 import comfy.text_encoders.llama
 import comfy.model_management
-from transformers import T5TokenizerFast, LlamaTokenizerFast
+from transformers import T5TokenizerFast, LlamaTokenizerFast, Qwen2Tokenizer
 import torch
 import os
 import json
@@ -172,3 +172,60 @@ def flux2_te(dtype_llama=None, llama_quantization_metadata=None, pruned=False):
                 model_options["num_layers"] = 30
             super().__init__(device=device, dtype=dtype, model_options=model_options)
     return Flux2TEModel_
+
+class Qwen3Tokenizer(sd1_clip.SDTokenizer):
+    def __init__(self, embedding_directory=None, tokenizer_data={}):
+        tokenizer_path = os.path.join(os.path.dirname(os.path.realpath(__file__)), "qwen25_tokenizer")
+        super().__init__(tokenizer_path, pad_with_end=False, embedding_size=2560, embedding_key='qwen3_4b', tokenizer_class=Qwen2Tokenizer, has_start_token=False, has_end_token=False, pad_to_max_length=False, max_length=99999999, min_length=512, pad_token=151643, tokenizer_data=tokenizer_data)
+
+class Qwen3Tokenizer8B(sd1_clip.SDTokenizer):
+    def __init__(self, embedding_directory=None, tokenizer_data={}):
+        tokenizer_path = os.path.join(os.path.dirname(os.path.realpath(__file__)), "qwen25_tokenizer")
+        super().__init__(tokenizer_path, pad_with_end=False, embedding_size=4096, embedding_key='qwen3_8b', tokenizer_class=Qwen2Tokenizer, has_start_token=False, has_end_token=False, pad_to_max_length=False, max_length=99999999, min_length=512, pad_token=151643, tokenizer_data=tokenizer_data)
+
+class KleinTokenizer(sd1_clip.SD1Tokenizer):
+    def __init__(self, embedding_directory=None, tokenizer_data={}, name="qwen3_4b"):
+        if name == "qwen3_4b":
+            tokenizer = Qwen3Tokenizer
+        elif name == "qwen3_8b":
+            tokenizer = Qwen3Tokenizer8B
+
+        super().__init__(embedding_directory=embedding_directory, tokenizer_data=tokenizer_data, name=name, tokenizer=tokenizer)
+        self.llama_template = "<|im_start|>user\n{}<|im_end|>\n<|im_start|>assistant\n<think>\n\n</think>\n\n"
+
+    def tokenize_with_weights(self, text, return_word_ids=False, llama_template=None, **kwargs):
+        if llama_template is None:
+            llama_text = self.llama_template.format(text)
+        else:
+            llama_text = llama_template.format(text)
+
+        tokens = super().tokenize_with_weights(llama_text, return_word_ids=return_word_ids, disable_weights=True, **kwargs)
+        return tokens
+
+class KleinTokenizer8B(KleinTokenizer):
+    def __init__(self, embedding_directory=None, tokenizer_data={}, name="qwen3_8b"):
+        super().__init__(embedding_directory=embedding_directory, tokenizer_data=tokenizer_data, name=name)
+
+class Qwen3_4BModel(sd1_clip.SDClipModel):
+    def __init__(self, device="cpu", layer=[9, 18, 27], layer_idx=None, dtype=None, attention_mask=True, model_options={}):
+        super().__init__(device=device, layer=layer, layer_idx=layer_idx, textmodel_json_config={}, dtype=dtype, special_tokens={"pad": 151643}, layer_norm_hidden_state=False, model_class=comfy.text_encoders.llama.Qwen3_4B, enable_attention_masks=attention_mask, return_attention_masks=attention_mask, model_options=model_options)
+
+class Qwen3_8BModel(sd1_clip.SDClipModel):
+    def __init__(self, device="cpu", layer=[9, 18, 27], layer_idx=None, dtype=None, attention_mask=True, model_options={}):
+        super().__init__(device=device, layer=layer, layer_idx=layer_idx, textmodel_json_config={}, dtype=dtype, special_tokens={"pad": 151643}, layer_norm_hidden_state=False, model_class=comfy.text_encoders.llama.Qwen3_8B, enable_attention_masks=attention_mask, return_attention_masks=attention_mask, model_options=model_options)
+
+def klein_te(dtype_llama=None, llama_quantization_metadata=None, model_type="qwen3_4b"):
+    if model_type == "qwen3_4b":
+        model = Qwen3_4BModel
+    elif model_type == "qwen3_8b":
+        model = Qwen3_8BModel
+
+    class Flux2TEModel_(Flux2TEModel):
+        def __init__(self, device="cpu", dtype=None, model_options={}):
+            if llama_quantization_metadata is not None:
+                model_options = model_options.copy()
+                model_options["quantization_metadata"] = llama_quantization_metadata
+            if dtype_llama is not None:
+                dtype = dtype_llama
+            super().__init__(device=device, dtype=dtype, name=model_type, model_options=model_options, clip_model=model)
+    return Flux2TEModel_

comfy/text_encoders/llama.py

@@ -99,6 +99,28 @@ class Qwen3_4BConfig:
     rope_scale = None
     final_norm: bool = True
 
+@dataclass
+class Qwen3_8BConfig:
+    vocab_size: int = 151936
+    hidden_size: int = 4096
+    intermediate_size: int = 12288
+    num_hidden_layers: int = 36
+    num_attention_heads: int = 32
+    num_key_value_heads: int = 8
+    max_position_embeddings: int = 40960
+    rms_norm_eps: float = 1e-6
+    rope_theta: float = 1000000.0
+    transformer_type: str = "llama"
+    head_dim = 128
+    rms_norm_add = False
+    mlp_activation = "silu"
+    qkv_bias = False
+    rope_dims = None
+    q_norm = "gemma3"
+    k_norm = "gemma3"
+    rope_scale = None
+    final_norm: bool = True
+
 @dataclass
 class Ovis25_2BConfig:
     vocab_size: int = 151936
@@ -628,6 +650,15 @@ class Qwen3_4B(BaseLlama, torch.nn.Module):
         self.model = Llama2_(config, device=device, dtype=dtype, ops=operations)
         self.dtype = dtype
 
+class Qwen3_8B(BaseLlama, torch.nn.Module):
+    def __init__(self, config_dict, dtype, device, operations):
+        super().__init__()
+        config = Qwen3_8BConfig(**config_dict)
+        self.num_layers = config.num_hidden_layers
+
+        self.model = Llama2_(config, device=device, dtype=dtype, ops=operations)
+        self.dtype = dtype
+
 class Ovis25_2B(BaseLlama, torch.nn.Module):
     def __init__(self, config_dict, dtype, device, operations):
         super().__init__()

comfy/text_encoders/lt.py

@@ -118,8 +118,9 @@ class LTXAVTEModel(torch.nn.Module):
             sdo = comfy.utils.state_dict_prefix_replace(sd, {"text_embedding_projection.aggregate_embed.weight": "text_embedding_projection.weight", "model.diffusion_model.video_embeddings_connector.": "video_embeddings_connector.", "model.diffusion_model.audio_embeddings_connector.": "audio_embeddings_connector."}, filter_keys=True)
             if len(sdo) == 0:
                 sdo = sd
-
-            return self.load_state_dict(sdo, strict=False)
+            missing, unexpected = self.load_state_dict(sdo, strict=False)
+            missing = [k for k in missing if not k.startswith("gemma3_12b.")] # filter out keys that belong to the main gemma model
+            return (missing, unexpected)
 
     def memory_estimation_function(self, token_weight_pairs, device=None):
         constant = 6.0

comfy/utils.py

@@ -929,7 +929,9 @@ def bislerp(samples, width, height):
     return result.to(orig_dtype)
 
 def lanczos(samples, width, height):
-    images = [Image.fromarray(np.clip(255. * image.movedim(0, -1).cpu().numpy(), 0, 255).astype(np.uint8)) for image in samples]
+    #the below API is strict and expects grayscale to be squeezed
+    samples = samples.squeeze(1) if samples.shape[1] == 1 else samples.movedim(1, -1)
+    images = [Image.fromarray(np.clip(255. * image.cpu().numpy(), 0, 255).astype(np.uint8)) for image in samples]
     images = [image.resize((width, height), resample=Image.Resampling.LANCZOS) for image in images]
     images = [torch.from_numpy(np.array(image).astype(np.float32) / 255.0).movedim(-1, 0) for image in images]
     result = torch.stack(images)

comfy/worker_native.py

@@ -0,0 +1,95 @@
+"""Native (in-process) worker for prompt execution."""
+
+import time
+import gc
+
+
+class NativeWorker:
+    """Executes prompts in the same process as the server."""
+
+    def __init__(self, server_instance, interrupt_checker=None):
+        self.server_instance = server_instance
+        self.interrupt_checker = interrupt_checker
+        self.executor = None
+        self.last_gc_collect = 0
+        self.need_gc = False
+        self.gc_collect_interval = 10.0
+
+    async def initialize(self):
+        """Load nodes and set up executor. Returns node count."""
+        from execution import PromptExecutor, CacheType
+        from comfy.cli_args import args
+        from comfy.execution_core import init_execution_environment, setup_progress_hook
+        import comfy.model_management as mm
+        import hook_breaker_ac10a0
+
+        hook_breaker_ac10a0.save_functions()
+        try:
+            node_count = await init_execution_environment()
+        finally:
+            hook_breaker_ac10a0.restore_functions()
+
+        interrupt_checker = self.interrupt_checker or mm.throw_exception_if_processing_interrupted
+        setup_progress_hook(self.server_instance, interrupt_checker=interrupt_checker)
+
+        cache_type = CacheType.CLASSIC
+        if args.cache_lru > 0:
+            cache_type = CacheType.LRU
+        elif args.cache_ram > 0:
+            cache_type = CacheType.RAM_PRESSURE
+        elif args.cache_none:
+            cache_type = CacheType.NONE
+
+        self.executor = PromptExecutor(
+            self.server_instance,
+            cache_type=cache_type,
+            cache_args={"lru": args.cache_lru, "ram": args.cache_ram}
+        )
+        return node_count
+
+    def execute_prompt(self, prompt, prompt_id, extra_data, execute_outputs, server=None):
+        self.executor.execute(prompt, prompt_id, extra_data, execute_outputs)
+        return {
+            'success': self.executor.success,
+            'history_result': self.executor.history_result,
+            'status_messages': self.executor.status_messages,
+            'prompt_id': prompt_id
+        }
+
+    def handle_flags(self, flags):
+        import comfy.model_management as mm
+        import hook_breaker_ac10a0
+
+        free_memory = flags.get("free_memory", False)
+
+        if flags.get("unload_models", free_memory):
+            mm.unload_all_models()
+            self.need_gc = True
+            self.last_gc_collect = 0
+
+        if free_memory:
+            if self.executor:
+                self.executor.reset()
+            self.need_gc = True
+            self.last_gc_collect = 0
+
+        if self.need_gc:
+            current_time = time.perf_counter()
+            if (current_time - self.last_gc_collect) > self.gc_collect_interval:
+                gc.collect()
+                mm.soft_empty_cache()
+                self.last_gc_collect = current_time
+                self.need_gc = False
+                hook_breaker_ac10a0.restore_functions()
+
+    def interrupt(self, value=True):
+        import comfy.model_management
+        comfy.model_management.interrupt_current_processing(value)
+
+    def mark_needs_gc(self):
+        self.need_gc = True
+
+    def get_gc_timeout(self):
+        if self.need_gc:
+            return max(self.gc_collect_interval - (time.perf_counter() - self.last_gc_collect), 0.0)
+        return 1000.0

comfy/worker_process.py

@@ -0,0 +1,179 @@
+"""Subprocess worker for isolated prompt execution with complete GPU/ROCm reset."""
+
+import logging
+import multiprocessing as mp
+import time
+import traceback
+
+mp.set_start_method('spawn', force=True)
+
+
+def _deserialize_preview(msg):
+    """Deserialize preview image from IPC transport."""
+    if not (isinstance(msg['data'], dict) and msg['data'].get('_serialized')):
+        return msg
+
+    from PIL import Image
+    from io import BytesIO
+    import base64
+
+    s = msg['data']
+    pil_image = Image.open(BytesIO(base64.b64decode(s['image_bytes'])))
+    msg['data'] = ((s['image_type'], pil_image, s['max_size']), s['metadata'])
+    return msg
+
+
+def _error_result(worker_id, prompt_id, error, tb=None):
+    return {
+        'success': False,
+        'error': error,
+        'traceback': tb,
+        'history_result': {},
+        'status_messages': [],
+        'worker_id': worker_id,
+        'prompt_id': prompt_id
+    }
+
+
+def _kill_worker(worker, worker_id):
+    if not worker.is_alive():
+        return
+    worker.terminate()
+    worker.join(timeout=2)
+    if worker.is_alive():
+        logging.warning(f"Worker {worker_id} didn't terminate, killing")
+        worker.kill()
+        worker.join()
+
+
+class SubprocessWorker:
+    """Executes each prompt in an isolated subprocess with fresh GPU context."""
+
+    def __init__(self, server_instance, timeout=600):
+        self.server_instance = server_instance
+        self.timeout = timeout
+        self.worker_counter = 0
+        self.current_worker = None
+        self.interrupt_event = None
+        logging.info("SubprocessWorker created - each job will run in isolated process")
+
+    async def initialize(self):
+        """Load node definitions for prompt validation. Returns node count."""
+        from comfy.execution_core import init_execution_environment
+        return await init_execution_environment()
+
+    def handle_flags(self, flags):
+        pass
+
+    def mark_needs_gc(self):
+        pass
+
+    def get_gc_timeout(self):
+        return 1000.0
+
+    def interrupt(self, value=True):
+        if not value:
+            return
+        if self.interrupt_event:
+            self.interrupt_event.set()
+        if self.current_worker and self.current_worker.is_alive():
+            self.current_worker.join(timeout=2)
+            _kill_worker(self.current_worker, self.worker_counter)
+            self.current_worker = None
+
+    def _relay_messages(self, message_queue, server):
+        """Relay queued messages to UI."""
+        while not message_queue.empty():
+            try:
+                msg = _deserialize_preview(message_queue.get_nowait())
+                if server:
+                    server.send_sync(msg['event'], msg['data'], msg['sid'])
+            except:
+                break
+
+    def execute_prompt(self, prompt, prompt_id, extra_data={}, execute_outputs=[], server=None):
+        self.worker_counter += 1
+        worker_id = self.worker_counter
+
+        job_queue = mp.Queue()
+        result_queue = mp.Queue()
+        message_queue = mp.Queue()
+        self.interrupt_event = mp.Event()
+
+        client_id = extra_data.get('client_id')
+        client_metadata = {}
+        if client_id and hasattr(server, 'sockets_metadata'):
+            client_metadata = server.sockets_metadata.get(client_id, {})
+
+        job_data = {
+            'prompt': prompt,
+            'prompt_id': prompt_id,
+            'extra_data': extra_data,
+            'execute_outputs': execute_outputs,
+            'client_sockets_metadata': client_metadata
+        }
+
+        from comfy.worker_process_child import worker_main
+        worker = mp.Process(
+            target=worker_main,
+            args=(job_queue, result_queue, message_queue, self.interrupt_event, worker_id),
+            name=f'ComfyUI-Worker-{worker_id}'
+        )
+
+        logging.info(f"Starting worker {worker_id} for prompt {prompt_id}")
+        self.current_worker = worker
+        worker.start()
+        job_queue.put(job_data)
+
+        try:
+            start_time = time.time()
+            result = None
+
+            while result is None:
+                if self.interrupt_event.is_set():
+                    logging.info(f"Worker {worker_id} interrupted")
+                    if server:
+                        server.send_sync("executing", {"node": None, "prompt_id": prompt_id}, server.client_id)
+                    return _error_result(worker_id, prompt_id, 'Execution interrupted by user')
+
+                if time.time() - start_time > self.timeout:
+                    raise TimeoutError()
+
+                self._relay_messages(message_queue, server)
+
+                try:
+                    result = result_queue.get(timeout=0.1)
+                except mp.queues.Empty:
+                    pass
+
+            self._relay_messages(message_queue, server)
+
+            worker.join(timeout=5)
+            if worker.is_alive():
+                _kill_worker(worker, worker_id)
+
+            logging.info(f"Worker {worker_id} cleaned up (exit code: {worker.exitcode})")
+            self.current_worker = None
+            return result
+
+        except TimeoutError:
+            error = f"Worker {worker_id} timed out after {self.timeout}s. Try --subprocess-timeout to increase."
+            logging.error(error)
+            _kill_worker(worker, worker_id)
+            self.current_worker = None
+            return _error_result(worker_id, prompt_id, error)
+
+        except Exception as e:
+            error = f"Worker {worker_id} IPC error: {e}"
+            logging.error(f"{error}\n{traceback.format_exc()}")
+            _kill_worker(worker, worker_id)
+            self.current_worker = None
+            return _error_result(worker_id, prompt_id, error, traceback.format_exc())
+
+        finally:
+            for q in (job_queue, result_queue, message_queue):
+                q.close()
+                try:
+                    q.join_thread()
+                except:
+                    pass

comfy/worker_process_child.py

@@ -0,0 +1,104 @@
+"""Subprocess worker child process entry point."""
+
+import logging
+import multiprocessing as mp
+import traceback
+
+
+class IPCMessageServer:
+    """IPC-based message server for subprocess workers."""
+
+    def __init__(self, message_queue, client_id=None, sockets_metadata=None):
+        self.message_queue = message_queue
+        self.client_id = client_id
+        self.last_node_id = None
+        self.last_prompt_id = None
+        self.sockets_metadata = sockets_metadata or {}
+
+    def send_sync(self, event, data, sid=None):
+        from protocol import BinaryEventTypes
+        from io import BytesIO
+        import base64
+
+        if event == BinaryEventTypes.PREVIEW_IMAGE_WITH_METADATA and isinstance(data, tuple):
+            preview_image, metadata = data
+            image_type, pil_image, max_size = preview_image
+
+            buffer = BytesIO()
+            pil_image.save(buffer, format=image_type)
+
+            data = {
+                '_serialized': True,
+                'image_type': image_type,
+                'image_bytes': base64.b64encode(buffer.getvalue()).decode('utf-8'),
+                'max_size': max_size,
+                'metadata': metadata
+            }
+
+        self.message_queue.put_nowait({'event': event, 'data': data, 'sid': sid})
+
+    def queue_updated(self):
+        pass
+
+
+def worker_main(job_queue, result_queue, message_queue, interrupt_event, worker_id):
+    """Subprocess worker entry point - spawned fresh for each execution."""
+    job_data = None
+    try:
+        logging.basicConfig(level=logging.INFO, format=f'[Worker-{worker_id}] %(levelname)s: %(message)s')
+        logging.info(f"Worker {worker_id} starting (PID: {mp.current_process().pid})")
+
+        import asyncio
+        import comfy.model_management
+        from comfy.worker_native import NativeWorker
+        from comfy.execution_core import WorkerServer
+
+        logging.info(f"Worker {worker_id} initialized. Device: {comfy.model_management.get_torch_device()}")
+
+        job_data = job_queue.get(timeout=30)
+        client_id = job_data.get('extra_data', {}).get('client_id')
+        client_metadata = job_data.get('client_sockets_metadata', {})
+
+        sockets_metadata = {client_id: client_metadata} if client_id and client_metadata else {}
+        ipc_server = IPCMessageServer(message_queue, client_id, sockets_metadata)
+        server = WorkerServer(ipc_server)
+
+        def check_interrupt():
+            if interrupt_event.is_set():
+                raise comfy.model_management.InterruptProcessingException()
+
+        worker = NativeWorker(server, interrupt_checker=check_interrupt)
+
+        import comfy.execution_core
+        comfy.execution_core._active_worker = worker
+
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+        node_count = loop.run_until_complete(worker.initialize())
+        logging.info(f"Worker {worker_id} loaded {node_count} node types")
+
+        result = worker.execute_prompt(
+            job_data['prompt'],
+            job_data['prompt_id'],
+            job_data.get('extra_data', {}),
+            job_data.get('execute_outputs', [])
+        )
+        result['worker_id'] = worker_id
+
+        logging.info(f"Worker {worker_id} completed successfully")
+        result_queue.put(result)
+
+    except Exception as e:
+        logging.error(f"Worker {worker_id} failed: {e}\n{traceback.format_exc()}")
+        result_queue.put({
+            'success': False,
+            'error': str(e),
+            'traceback': traceback.format_exc(),
+            'history_result': {},
+            'status_messages': [],
+            'worker_id': worker_id,
+            'prompt_id': job_data.get('prompt_id', 'unknown') if job_data else 'unknown'
+        })
+
+    finally:
+        logging.info(f"Worker {worker_id} exiting")

comfy_api_nodes/apis/meshy.py

@@ -0,0 +1,160 @@
+from typing import TypedDict
+
+from pydantic import BaseModel, Field
+
+from comfy_api.latest import Input
+
+
+class InputShouldRemesh(TypedDict):
+    should_remesh: str
+    topology: str
+    target_polycount: int
+
+
+class InputShouldTexture(TypedDict):
+    should_texture: str
+    enable_pbr: bool
+    texture_prompt: str
+    texture_image: Input.Image | None
+
+
+class MeshyTaskResponse(BaseModel):
+    result: str = Field(...)
+
+
+class MeshyTextToModelRequest(BaseModel):
+    mode: str = Field("preview")
+    prompt: str = Field(..., max_length=600)
+    art_style: str = Field(..., description="'realistic' or 'sculpture'")
+    ai_model: str = Field(...)
+    topology: str | None = Field(..., description="'quad' or 'triangle'")
+    target_polycount: int | None = Field(..., ge=100, le=300000)
+    should_remesh: bool = Field(
+        True,
+        description="False returns the original mesh, ignoring topology and polycount.",
+    )
+    symmetry_mode: str = Field(..., description="'auto', 'off' or 'on'")
+    pose_mode: str = Field(...)
+    seed: int = Field(...)
+    moderation: bool = Field(False)
+
+
+class MeshyRefineTask(BaseModel):
+    mode: str = Field("refine")
+    preview_task_id: str = Field(...)
+    enable_pbr: bool | None = Field(...)
+    texture_prompt: str | None = Field(...)
+    texture_image_url: str | None = Field(...)
+    ai_model: str = Field(...)
+    moderation: bool = Field(False)
+
+
+class MeshyImageToModelRequest(BaseModel):
+    image_url: str = Field(...)
+    ai_model: str = Field(...)
+    topology: str | None = Field(..., description="'quad' or 'triangle'")
+    target_polycount: int | None = Field(..., ge=100, le=300000)
+    symmetry_mode: str = Field(..., description="'auto', 'off' or 'on'")
+    should_remesh: bool = Field(
+        True,
+        description="False returns the original mesh, ignoring topology and polycount.",
+    )
+    should_texture: bool = Field(...)
+    enable_pbr: bool | None = Field(...)
+    pose_mode: str = Field(...)
+    texture_prompt: str | None = Field(None, max_length=600)
+    texture_image_url: str | None = Field(None)
+    seed: int = Field(...)
+    moderation: bool = Field(False)
+
+
+class MeshyMultiImageToModelRequest(BaseModel):
+    image_urls: list[str] = Field(...)
+    ai_model: str = Field(...)
+    topology: str | None = Field(..., description="'quad' or 'triangle'")
+    target_polycount: int | None = Field(..., ge=100, le=300000)
+    symmetry_mode: str = Field(..., description="'auto', 'off' or 'on'")
+    should_remesh: bool = Field(
+        True,
+        description="False returns the original mesh, ignoring topology and polycount.",
+    )
+    should_texture: bool = Field(...)
+    enable_pbr: bool | None = Field(...)
+    pose_mode: str = Field(...)
+    texture_prompt: str | None = Field(None, max_length=600)
+    texture_image_url: str | None = Field(None)
+    seed: int = Field(...)
+    moderation: bool = Field(False)
+
+
+class MeshyRiggingRequest(BaseModel):
+    input_task_id: str = Field(...)
+    height_meters: float = Field(...)
+    texture_image_url: str | None = Field(...)
+
+
+class MeshyAnimationRequest(BaseModel):
+    rig_task_id: str = Field(...)
+    action_id: int = Field(...)
+
+
+class MeshyTextureRequest(BaseModel):
+    input_task_id: str = Field(...)
+    ai_model: str = Field(...)
+    enable_original_uv: bool = Field(...)
+    enable_pbr: bool = Field(...)
+    text_style_prompt: str | None = Field(...)
+    image_style_url: str | None = Field(...)
+
+
+class MeshyModelsUrls(BaseModel):
+    glb: str = Field("")
+
+
+class MeshyRiggedModelsUrls(BaseModel):
+    rigged_character_glb_url: str = Field("")
+
+
+class MeshyAnimatedModelsUrls(BaseModel):
+    animation_glb_url: str = Field("")
+
+
+class MeshyResultTextureUrls(BaseModel):
+    base_color: str = Field(...)
+    metallic: str | None = Field(None)
+    normal: str | None = Field(None)
+    roughness: str | None = Field(None)
+
+
+class MeshyTaskError(BaseModel):
+    message: str | None = Field(None)
+
+
+class MeshyModelResult(BaseModel):
+    id: str = Field(...)
+    type: str = Field(...)
+    model_urls: MeshyModelsUrls = Field(MeshyModelsUrls())
+    thumbnail_url: str = Field(...)
+    video_url: str | None = Field(None)
+    status: str = Field(...)
+    progress: int = Field(0)
+    texture_urls: list[MeshyResultTextureUrls] | None = Field([])
+    task_error: MeshyTaskError | None = Field(None)
+
+
+class MeshyRiggedResult(BaseModel):
+    id: str = Field(...)
+    type: str = Field(...)
+    status: str = Field(...)
+    progress: int = Field(0)
+    result: MeshyRiggedModelsUrls = Field(MeshyRiggedModelsUrls())
+    task_error: MeshyTaskError | None = Field(None)
+
+
+class MeshyAnimationResult(BaseModel):
+    id: str = Field(...)
+    type: str = Field(...)
+    status: str = Field(...)
+    progress: int = Field(0)
+    result: MeshyAnimatedModelsUrls = Field(MeshyAnimatedModelsUrls())
+    task_error: MeshyTaskError | None = Field(None)

comfy_api_nodes/nodes_meshy.py

@@ -0,0 +1,790 @@
+import os
+
+from typing_extensions import override
+
+from comfy_api.latest import IO, ComfyExtension, Input
+from comfy_api_nodes.apis.meshy import (
+    InputShouldRemesh,
+    InputShouldTexture,
+    MeshyAnimationRequest,
+    MeshyAnimationResult,
+    MeshyImageToModelRequest,
+    MeshyModelResult,
+    MeshyMultiImageToModelRequest,
+    MeshyRefineTask,
+    MeshyRiggedResult,
+    MeshyRiggingRequest,
+    MeshyTaskResponse,
+    MeshyTextToModelRequest,
+    MeshyTextureRequest,
+)
+from comfy_api_nodes.util import (
+    ApiEndpoint,
+    download_url_to_bytesio,
+    poll_op,
+    sync_op,
+    upload_images_to_comfyapi,
+    validate_string,
+)
+from folder_paths import get_output_directory
+
+
+class MeshyTextToModelNode(IO.ComfyNode):
+
+    @classmethod
+    def define_schema(cls):
+        return IO.Schema(
+            node_id="MeshyTextToModelNode",
+            display_name="Meshy: Text to Model",
+            category="api node/3d/Meshy",
+            inputs=[
+                IO.Combo.Input("model", options=["latest"]),
+                IO.String.Input("prompt", multiline=True, default=""),
+                IO.Combo.Input("style", options=["realistic", "sculpture"]),
+                IO.DynamicCombo.Input(
+                    "should_remesh",
+                    options=[
+                        IO.DynamicCombo.Option(
+                            "true",
+                            [
+                                IO.Combo.Input("topology", options=["triangle", "quad"]),
+                                IO.Int.Input(
+                                    "target_polycount",
+                                    default=300000,
+                                    min=100,
+                                    max=300000,
+                                    display_mode=IO.NumberDisplay.number,
+                                ),
+                            ],
+                        ),
+                        IO.DynamicCombo.Option("false", []),
+                    ],
+                    tooltip="When set to false, returns an unprocessed triangular mesh.",
+                ),
+                IO.Combo.Input("symmetry_mode", options=["auto", "on", "off"]),
+                IO.Combo.Input(
+                    "pose_mode",
+                    options=["", "A-pose", "T-pose"],
+                    tooltip="Specify the pose mode for the generated model.",
+                ),
+                IO.Int.Input(
+                    "seed",
+                    default=0,
+                    min=0,
+                    max=2147483647,
+                    display_mode=IO.NumberDisplay.number,
+                    control_after_generate=True,
+                    tooltip="Seed controls whether the node should re-run; "
+                    "results are non-deterministic regardless of seed.",
+                ),
+            ],
+            outputs=[
+                IO.String.Output(display_name="model_file"),
+                IO.Custom("MESHY_TASK_ID").Output(display_name="meshy_task_id"),
+            ],
+            hidden=[
+                IO.Hidden.auth_token_comfy_org,
+                IO.Hidden.api_key_comfy_org,
+                IO.Hidden.unique_id,
+            ],
+            is_api_node=True,
+            is_output_node=True,
+            price_badge=IO.PriceBadge(
+                expr="""{"type":"usd","usd":0.8}""",
+            ),
+        )
+
+    @classmethod
+    async def execute(
+        cls,
+        model: str,
+        prompt: str,
+        style: str,
+        should_remesh: InputShouldRemesh,
+        symmetry_mode: str,
+        pose_mode: str,
+        seed: int,
+    ) -> IO.NodeOutput:
+        validate_string(prompt, field_name="prompt", min_length=1, max_length=600)
+        response = await sync_op(
+            cls,
+            ApiEndpoint(path="/proxy/meshy/openapi/v2/text-to-3d", method="POST"),
+            response_model=MeshyTaskResponse,
+            data=MeshyTextToModelRequest(
+                prompt=prompt,
+                art_style=style,
+                ai_model=model,
+                topology=should_remesh.get("topology", None),
+                target_polycount=should_remesh.get("target_polycount", None),
+                should_remesh=should_remesh["should_remesh"] == "true",
+                symmetry_mode=symmetry_mode,
+                pose_mode=pose_mode.lower(),
+                seed=seed,
+            ),
+        )
+        result = await poll_op(
+            cls,
+            ApiEndpoint(path=f"/proxy/meshy/openapi/v2/text-to-3d/{response.result}"),
+            response_model=MeshyModelResult,
+            status_extractor=lambda r: r.status,
+            progress_extractor=lambda r: r.progress,
+        )
+        model_file = f"meshy_model_{response.result}.glb"
+        await download_url_to_bytesio(result.model_urls.glb, os.path.join(get_output_directory(), model_file))
+        return IO.NodeOutput(model_file, response.result)
+
+
+class MeshyRefineNode(IO.ComfyNode):
+
+    @classmethod
+    def define_schema(cls):
+        return IO.Schema(
+            node_id="MeshyRefineNode",
+            display_name="Meshy: Refine Draft Model",
+            category="api node/3d/Meshy",
+            description="Refine a previously created draft model.",
+            inputs=[
+                IO.Combo.Input("model", options=["latest"]),
+                IO.Custom("MESHY_TASK_ID").Input("meshy_task_id"),
+                IO.Boolean.Input(
+                    "enable_pbr",
+                    default=False,
+                    tooltip="Generate PBR Maps (metallic, roughness, normal) in addition to the base color. "
+                    "Note: this should be set to false when using Sculpture style, "
+                    "as Sculpture style generates its own set of PBR maps.",
+                ),
+                IO.String.Input(
+                    "texture_prompt",
+                    default="",
+                    multiline=True,
+                    tooltip="Provide a text prompt to guide the texturing process. "
+                    "Maximum 600 characters. Cannot be used at the same time as 'texture_image'.",
+                ),
+                IO.Image.Input(
+                    "texture_image",
+                    tooltip="Only one of 'texture_image' or 'texture_prompt' may be used at the same time.",
+                    optional=True,
+                ),
+            ],
+            outputs=[
+                IO.String.Output(display_name="model_file"),
+                IO.Custom("MESHY_TASK_ID").Output(display_name="meshy_task_id"),
+            ],
+            hidden=[
+                IO.Hidden.auth_token_comfy_org,
+                IO.Hidden.api_key_comfy_org,
+                IO.Hidden.unique_id,
+            ],
+            is_api_node=True,
+            is_output_node=True,
+            price_badge=IO.PriceBadge(
+                expr="""{"type":"usd","usd":0.4}""",
+            ),
+        )
+
+    @classmethod
+    async def execute(
+        cls,
+        model: str,
+        meshy_task_id: str,
+        enable_pbr: bool,
+        texture_prompt: str,
+        texture_image: Input.Image | None = None,
+    ) -> IO.NodeOutput:
+        if texture_prompt and texture_image is not None:
+            raise ValueError("texture_prompt and texture_image cannot be used at the same time")
+        texture_image_url = None
+        if texture_prompt:
+            validate_string(texture_prompt, field_name="texture_prompt", max_length=600)
+        if texture_image is not None:
+            texture_image_url = (await upload_images_to_comfyapi(cls, texture_image, wait_label="Uploading texture"))[0]
+        response = await sync_op(
+            cls,
+            endpoint=ApiEndpoint(path="/proxy/meshy/openapi/v2/text-to-3d", method="POST"),
+            response_model=MeshyTaskResponse,
+            data=MeshyRefineTask(
+                preview_task_id=meshy_task_id,
+                enable_pbr=enable_pbr,
+                texture_prompt=texture_prompt if texture_prompt else None,
+                texture_image_url=texture_image_url,
+                ai_model=model,
+            ),
+        )
+        result = await poll_op(
+            cls,
+            ApiEndpoint(path=f"/proxy/meshy/openapi/v2/text-to-3d/{response.result}"),
+            response_model=MeshyModelResult,
+            status_extractor=lambda r: r.status,
+            progress_extractor=lambda r: r.progress,
+        )
+        model_file = f"meshy_model_{response.result}.glb"
+        await download_url_to_bytesio(result.model_urls.glb, os.path.join(get_output_directory(), model_file))
+        return IO.NodeOutput(model_file, response.result)
+
+
+class MeshyImageToModelNode(IO.ComfyNode):
+
+    @classmethod
+    def define_schema(cls):
+        return IO.Schema(
+            node_id="MeshyImageToModelNode",
+            display_name="Meshy: Image to Model",
+            category="api node/3d/Meshy",
+            inputs=[
+                IO.Combo.Input("model", options=["latest"]),
+                IO.Image.Input("image"),
+                IO.DynamicCombo.Input(
+                    "should_remesh",
+                    options=[
+                        IO.DynamicCombo.Option(
+                            "true",
+                            [
+                                IO.Combo.Input("topology", options=["triangle", "quad"]),
+                                IO.Int.Input(
+                                    "target_polycount",
+                                    default=300000,
+                                    min=100,
+                                    max=300000,
+                                    display_mode=IO.NumberDisplay.number,
+                                ),
+                            ],
+                        ),
+                        IO.DynamicCombo.Option("false", []),
+                    ],
+                    tooltip="When set to false, returns an unprocessed triangular mesh.",
+                ),
+                IO.Combo.Input("symmetry_mode", options=["auto", "on", "off"]),
+                IO.DynamicCombo.Input(
+                    "should_texture",
+                    options=[
+                        IO.DynamicCombo.Option(
+                            "true",
+                            [
+                                IO.Boolean.Input(
+                                    "enable_pbr",
+                                    default=False,
+                                    tooltip="Generate PBR Maps (metallic, roughness, normal) "
+                                    "in addition to the base color.",
+                                ),
+                                IO.String.Input(
+                                    "texture_prompt",
+                                    default="",
+                                    multiline=True,
+                                    tooltip="Provide a text prompt to guide the texturing process. "
+                                    "Maximum 600 characters. Cannot be used at the same time as 'texture_image'.",
+                                ),
+                                IO.Image.Input(
+                                    "texture_image",
+                                    tooltip="Only one of 'texture_image' or 'texture_prompt' "
+                                    "may be used at the same time.",
+                                    optional=True,
+                                ),
+                            ],
+                        ),
+                        IO.DynamicCombo.Option("false", []),
+                    ],
+                    tooltip="Determines whether textures are generated. "
+                    "Setting it to false skips the texture phase and returns a mesh without textures.",
+                ),
+                IO.Combo.Input(
+                    "pose_mode",
+                    options=["", "A-pose", "T-pose"],
+                    tooltip="Specify the pose mode for the generated model.",
+                ),
+                IO.Int.Input(
+                    "seed",
+                    default=0,
+                    min=0,
+                    max=2147483647,
+                    display_mode=IO.NumberDisplay.number,
+                    control_after_generate=True,
+                    tooltip="Seed controls whether the node should re-run; "
+                    "results are non-deterministic regardless of seed.",
+                ),
+            ],
+            outputs=[
+                IO.String.Output(display_name="model_file"),
+                IO.Custom("MESHY_TASK_ID").Output(display_name="meshy_task_id"),
+            ],
+            hidden=[
+                IO.Hidden.auth_token_comfy_org,
+                IO.Hidden.api_key_comfy_org,
+                IO.Hidden.unique_id,
+            ],
+            is_api_node=True,
+            is_output_node=True,
+            price_badge=IO.PriceBadge(
+                depends_on=IO.PriceBadgeDepends(widgets=["should_texture"]),
+                expr="""
+                (
+                  $prices := {"true": 1.2, "false": 0.8};
+                  {"type":"usd","usd": $lookup($prices, widgets.should_texture)}
+                )
+                """,
+            ),
+        )
+
+    @classmethod
+    async def execute(
+        cls,
+        model: str,
+        image: Input.Image,
+        should_remesh: InputShouldRemesh,
+        symmetry_mode: str,
+        should_texture: InputShouldTexture,
+        pose_mode: str,
+        seed: int,
+    ) -> IO.NodeOutput:
+        texture = should_texture["should_texture"] == "true"
+        texture_image_url = texture_prompt = None
+        if texture:
+            if should_texture["texture_prompt"] and should_texture["texture_image"] is not None:
+                raise ValueError("texture_prompt and texture_image cannot be used at the same time")
+            if should_texture["texture_prompt"]:
+                validate_string(should_texture["texture_prompt"], field_name="texture_prompt", max_length=600)
+                texture_prompt = should_texture["texture_prompt"]
+            if should_texture["texture_image"] is not None:
+                texture_image_url = (
+                    await upload_images_to_comfyapi(
+                        cls, should_texture["texture_image"], wait_label="Uploading texture"
+                    )
+                )[0]
+        response = await sync_op(
+            cls,
+            ApiEndpoint(path="/proxy/meshy/openapi/v1/image-to-3d", method="POST"),
+            response_model=MeshyTaskResponse,
+            data=MeshyImageToModelRequest(
+                image_url=(await upload_images_to_comfyapi(cls, image, wait_label="Uploading base image"))[0],
+                ai_model=model,
+                topology=should_remesh.get("topology", None),
+                target_polycount=should_remesh.get("target_polycount", None),
+                symmetry_mode=symmetry_mode,
+                should_remesh=should_remesh["should_remesh"] == "true",
+                should_texture=texture,
+                enable_pbr=should_texture.get("enable_pbr", None),
+                pose_mode=pose_mode.lower(),
+                texture_prompt=texture_prompt,
+                texture_image_url=texture_image_url,
+                seed=seed,
+            ),
+        )
+        result = await poll_op(
+            cls,
+            ApiEndpoint(path=f"/proxy/meshy/openapi/v1/image-to-3d/{response.result}"),
+            response_model=MeshyModelResult,
+            status_extractor=lambda r: r.status,
+            progress_extractor=lambda r: r.progress,
+        )
+        model_file = f"meshy_model_{response.result}.glb"
+        await download_url_to_bytesio(result.model_urls.glb, os.path.join(get_output_directory(), model_file))
+        return IO.NodeOutput(model_file, response.result)
+
+
+class MeshyMultiImageToModelNode(IO.ComfyNode):
+
+    @classmethod
+    def define_schema(cls):
+        return IO.Schema(
+            node_id="MeshyMultiImageToModelNode",
+            display_name="Meshy: Multi-Image to Model",
+            category="api node/3d/Meshy",
+            inputs=[
+                IO.Combo.Input("model", options=["latest"]),
+                IO.Autogrow.Input(
+                    "images",
+                    template=IO.Autogrow.TemplatePrefix(IO.Image.Input("image"), prefix="image", min=2, max=4),
+                ),
+                IO.DynamicCombo.Input(
+                    "should_remesh",
+                    options=[
+                        IO.DynamicCombo.Option(
+                            "true",
+                            [
+                                IO.Combo.Input("topology", options=["triangle", "quad"]),
+                                IO.Int.Input(
+                                    "target_polycount",
+                                    default=300000,
+                                    min=100,
+                                    max=300000,
+                                    display_mode=IO.NumberDisplay.number,
+                                ),
+                            ],
+                        ),
+                        IO.DynamicCombo.Option("false", []),
+                    ],
+                    tooltip="When set to false, returns an unprocessed triangular mesh.",
+                ),
+                IO.Combo.Input("symmetry_mode", options=["auto", "on", "off"]),
+                IO.DynamicCombo.Input(
+                    "should_texture",
+                    options=[
+                        IO.DynamicCombo.Option(
+                            "true",
+                            [
+                                IO.Boolean.Input(
+                                    "enable_pbr",
+                                    default=False,
+                                    tooltip="Generate PBR Maps (metallic, roughness, normal) "
+                                    "in addition to the base color.",
+                                ),
+                                IO.String.Input(
+                                    "texture_prompt",
+                                    default="",
+                                    multiline=True,
+                                    tooltip="Provide a text prompt to guide the texturing process. "
+                                    "Maximum 600 characters. Cannot be used at the same time as 'texture_image'.",
+                                ),
+                                IO.Image.Input(
+                                    "texture_image",
+                                    tooltip="Only one of 'texture_image' or 'texture_prompt' "
+                                    "may be used at the same time.",
+                                    optional=True,
+                                ),
+                            ],
+                        ),
+                        IO.DynamicCombo.Option("false", []),
+                    ],
+                    tooltip="Determines whether textures are generated. "
+                    "Setting it to false skips the texture phase and returns a mesh without textures.",
+                ),
+                IO.Combo.Input(
+                    "pose_mode",
+                    options=["", "A-pose", "T-pose"],
+                    tooltip="Specify the pose mode for the generated model.",
+                ),
+                IO.Int.Input(
+                    "seed",
+                    default=0,
+                    min=0,
+                    max=2147483647,
+                    display_mode=IO.NumberDisplay.number,
+                    control_after_generate=True,
+                    tooltip="Seed controls whether the node should re-run; "
+                    "results are non-deterministic regardless of seed.",
+                ),
+            ],
+            outputs=[
+                IO.String.Output(display_name="model_file"),
+                IO.Custom("MESHY_TASK_ID").Output(display_name="meshy_task_id"),
+            ],
+            hidden=[
+                IO.Hidden.auth_token_comfy_org,
+                IO.Hidden.api_key_comfy_org,
+                IO.Hidden.unique_id,
+            ],
+            is_api_node=True,
+            is_output_node=True,
+            price_badge=IO.PriceBadge(
+                depends_on=IO.PriceBadgeDepends(widgets=["should_texture"]),
+                expr="""
+                (
+                  $prices := {"true": 0.6, "false": 0.2};
+                  {"type":"usd","usd": $lookup($prices, widgets.should_texture)}
+                )
+                """,
+            ),
+        )
+
+    @classmethod
+    async def execute(
+        cls,
+        model: str,
+        images: IO.Autogrow.Type,
+        should_remesh: InputShouldRemesh,
+        symmetry_mode: str,
+        should_texture: InputShouldTexture,
+        pose_mode: str,
+        seed: int,
+    ) -> IO.NodeOutput:
+        texture = should_texture["should_texture"] == "true"
+        texture_image_url = texture_prompt = None
+        if texture:
+            if should_texture["texture_prompt"] and should_texture["texture_image"] is not None:
+                raise ValueError("texture_prompt and texture_image cannot be used at the same time")
+            if should_texture["texture_prompt"]:
+                validate_string(should_texture["texture_prompt"], field_name="texture_prompt", max_length=600)
+                texture_prompt = should_texture["texture_prompt"]
+            if should_texture["texture_image"] is not None:
+                texture_image_url = (
+                    await upload_images_to_comfyapi(
+                        cls, should_texture["texture_image"], wait_label="Uploading texture"
+                    )
+                )[0]
+        response = await sync_op(
+            cls,
+            ApiEndpoint(path="/proxy/meshy/openapi/v1/multi-image-to-3d", method="POST"),
+            response_model=MeshyTaskResponse,
+            data=MeshyMultiImageToModelRequest(
+                image_urls=await upload_images_to_comfyapi(
+                    cls, list(images.values()), wait_label="Uploading base images"
+                ),
+                ai_model=model,
+                topology=should_remesh.get("topology", None),
+                target_polycount=should_remesh.get("target_polycount", None),
+                symmetry_mode=symmetry_mode,
+                should_remesh=should_remesh["should_remesh"] == "true",
+                should_texture=texture,
+                enable_pbr=should_texture.get("enable_pbr", None),
+                pose_mode=pose_mode.lower(),
+                texture_prompt=texture_prompt,
+                texture_image_url=texture_image_url,
+                seed=seed,
+            ),
+        )
+        result = await poll_op(
+            cls,
+            ApiEndpoint(path=f"/proxy/meshy/openapi/v1/multi-image-to-3d/{response.result}"),
+            response_model=MeshyModelResult,
+            status_extractor=lambda r: r.status,
+            progress_extractor=lambda r: r.progress,
+        )
+        model_file = f"meshy_model_{response.result}.glb"
+        await download_url_to_bytesio(result.model_urls.glb, os.path.join(get_output_directory(), model_file))
+        return IO.NodeOutput(model_file, response.result)
+
+
+class MeshyRigModelNode(IO.ComfyNode):
+
+    @classmethod
+    def define_schema(cls):
+        return IO.Schema(
+            node_id="MeshyRigModelNode",
+            display_name="Meshy: Rig Model",
+            category="api node/3d/Meshy",
+            description="Provides a rigged character in standard formats. "
+            "Auto-rigging is currently not suitable for untextured meshes, non-humanoid assets, "
+            "or humanoid assets with unclear limb and body structure.",
+            inputs=[
+                IO.Custom("MESHY_TASK_ID").Input("meshy_task_id"),
+                IO.Float.Input(
+                    "height_meters",
+                    min=0.1,
+                    max=15.0,
+                    default=1.7,
+                    tooltip="The approximate height of the character model in meters. "
+                    "This aids in scaling and rigging accuracy.",
+                ),
+                IO.Image.Input(
+                    "texture_image",
+                    tooltip="The model's UV-unwrapped base color texture image.",
+                    optional=True,
+                ),
+            ],
+            outputs=[
+                IO.String.Output(display_name="model_file"),
+                IO.Custom("MESHY_RIGGED_TASK_ID").Output(display_name="rig_task_id"),
+            ],
+            hidden=[
+                IO.Hidden.auth_token_comfy_org,
+                IO.Hidden.api_key_comfy_org,
+                IO.Hidden.unique_id,
+            ],
+            is_api_node=True,
+            is_output_node=True,
+            price_badge=IO.PriceBadge(
+                expr="""{"type":"usd","usd":0.2}""",
+            ),
+        )
+
+    @classmethod
+    async def execute(
+        cls,
+        meshy_task_id: str,
+        height_meters: float,
+        texture_image: Input.Image | None = None,
+    ) -> IO.NodeOutput:
+        texture_image_url = None
+        if texture_image is not None:
+            texture_image_url = (await upload_images_to_comfyapi(cls, texture_image, wait_label="Uploading texture"))[0]
+        response = await sync_op(
+            cls,
+            endpoint=ApiEndpoint(path="/proxy/meshy/openapi/v1/rigging", method="POST"),
+            response_model=MeshyTaskResponse,
+            data=MeshyRiggingRequest(
+                input_task_id=meshy_task_id,
+                height_meters=height_meters,
+                texture_image_url=texture_image_url,
+            ),
+        )
+        result = await poll_op(
+            cls,
+            ApiEndpoint(path=f"/proxy/meshy/openapi/v1/rigging/{response.result}"),
+            response_model=MeshyRiggedResult,
+            status_extractor=lambda r: r.status,
+            progress_extractor=lambda r: r.progress,
+        )
+        model_file = f"meshy_model_{response.result}.glb"
+        await download_url_to_bytesio(
+            result.result.rigged_character_glb_url, os.path.join(get_output_directory(), model_file)
+        )
+        return IO.NodeOutput(model_file, response.result)
+
+
+class MeshyAnimateModelNode(IO.ComfyNode):
+
+    @classmethod
+    def define_schema(cls):
+        return IO.Schema(
+            node_id="MeshyAnimateModelNode",
+            display_name="Meshy: Animate Model",
+            category="api node/3d/Meshy",
+            description="Apply a specific animation action to a previously rigged character.",
+            inputs=[
+                IO.Custom("MESHY_RIGGED_TASK_ID").Input("rig_task_id"),
+                IO.Int.Input(
+                    "action_id",
+                    default=0,
+                    min=0,
+                    max=696,
+                    tooltip="Visit https://docs.meshy.ai/en/api/animation-library for a list of available values.",
+                ),
+            ],
+            outputs=[
+                IO.String.Output(display_name="model_file"),
+            ],
+            hidden=[
+                IO.Hidden.auth_token_comfy_org,
+                IO.Hidden.api_key_comfy_org,
+                IO.Hidden.unique_id,
+            ],
+            is_api_node=True,
+            is_output_node=True,
+            price_badge=IO.PriceBadge(
+                expr="""{"type":"usd","usd":0.12}""",
+            ),
+        )
+
+    @classmethod
+    async def execute(
+        cls,
+        rig_task_id: str,
+        action_id: int,
+    ) -> IO.NodeOutput:
+        response = await sync_op(
+            cls,
+            endpoint=ApiEndpoint(path="/proxy/meshy/openapi/v1/animations", method="POST"),
+            response_model=MeshyTaskResponse,
+            data=MeshyAnimationRequest(
+                rig_task_id=rig_task_id,
+                action_id=action_id,
+            ),
+        )
+        result = await poll_op(
+            cls,
+            ApiEndpoint(path=f"/proxy/meshy/openapi/v1/animations/{response.result}"),
+            response_model=MeshyAnimationResult,
+            status_extractor=lambda r: r.status,
+            progress_extractor=lambda r: r.progress,
+        )
+        model_file = f"meshy_model_{response.result}.glb"
+        await download_url_to_bytesio(result.result.animation_glb_url, os.path.join(get_output_directory(), model_file))
+        return IO.NodeOutput(model_file, response.result)
+
+
+class MeshyTextureNode(IO.ComfyNode):
+
+    @classmethod
+    def define_schema(cls):
+        return IO.Schema(
+            node_id="MeshyTextureNode",
+            display_name="Meshy: Texture Model",
+            category="api node/3d/Meshy",
+            inputs=[
+                IO.Combo.Input("model", options=["latest"]),
+                IO.Custom("MESHY_TASK_ID").Input("meshy_task_id"),
+                IO.Boolean.Input(
+                    "enable_original_uv",
+                    default=True,
+                    tooltip="Use the original UV of the model instead of generating new UVs. "
+                    "When enabled, Meshy preserves existing textures from the uploaded model. "
+                    "If the model has no original UV, the quality of the output might not be as good.",
+                ),
+                IO.Boolean.Input("pbr", default=False),
+                IO.String.Input(
+                    "text_style_prompt",
+                    default="",
+                    multiline=True,
+                    tooltip="Describe your desired texture style of the object using text. Maximum 600 characters."
+                    "Maximum 600 characters. Cannot be used at the same time as 'image_style'.",
+                ),
+                IO.Image.Input(
+                    "image_style",
+                    optional=True,
+                    tooltip="A 2d image to guide the texturing process. "
+                    "Can not be used at the same time with 'text_style_prompt'.",
+                ),
+            ],
+            outputs=[
+                IO.String.Output(display_name="model_file"),
+                IO.Custom("MODEL_TASK_ID").Output(display_name="meshy_task_id"),
+            ],
+            hidden=[
+                IO.Hidden.auth_token_comfy_org,
+                IO.Hidden.api_key_comfy_org,
+                IO.Hidden.unique_id,
+            ],
+            is_api_node=True,
+            is_output_node=True,
+            price_badge=IO.PriceBadge(
+                expr="""{"type":"usd","usd":0.4}""",
+            ),
+        )
+
+    @classmethod
+    async def execute(
+        cls,
+        model: str,
+        meshy_task_id: str,
+        enable_original_uv: bool,
+        pbr: bool,
+        text_style_prompt: str,
+        image_style: Input.Image | None = None,
+    ) -> IO.NodeOutput:
+        if text_style_prompt and image_style is not None:
+            raise ValueError("text_style_prompt and image_style cannot be used at the same time")
+        if not text_style_prompt and image_style is None:
+            raise ValueError("Either text_style_prompt or image_style is required")
+        image_style_url = None
+        if image_style is not None:
+            image_style_url = (await upload_images_to_comfyapi(cls, image_style, wait_label="Uploading style"))[0]
+        response = await sync_op(
+            cls,
+            endpoint=ApiEndpoint(path="/proxy/meshy/openapi/v1/retexture", method="POST"),
+            response_model=MeshyTaskResponse,
+            data=MeshyTextureRequest(
+                input_task_id=meshy_task_id,
+                ai_model=model,
+                enable_original_uv=enable_original_uv,
+                enable_pbr=pbr,
+                text_style_prompt=text_style_prompt if text_style_prompt else None,
+                image_style_url=image_style_url,
+            ),
+        )
+        result = await poll_op(
+            cls,
+            ApiEndpoint(path=f"/proxy/meshy/openapi/v1/retexture/{response.result}"),
+            response_model=MeshyModelResult,
+            status_extractor=lambda r: r.status,
+            progress_extractor=lambda r: r.progress,
+        )
+        model_file = f"meshy_model_{response.result}.glb"
+        await download_url_to_bytesio(result.model_urls.glb, os.path.join(get_output_directory(), model_file))
+        return IO.NodeOutput(model_file, response.result)
+
+
+class MeshyExtension(ComfyExtension):
+    @override
+    async def get_node_list(self) -> list[type[IO.ComfyNode]]:
+        return [
+            MeshyTextToModelNode,
+            MeshyRefineNode,
+            MeshyImageToModelNode,
+            MeshyMultiImageToModelNode,
+            MeshyRigModelNode,
+            MeshyAnimateModelNode,
+            MeshyTextureNode,
+        ]
+
+
+async def comfy_entrypoint() -> MeshyExtension:
+    return MeshyExtension()

comfy_api_nodes/util/upload_helpers.py

@@ -43,7 +43,7 @@ class UploadResponse(BaseModel):
 
 async def upload_images_to_comfyapi(
     cls: type[IO.ComfyNode],
-    image: torch.Tensor,
+    image: torch.Tensor | list[torch.Tensor],
     *,
     max_images: int = 8,
     mime_type: str | None = None,
@@ -55,15 +55,28 @@ async def upload_images_to_comfyapi(
     Uploads images to ComfyUI API and returns download URLs.
     To upload multiple images, stack them in the batch dimension first.
     """
+    tensors: list[torch.Tensor] = []
+    if isinstance(image, list):
+        for img in image:
+            is_batch = len(img.shape) > 3
+            if is_batch:
+                tensors.extend(img[i] for i in range(img.shape[0]))
+            else:
+                tensors.append(img)
+    else:
+        is_batch = len(image.shape) > 3
+        if is_batch:
+            tensors.extend(image[i] for i in range(image.shape[0]))
+        else:
+            tensors.append(image)
+
     # if batched, try to upload each file if max_images is greater than 0
     download_urls: list[str] = []
-    is_batch = len(image.shape) > 3
-    batch_len = image.shape[0] if is_batch else 1
-    num_to_upload = min(batch_len, max_images)
+    num_to_upload = min(len(tensors), max_images)
     batch_start_ts = time.monotonic()
 
     for idx in range(num_to_upload):
-        tensor = image[idx] if is_batch else image
+        tensor = tensors[idx]
         img_io = tensor_to_bytesio(tensor, total_pixels=total_pixels, mime_type=mime_type)
 
         effective_label = wait_label

comfyui_version.py

@@ -1,3 +1,3 @@
 # This file is automatically generated by the build process when version is
 # updated in pyproject.toml.
-__version__ = "0.9.1"
+__version__ = "0.9.2"

main.py

@@ -12,9 +12,6 @@ import itertools
 import utils.extra_config
 import logging
 import sys
-from comfy_execution.progress import get_progress_state
-from comfy_execution.utils import get_executing_context
-from comfy_api import feature_flags
 
 
 if __name__ == "__main__":
@@ -175,16 +172,22 @@ if 'torch' in sys.modules:
 
 import comfy.utils
 
-import execution
 import server
-from protocol import BinaryEventTypes
-import nodes
-import comfy.model_management
 import comfyui_version
 import app.logger
-import hook_breaker_ac10a0
+
+# Import modules needed for server operation
+# GPU initialization happens lazily when GPU functions are called
+# In subprocess mode, main process won't call GPU functions - workers will
+if __name__ == "__main__":
+    import execution
+    import nodes
+    import comfy.model_management
+
 
 def cuda_malloc_warning():
+    if args.use_subprocess_workers:
+        return
     device = comfy.model_management.get_torch_device()
     device_name = comfy.model_management.get_torch_device_name(device)
     cuda_malloc_warning = False
@@ -196,84 +199,6 @@ def cuda_malloc_warning():
             logging.warning("\nWARNING: this card most likely does not support cuda-malloc, if you get \"CUDA error\" please run ComfyUI with: --disable-cuda-malloc\n")
 
 
-def prompt_worker(q, server_instance):
-    current_time: float = 0.0
-    cache_type = execution.CacheType.CLASSIC
-    if args.cache_lru > 0:
-        cache_type = execution.CacheType.LRU
-    elif args.cache_ram > 0:
-        cache_type = execution.CacheType.RAM_PRESSURE
-    elif args.cache_none:
-        cache_type = execution.CacheType.NONE
-
-    e = execution.PromptExecutor(server_instance, cache_type=cache_type, cache_args={ "lru" : args.cache_lru, "ram" : args.cache_ram } )
-    last_gc_collect = 0
-    need_gc = False
-    gc_collect_interval = 10.0
-
-    while True:
-        timeout = 1000.0
-        if need_gc:
-            timeout = max(gc_collect_interval - (current_time - last_gc_collect), 0.0)
-
-        queue_item = q.get(timeout=timeout)
-        if queue_item is not None:
-            item, item_id = queue_item
-            execution_start_time = time.perf_counter()
-            prompt_id = item[1]
-            server_instance.last_prompt_id = prompt_id
-
-            sensitive = item[5]
-            extra_data = item[3].copy()
-            for k in sensitive:
-                extra_data[k] = sensitive[k]
-
-            e.execute(item[2], prompt_id, extra_data, item[4])
-            need_gc = True
-
-            remove_sensitive = lambda prompt: prompt[:5] + prompt[6:]
-            q.task_done(item_id,
-                        e.history_result,
-                        status=execution.PromptQueue.ExecutionStatus(
-                            status_str='success' if e.success else 'error',
-                            completed=e.success,
-                            messages=e.status_messages), process_item=remove_sensitive)
-            if server_instance.client_id is not None:
-                server_instance.send_sync("executing", {"node": None, "prompt_id": prompt_id}, server_instance.client_id)
-
-            current_time = time.perf_counter()
-            execution_time = current_time - execution_start_time
-
-            # Log Time in a more readable way after 10 minutes
-            if execution_time > 600:
-                execution_time = time.strftime("%H:%M:%S", time.gmtime(execution_time))
-                logging.info(f"Prompt executed in {execution_time}")
-            else:
-                logging.info("Prompt executed in {:.2f} seconds".format(execution_time))
-
-        flags = q.get_flags()
-        free_memory = flags.get("free_memory", False)
-
-        if flags.get("unload_models", free_memory):
-            comfy.model_management.unload_all_models()
-            need_gc = True
-            last_gc_collect = 0
-
-        if free_memory:
-            e.reset()
-            need_gc = True
-            last_gc_collect = 0
-
-        if need_gc:
-            current_time = time.perf_counter()
-            if (current_time - last_gc_collect) > gc_collect_interval:
-                gc.collect()
-                comfy.model_management.soft_empty_cache()
-                last_gc_collect = current_time
-                need_gc = False
-                hook_breaker_ac10a0.restore_functions()
-
-
 async def run(server_instance, address='', port=8188, verbose=True, call_on_start=None):
     addresses = []
     for addr in address.split(","):
@@ -282,37 +207,6 @@ async def run(server_instance, address='', port=8188, verbose=True, call_on_star
         server_instance.start_multi_address(addresses, call_on_start, verbose), server_instance.publish_loop()
     )
 
-def hijack_progress(server_instance):
-    def hook(value, total, preview_image, prompt_id=None, node_id=None):
-        executing_context = get_executing_context()
-        if prompt_id is None and executing_context is not None:
-            prompt_id = executing_context.prompt_id
-        if node_id is None and executing_context is not None:
-            node_id = executing_context.node_id
-        comfy.model_management.throw_exception_if_processing_interrupted()
-        if prompt_id is None:
-            prompt_id = server_instance.last_prompt_id
-        if node_id is None:
-            node_id = server_instance.last_node_id
-        progress = {"value": value, "max": total, "prompt_id": prompt_id, "node": node_id}
-        get_progress_state().update_progress(node_id, value, total, preview_image)
-
-        server_instance.send_sync("progress", progress, server_instance.client_id)
-        if preview_image is not None:
-            # Only send old method if client doesn't support preview metadata
-            if not feature_flags.supports_feature(
-                server_instance.sockets_metadata,
-                server_instance.client_id,
-                "supports_preview_metadata",
-            ):
-                server_instance.send_sync(
-                    BinaryEventTypes.UNENCODED_PREVIEW_IMAGE,
-                    preview_image,
-                    server_instance.client_id,
-                )
-
-    comfy.utils.set_progress_bar_global_hook(hook)
-
 
 def cleanup_temp():
     temp_dir = folder_paths.get_temp_directory()
@@ -357,20 +251,16 @@ def start_comfyui(asyncio_loop=None):
     if args.enable_manager and not args.disable_manager_ui:
         comfyui_manager.start()
 
-    hook_breaker_ac10a0.save_functions()
-    asyncio_loop.run_until_complete(nodes.init_extra_nodes(
-        init_custom_nodes=(not args.disable_all_custom_nodes) or len(args.whitelist_custom_nodes) > 0,
-        init_api_nodes=not args.disable_api_nodes
-    ))
-    hook_breaker_ac10a0.restore_functions()
+    from comfy.execution_core import create_worker, prompt_worker
+    worker = create_worker(prompt_server)
+    node_count = asyncio_loop.run_until_complete(worker.initialize())
+    logging.info(f"Loaded {node_count} node types")
+    threading.Thread(target=prompt_worker, daemon=True, args=(prompt_server.prompt_queue, worker), name="PromptWorker").start()
 
     cuda_malloc_warning()
     setup_database()
 
     prompt_server.add_routes()
-    hijack_progress(prompt_server)
-
-    threading.Thread(target=prompt_worker, daemon=True, args=(prompt_server.prompt_queue, prompt_server,)).start()
 
     if args.quick_test_for_ci:
         exit(0)

nodes.py

@@ -50,7 +50,8 @@ def before_node_execution():
     comfy.model_management.throw_exception_if_processing_interrupted()
 
 def interrupt_processing(value=True):
-    comfy.model_management.interrupt_current_processing(value)
+    from comfy.execution_core import interrupt_processing as core_interrupt
+    core_interrupt(value)
 
 MAX_RESOLUTION=16384
 
@@ -788,6 +789,7 @@ class VAELoader:
 
     #TODO: scale factor?
     def load_vae(self, vae_name):
+        metadata = None
         if vae_name == "pixel_space":
             sd = {}
             sd["pixel_space_vae"] = torch.tensor(1.0)
@@ -2400,6 +2402,7 @@ async def init_builtin_api_nodes():
         "nodes_sora.py",
         "nodes_topaz.py",
         "nodes_tripo.py",
+        "nodes_meshy.py",
         "nodes_moonvalley.py",
         "nodes_rodin.py",
         "nodes_gemini.py",

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "ComfyUI"
-version = "0.9.1"
+version = "0.9.2"
 readme = "README.md"
 license = { file = "LICENSE" }
 requires-python = ">=3.10"