fix: keep local LTX tiling in execution context

Fixes #85

.github/FUNDING.yml

@@ -1,4 +1,3 @@
 # These are supported funding model platforms
 
 github: robertvoy
-buy_me_a_coffee: robertvoy

.github/workflows/publish_action.yml

@@ -19,6 +19,6 @@ jobs:
       - name: Check out code
         uses: actions/checkout@v4
       - name: Publish Custom Node
-        uses: Comfy-Org/publish-node-action@v1
+        uses: Comfy-Org/publish-node-action@main
         with:
           personal_access_token: ${{ secrets.REGISTRY_ACCESS_TOKEN }}

README.md

@@ -15,7 +15,7 @@
 
 ---
 
-## Key Features
+## Key Features
 
 #### Parallel Workflow Processing
 - Run your workflow on multiple GPUs simultaneously with varied seeds, collect results on the master
@@ -27,22 +27,12 @@
 - Intelligent distribution
 - Handles single images and videos
 
-#### Ease of Use
-- Auto-setup local workers; easily add remote/cloud ones
-- Convert any workflow to distributed with 2 nodes
-- JSON configuration with UI controls
-
----
-
-## Current Architecture
-
-- Workflow-level load balancing is controlled by **Distributed Collector** via the `load_balance` toggle.
-- There is **no Distributed Queue node** anymore.
-- With `load_balance=true`, orchestration selects one least-busy execution participant:
-  - If master participation is enabled, master is included as a candidate.
-  - If master is in orchestrator-only mode, only workers are considered.
-
----
+#### Ease of Use
+- Auto-setup local workers; easily add remote/cloud ones
+- Convert any workflow to distributed with 2 nodes
+- JSON configuration with UI controls
+
+---
 
 ## Worker Types
 
@@ -61,7 +51,6 @@ ComfyUI Distributed supports three types of workers:
 ## Requirements
 
 - ComfyUI
-> Note: Desktop app not currently supported
 - Multiple NVIDIA GPUs
 > No additional GPUs? Use [Cloud Workers](https://github.com/robertvoy/ComfyUI-Distributed/blob/main/docs/worker-setup-guides.md#cloud-workers)
 - That's it
@@ -92,19 +81,19 @@ Join Runpod with [this link](https://get.runpod.io/0bw29uf3ug0p) and unlock a sp
 
 ## Workflow Examples
 
-### Basic Parallel Generation
-Generate multiple images in the time it takes to generate one. Each worker uses a different seed.
+### Basic Parallel Generation
+Generate multiple images in the time it takes to generate one. Each worker uses a different seed.
 
 ![Clipboard Image (6)](https://github.com/user-attachments/assets/9598c94c-d9b4-4ccf-ab16-a21398220aeb)
 
 > [Download workflow](/workflows/distributed-txt2img.json)
 
 1. Open your ComfyUI workflow
-2. Add **Distributed Seed** → connect to sampler's seed
-3. Add **Distributed Collector** → after VAE Decode
-4. Optional: enable `load_balance` on Distributed Collector to run on one least-busy participant
-5. Enable workers in the UI
-6. Run the workflow!
+2. Add **Distributed Seed** → connect to sampler's seed
+3. Add **Distributed Collector** → after VAE Decode
+4. Optional: enable `load_balance` on Distributed Collector to run on one least-busy participant
+5. Enable workers in the UI
+6. Run the workflow!
 
 ### Parallel WAN Generation
 Generate multiple videos in the time it takes to generate one. Each worker uses a different seed.
@@ -122,12 +111,12 @@ Generate multiple videos in the time it takes to generate one. Each worker uses
 7. Enable workers in the UI
 8. Run the workflow!
 
-### Distributed Image Upscaling
-Accelerate Ultimate SD Upscaler by distributing tiles across multiple workers, with speed scaling as you add more GPUs.
+### Distributed Image Upscaling
+Accelerate Ultimate SD Upscaler by distributing tiles across multiple workers, with speed scaling as you add more GPUs.
 
 ![Clipboard Image (3)](https://github.com/user-attachments/assets/ffb57a0d-7b75-4497-96d2-875d60865a1a)
 
-> [Download workflow](/workflows/distributed-upscale.json)
+> [Download workflow](/workflows/distributed-upscale.json)
 
 1. Load your image
 2. Upscale with ESRGAN or similar
@@ -153,7 +142,7 @@ Accelerate Ultimate SD Upscaler by distributing video tiles across multiple work
 
 ---
 
-## Developer API
+## Developer API
 
 Control your distributed cluster programmatically without opening the browser.
 
@@ -161,32 +150,32 @@ Control your distributed cluster programmatically without opening the browser.
 * **Functionality:** Accepts a standard ComfyUI workflow JSON, automatically distributes it to available workers, and returns the execution ID.
 * **Documentation:** [See API Examples & Scripts](https://github.com/robertvoy/ComfyUI-Distributed/blob/main/docs/comfyui-distributed-api.md)
 
-> **⚠️ Security Warning:** Do not expose your ComfyUI port to the public internet. If you need remote access, run ComfyUI behind a secure proxy (like Cloudflare or a VPN).
-
----
-
-## Distributed Value
-
-Use **Distributed Value** when you want per-worker overrides (for example, different prompts/models/settings per worker).
-
-- Output type adapts to the connected input where possible (`STRING`, `INT`, `FLOAT`, `COMBO`).
-- The node shows only currently enabled workers.
-- If worker enablement changes, worker fields update automatically.
-- When disconnected, it resets to default string mode and clears per-worker overrides.
-- On execution, master uses `default_value`; workers use their mapped override with typed coercion fallback to default.
-
----
-
-## Nodes
-
-| Node | Description |
-|------|-------------|
-| **Distributed Seed** | Generates unique seeds for each worker |
-| **Distributed Collector** | Collects results (image/video frames and optionally audio) from workers back to the master; `load_balance` can route the run to one least-busy participant |
-| **Distributed Value** | Outputs per-worker override values with fallback to default |
-| **Ultimate SD Upscale Distributed** | Distributes upscale tiles across workers |
-| **Image Batch Divider** | Splits image batches for multi-GPU output |
-| **Audio Batch Divider** | Splits audio batches for multi-GPU output |
+> **⚠️ Security Warning:** Do not expose your ComfyUI port to the public internet. If you need remote access, run ComfyUI behind a secure proxy (like Cloudflare or a VPN).
+
+---
+
+## Distributed Value
+
+Use **Distributed Value** when you want per-worker overrides (for example, different prompts/models/settings per worker).
+
+- Output type adapts to the connected input where possible (`STRING`, `INT`, `FLOAT`, `COMBO`).
+- The node shows only currently enabled workers.
+- If worker enablement changes, worker fields update automatically.
+- When disconnected, it resets to default string mode and clears per-worker overrides.
+- On execution, master uses `default_value`; workers use their mapped override with typed coercion fallback to default.
+
+---
+
+## Nodes
+
+| Node | Description |
+|------|-------------|
+| **Distributed Seed** | Generates unique seeds for each worker |
+| **Distributed Collector** | Collects results (image/video frames and optionally audio) from workers back to the master; `load_balance` can route the run to one least-busy participant |
+| **Distributed Value** | Outputs per-worker override values with fallback to default |
+| **Ultimate SD Upscale Distributed** | Distributes upscale tiles across workers |
+| **Image Batch Divider** | Splits image batches for multi-GPU output |
+| **Audio Batch Divider** | Splits audio batches for multi-GPU output |
 | **Distributed Model Name** | Passes model paths to workers, enabling workflows to use models not present on the master in orchestrator-only mode |
 | **Distributed Empty Image** | Produces an empty IMAGE batch used when the master delegates all work |
 
@@ -206,7 +195,7 @@ No, it does not speed up the generation of a single image or video. Instead, it
 
 <details>
 <summary>Does it work with the ComfyUI desktop app?</summary>
-Currently, it is not compatible with the ComfyUI desktop app.
+Yes, it does now.
 </details>
 
 <details>
@@ -249,3 +238,4 @@ Buy me a coffee at: https://buymeacoffee.com/robertvoy
 
 
 
+

api/__init__.py

@@ -3,3 +3,4 @@ from . import tunnel_routes  # noqa: F401
 from . import worker_routes  # noqa: F401
 from . import job_routes  # noqa: F401
 from . import usdu_routes  # noqa: F401
+from . import ltx_routes  # noqa: F401

api/job_routes.py

@@ -217,7 +217,7 @@ async def distributed_queue_endpoint(request):
         return await handle_api_error(request, exc, 400)
 
     try:
-        prompt_id, worker_count = await orchestrate_distributed_execution(
+        prompt_id, prompt_number, worker_count, node_errors = await orchestrate_distributed_execution(
             payload.prompt,
             payload.workflow_meta,
             payload.client_id,
@@ -227,6 +227,8 @@ async def distributed_queue_endpoint(request):
         )
         return web.json_response({
             "prompt_id": prompt_id,
+            "number": prompt_number,
+            "node_errors": node_errors,
             "worker_count": worker_count,
             "auto_prepare_supported": True,
         })

api/ltx_routes.py

@@ -0,0 +1,76 @@
+"""HTTP routes for distributed LTX latent tile results."""
+
+from __future__ import annotations
+
+import time
+
+from aiohttp import web
+import server
+
+from ..upscale.job_models import TileJobState
+from ..upscale.job_store import MAX_PAYLOAD_SIZE, ensure_tile_jobs_initialized
+from ..utils.logging import debug_log
+from ..utils.network import handle_api_error
+
+
+@server.PromptServer.instance.routes.post("/distributed/ltx/submit_tile")
+async def submit_ltx_tile_endpoint(request):
+    """Accept one processed LTX latent tile from a worker.
+
+    Body is JSON:
+    - multi_job_id: distributed job id
+    - worker_id: worker id
+    - tile_index: spatial tile id, or -1 for completion-only messages
+    - payload: base64 safetensors payload produced by nodes.ltx_tiled_sampler
+    - is_last: optional worker completion flag
+    """
+    try:
+        content_length = request.headers.get("content-length")
+        if content_length and int(content_length) > MAX_PAYLOAD_SIZE:
+            return await handle_api_error(request, f"Payload too large: {content_length} bytes", 413)
+
+        data = await request.json()
+        multi_job_id = data.get("multi_job_id")
+        worker_id = str(data.get("worker_id") or "")
+        is_last = bool(data.get("is_last", False))
+        payload = data.get("payload")
+        tile_index = data.get("tile_index")
+
+        if not multi_job_id or not worker_id:
+            return await handle_api_error(request, "Missing multi_job_id or worker_id", 400)
+
+        prompt_server = ensure_tile_jobs_initialized()
+        async with prompt_server.distributed_tile_jobs_lock:
+            job_data = prompt_server.distributed_pending_tile_jobs.get(multi_job_id)
+            if not isinstance(job_data, TileJobState):
+                return await handle_api_error(request, "Job not configured for tile submissions", 404)
+
+            job_data.worker_status[worker_id] = time.time()
+
+            if is_last and (payload is None or int(tile_index or -1) < 0):
+                await job_data.queue.put({
+                    "worker_id": worker_id,
+                    "is_last": True,
+                    "tiles": [],
+                })
+                debug_log(f"Received LTX completion signal from worker {worker_id}")
+                return web.json_response({"status": "success"})
+
+            if payload is None or tile_index is None:
+                return await handle_api_error(request, "Missing tile_index or payload", 400)
+
+            tile_index = int(tile_index)
+            await job_data.queue.put({
+                "worker_id": worker_id,
+                "is_last": is_last,
+                "tiles": [{
+                    "tile_idx": tile_index,
+                    "global_idx": tile_index,
+                    "payload": payload,
+                    "worker_id": worker_id,
+                }],
+            })
+            debug_log(f"Received LTX tile {tile_index} from worker {worker_id} (is_last={is_last})")
+            return web.json_response({"status": "success"})
+    except Exception as exc:
+        return await handle_api_error(request, exc, 500)

api/orchestration/prompt_transform.py

@@ -125,6 +125,209 @@ def _find_upstream_nodes(prompt_obj, start_ids):
     return connected
 
 
+_DELEGATE_MASTER_RETAINED_UPSTREAM_CLASSES = {
+    "PrimitiveBoolean",
+    "PrimitiveFloat",
+    "PrimitiveInt",
+    "PrimitiveNode",
+    "PrimitiveString",
+}
+
+_DELEGATE_MASTER_ALWAYS_RETAINED_UPSTREAM_CLASSES = {
+    "LoadImage",
+}
+
+_DELEGATE_MASTER_SAFE_SCALAR_TYPES = {"BOOLEAN", "FLOAT", "INT", "STRING"}
+_DELEGATE_MASTER_SAFE_LIST_TYPES = {"LIST"}
+
+# ComfyUI 0.23 exposes CreateList via the newer schema API rather than the
+# legacy RETURN_TYPES/INPUT_TYPES attributes. Treat it as a safe config utility
+# only after its connected inputs recursively prove safe.
+_DELEGATE_MASTER_SAFE_DYNAMIC_CONFIG_OUTPUT_CLASSES = {"CreateList"}
+_DELEGATE_MASTER_SAFE_DYNAMIC_CONFIG_INPUT_PREFIXES = {
+    "CreateList": ("inputs.",),
+}
+
+# Test hook. At runtime this stays None and the ComfyUI node registry is loaded lazily.
+_DELEGATE_MASTER_NODE_CLASS_MAPPINGS = None
+
+
+def _get_delegate_master_node_class_mappings():
+    """Return ComfyUI node-class mappings when available."""
+    if _DELEGATE_MASTER_NODE_CLASS_MAPPINGS is not None:
+        return _DELEGATE_MASTER_NODE_CLASS_MAPPINGS
+    try:
+        import nodes as comfy_nodes  # type: ignore
+    except Exception:  # pragma: no cover - depends on ComfyUI runtime imports
+        return {}
+    return getattr(comfy_nodes, "NODE_CLASS_MAPPINGS", {}) or {}
+
+
+def _get_delegate_master_node_class(class_type):
+    mappings = _get_delegate_master_node_class_mappings()
+    return mappings.get(class_type) if isinstance(mappings, dict) else None
+
+
+def _normalize_delegate_master_return_type(return_type):
+    if return_type is None:
+        return ""
+    return str(return_type).strip().upper()
+
+
+def _delegate_master_type_is_safe_scalar(type_name):
+    return type_name in _DELEGATE_MASTER_SAFE_SCALAR_TYPES
+
+
+def _delegate_master_type_is_safe_config(type_name):
+    return _delegate_master_type_is_safe_scalar(type_name) or type_name in _DELEGATE_MASTER_SAFE_LIST_TYPES
+
+
+def _delegate_master_output_is_safe_scalar(class_type, output_index):
+    """Return True when a registered node output is lightweight config data."""
+    if class_type in _DELEGATE_MASTER_SAFE_DYNAMIC_CONFIG_OUTPUT_CLASSES:
+        return True
+    node_class = _get_delegate_master_node_class(class_type)
+    return_types = getattr(node_class, "RETURN_TYPES", ()) if node_class is not None else ()
+    try:
+        output_type = return_types[int(output_index)]
+    except (IndexError, TypeError, ValueError):
+        return False
+    return _delegate_master_type_is_safe_config(_normalize_delegate_master_return_type(output_type))
+
+
+def _get_delegate_master_input_types(class_type):
+    node_class = _get_delegate_master_node_class(class_type)
+    input_types = getattr(node_class, "INPUT_TYPES", None) if node_class is not None else None
+    if callable(input_types):
+        try:
+            input_types = input_types()
+        except TypeError:
+            return {}
+    return input_types if isinstance(input_types, dict) else {}
+
+
+def _normalize_delegate_master_input_type(input_spec):
+    if isinstance(input_spec, (list, tuple)) and input_spec:
+        return _normalize_delegate_master_return_type(input_spec[0])
+    return _normalize_delegate_master_return_type(input_spec)
+
+
+def _delegate_master_input_is_safe_scalar(class_type, input_name):
+    """Return True when a registered downstream input expects config data."""
+    for prefix in _DELEGATE_MASTER_SAFE_DYNAMIC_CONFIG_INPUT_PREFIXES.get(class_type, ()):
+        if input_name.startswith(prefix):
+            return True
+    input_types = _get_delegate_master_input_types(class_type)
+    for section_name in ("required", "optional"):
+        section = input_types.get(section_name, {})
+        if isinstance(section, dict) and input_name in section:
+            input_type = _normalize_delegate_master_input_type(section[input_name])
+            return _delegate_master_type_is_safe_config(input_type)
+    return False
+
+
+def _is_delegate_master_always_retained_upstream_node(node):
+    if not isinstance(node, dict):
+        return False
+    class_type = node.get("class_type")
+    return isinstance(class_type, str) and class_type in _DELEGATE_MASTER_ALWAYS_RETAINED_UPSTREAM_CLASSES
+
+
+def _is_delegate_master_retained_upstream_node(node, output_index=0):
+    """Return True for lightweight upstream nodes safe to keep on the master."""
+    if not isinstance(node, dict):
+        return False
+    class_type = node.get("class_type")
+    if not isinstance(class_type, str):
+        return False
+    return (
+        class_type in _DELEGATE_MASTER_RETAINED_UPSTREAM_CLASSES
+        or class_type.startswith("Primitive")
+        or _delegate_master_output_is_safe_scalar(class_type, output_index)
+    )
+
+
+def _collect_delegate_master_retained_upstream_branch(
+    prompt_obj,
+    node_id,
+    output_index,
+    memo,
+    visiting,
+):
+    """Return safe retained branch nodes, or None when the branch is not safe."""
+    node_id = str(node_id)
+    cache_key = (node_id, output_index)
+    if cache_key in memo:
+        cached = memo[cache_key]
+        return None if cached is None else set(cached)
+    if cache_key in visiting:
+        memo[cache_key] = None
+        return None
+
+    node = prompt_obj.get(node_id)
+    if not _is_delegate_master_retained_upstream_node(node, output_index):
+        memo[cache_key] = None
+        return None
+
+    visiting.add(cache_key)
+    retained = {node_id}
+    inputs = node.get("inputs", {}) if isinstance(node, dict) else {}
+    class_type = node.get("class_type") if isinstance(node, dict) else None
+    for input_name, value in inputs.items():
+        if not (isinstance(value, list) and len(value) == 2):
+            continue
+        if not _delegate_master_input_is_safe_scalar(class_type, input_name):
+            visiting.remove(cache_key)
+            memo[cache_key] = None
+            return None
+        source_id = str(value[0])
+        branch = _collect_delegate_master_retained_upstream_branch(
+            prompt_obj,
+            source_id,
+            value[1],
+            memo,
+            visiting,
+        )
+        if branch is None:
+            visiting.remove(cache_key)
+            memo[cache_key] = None
+            return None
+        retained.update(branch)
+
+    visiting.remove(cache_key)
+    memo[cache_key] = frozenset(retained)
+    return retained
+
+
+def _find_delegate_master_retained_upstream_nodes(prompt_obj, start_ids):
+    """Return lightweight upstream nodes needed by kept delegate-master nodes."""
+    connected = set()
+    memo = {}
+    for node_id in start_ids:
+        node = prompt_obj.get(str(node_id)) or {}
+        inputs = node.get("inputs", {})
+        class_type = node.get("class_type") if isinstance(node, dict) else None
+        for input_name, value in inputs.items():
+            if not (isinstance(value, list) and len(value) == 2):
+                continue
+            source_node = prompt_obj.get(str(value[0]))
+            if _is_delegate_master_always_retained_upstream_node(source_node):
+                connected.add(str(value[0]))
+                continue
+            if not _delegate_master_input_is_safe_scalar(class_type, input_name):
+                continue
+            branch = _collect_delegate_master_retained_upstream_branch(
+                prompt_obj,
+                value[0],
+                value[1],
+                memo,
+                set(),
+            )
+            if branch is not None:
+                connected.update(branch)
+    return connected
+
+
 def prune_prompt_for_worker(prompt_obj):
     """Prune worker prompt to distributed nodes and their upstream dependencies."""
     collector_ids = find_nodes_by_class(prompt_obj, "DistributedCollector")
@@ -167,6 +370,9 @@ def prepare_delegate_master_prompt(prompt_obj, collector_ids):
     downstream = _find_downstream_nodes(prompt_obj, collector_ids)
     nodes_to_keep = set(collector_ids)
     nodes_to_keep.update(downstream)
+    nodes_to_keep.update(
+        _find_delegate_master_retained_upstream_nodes(prompt_obj, nodes_to_keep)
+    )
 
     pruned_prompt = {}
     for node_id in nodes_to_keep:

api/queue_orchestration.py

@@ -13,7 +13,7 @@ from ..utils.constants import (
     ORCHESTRATION_WORKER_PREP_CONCURRENCY,
 )
 from ..utils.logging import debug_log, log
-from ..utils.network import build_master_url
+from ..utils.network import build_master_url, build_master_callback_url
 from ..utils.trace_logger import trace_debug
 from .schemas import parse_positive_float, parse_positive_int
 from .orchestration.dispatch import (
@@ -144,6 +144,7 @@ async def _prepare_worker_payload(
     enabled_ids,
     job_id_map,
     master_url,
+    config,
     delegate_master,
     trace_execution_id,
     worker_prep_semaphore,
@@ -153,6 +154,11 @@ async def _prepare_worker_payload(
     """Prepare one worker prompt payload with bounded concurrency and media-sync timeout."""
     async with worker_prep_semaphore:
         worker_prompt = prompt_index.copy_prompt()
+        worker_master_url = build_master_callback_url(
+            worker,
+            config=config,
+            prompt_server_instance=prompt_server,
+        )
 
         worker_type = str(worker.get("type") or "local").strip().lower()
         is_remote_like = bool(worker.get("host")) and worker_type != "local"
@@ -167,7 +173,7 @@ async def _prepare_worker_payload(
             worker["id"],
             enabled_ids,
             job_id_map,
-            master_url,
+            worker_master_url,
             delegate_master,
             prompt_index,
         )
@@ -202,7 +208,7 @@ async def orchestrate_distributed_execution(
     """Core orchestration logic for the /distributed/queue endpoint.
 
     Returns:
-        tuple[str, int]: (prompt_id, worker_count)
+        tuple[str, int, int, dict]: (prompt_id, number, worker_count, node_errors)
     """
     ensure_distributed_state()
     execution_trace_id = trace_execution_id or _generate_execution_trace_id()
@@ -311,8 +317,18 @@ async def orchestrate_distributed_execution(
 
     if not job_id_map:
         trace_debug(execution_trace_id, "No distributed nodes detected; queueing prompt on master only.")
-        prompt_id = await queue_prompt_payload(prompt_obj, workflow_meta, client_id)
-        return prompt_id, 0
+        queue_result = await queue_prompt_payload(
+            prompt_obj,
+            workflow_meta,
+            client_id,
+            include_queue_metadata=True,
+        )
+        return (
+            queue_result["prompt_id"],
+            queue_result["number"],
+            0,
+            queue_result.get("node_errors", {}),
+        )
 
     for job_id in job_id_map.values():
         await _ensure_distributed_queue(job_id)
@@ -360,6 +376,7 @@ async def orchestrate_distributed_execution(
                     enabled_ids,
                     job_id_map,
                     master_url,
+                    config,
                     delegate_master,
                     execution_trace_id,
                     worker_prep_semaphore,
@@ -385,9 +402,17 @@ async def orchestrate_distributed_execution(
             ]
         )
 
-    prompt_id = await queue_prompt_payload(master_prompt, workflow_meta, client_id)
+    queue_result = await queue_prompt_payload(
+        master_prompt,
+        workflow_meta,
+        client_id,
+        include_queue_metadata=True,
+    )
+    prompt_id = queue_result["prompt_id"]
+    prompt_number = queue_result["number"]
+    node_errors = queue_result.get("node_errors", {})
     trace_debug(
         execution_trace_id,
         f"Orchestration complete: prompt_id={prompt_id}, dispatched_workers={len(worker_payloads)}, delegate_master={delegate_master}",
     )
-    return prompt_id, len(worker_payloads)
+    return prompt_id, prompt_number, len(worker_payloads), node_errors

nodes/__init__.py

@@ -10,6 +10,10 @@ from .utilities import (
     any_type,
 )
 from .collector import DistributedCollectorNode
+from .ltx_tiled_sampler import (
+    NODE_CLASS_MAPPINGS as LTX_CLASS_MAPPINGS,
+    NODE_DISPLAY_NAME_MAPPINGS as LTX_DISPLAY_NAME_MAPPINGS,
+)
 
 NODE_CLASS_MAPPINGS = {
     "DistributedCollector": DistributedCollectorNode,
@@ -19,6 +23,7 @@ NODE_CLASS_MAPPINGS = {
     "ImageBatchDivider": ImageBatchDivider,
     "AudioBatchDivider": AudioBatchDivider,
     "DistributedEmptyImage": DistributedEmptyImage,
+    **LTX_CLASS_MAPPINGS,
 }
 NODE_DISPLAY_NAME_MAPPINGS = {
     "DistributedCollector": "Distributed Collector",
@@ -28,4 +33,5 @@ NODE_DISPLAY_NAME_MAPPINGS = {
     "ImageBatchDivider": "Image Batch Divider",
     "AudioBatchDivider": "Audio Batch Divider",
     "DistributedEmptyImage": "Distributed Empty Image",
+    **LTX_DISPLAY_NAME_MAPPINGS,
 }

nodes/collector.py

@@ -22,6 +22,7 @@ prompt_server = _server.PromptServer.instance
 
 
 class DistributedCollectorNode:
+    INPUT_IS_LIST = True
     EMPTY_AUDIO = {"waveform": torch.zeros(1, 2, 1), "sample_rate": 44100}
 
     @classmethod
@@ -55,7 +56,65 @@ class DistributedCollectorNode:
     FUNCTION = "run"
     CATEGORY = "image"
     
+    @staticmethod
+    def _unwrap_list_input(value):
+        """Unwrap scalar inputs when ComfyUI passes them via INPUT_IS_LIST."""
+        if isinstance(value, (list, tuple)) and len(value) == 1:
+            return value[0]
+        return value
+
+    def _normalize_images_input(self, images):
+        """Collapse ComfyUI list IMAGE inputs into a normal batched IMAGE tensor."""
+        if isinstance(images, (list, tuple)):
+            if not images:
+                raise ValueError("Collector received an empty image list")
+            if not all(isinstance(image, torch.Tensor) for image in images):
+                raise TypeError("Collector expected IMAGE list items to be torch.Tensor instances")
+            if len(images) == 1:
+                return ensure_contiguous(images[0])
+            return ensure_contiguous(torch.cat([ensure_contiguous(image) for image in images], dim=0))
+        return ensure_contiguous(images)
+
+    def _normalize_audio_input(self, audio):
+        """Collapse ComfyUI list AUDIO inputs into a single AUDIO payload when present."""
+        if not isinstance(audio, (list, tuple)):
+            return audio
+
+        audio_items = [item for item in audio if item is not None]
+        if not audio_items:
+            return None
+        if len(audio_items) == 1:
+            return audio_items[0]
+
+        waveforms = []
+        sample_rate = 44100
+        for item in audio_items:
+            if not isinstance(item, dict):
+                raise TypeError("Collector expected AUDIO list items to be dictionaries")
+            waveform = item.get("waveform")
+            if waveform is None or waveform.numel() == 0:
+                continue
+            waveforms.append(waveform)
+            if sample_rate == 44100:
+                sample_rate = item.get("sample_rate", 44100)
+
+        if not waveforms:
+            return None
+        return {"waveform": torch.cat(waveforms, dim=-1), "sample_rate": sample_rate}
+
     def run(self, images, load_balance=False, audio=None, multi_job_id="", is_worker=False, master_url="", enabled_worker_ids="[]", worker_batch_size=1, worker_id="", pass_through=False, delegate_only=False):
+        images = self._normalize_images_input(images)
+        audio = self._normalize_audio_input(audio)
+        load_balance = self._unwrap_list_input(load_balance)
+        multi_job_id = self._unwrap_list_input(multi_job_id)
+        is_worker = self._unwrap_list_input(is_worker)
+        master_url = self._unwrap_list_input(master_url)
+        enabled_worker_ids = self._unwrap_list_input(enabled_worker_ids)
+        worker_batch_size = self._unwrap_list_input(worker_batch_size)
+        worker_id = self._unwrap_list_input(worker_id)
+        pass_through = self._unwrap_list_input(pass_through)
+        delegate_only = self._unwrap_list_input(delegate_only)
+
         # Create empty audio if not provided
         empty_audio = {"waveform": torch.zeros(1, 2, 1), "sample_rate": 44100}
 

nodes/ltx_tiled_sampler.py

@@ -0,0 +1,668 @@
+"""Distributed LTX tiled sampler node.
+
+This node adapts the upstream LTX tiled sampler interface to the existing
+ComfyUI-Distributed job queue. It distributes spatial latent tiles across the
+same worker/master infrastructure used by the distributed upscaler instead of
+creating a parallel queue implementation.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import base64
+import copy
+import math
+from typing import Callable, Dict, Iterable, List, Optional, Tuple
+
+import aiohttp
+import torch
+from safetensors.torch import load as safetensors_load
+from safetensors.torch import save as safetensors_save
+
+from ..upscale.job_models import TileJobState
+from ..upscale.job_store import (
+    _cleanup_job,
+    _drain_results_queue,
+    _get_completed_count,
+    _mark_task_completed,
+    ensure_tile_jobs_initialized,
+    init_static_job_batched,
+)
+from ..upscale.job_timeout import _check_and_requeue_timed_out_workers
+from ..utils.async_helpers import run_async_in_server_loop
+from ..utils.constants import HEARTBEAT_INTERVAL, TILE_WAIT_TIMEOUT
+from ..utils.logging import debug_log, log
+from ..utils.network import get_client_session
+from ..utils.usdu_managment import _send_heartbeat_to_master
+
+try:  # Optional external LTX tiled sampler dependency.
+    from latent_tiled_sampler import LTXTiledSampler  # type: ignore
+except Exception:  # pragma: no cover - depends on optional custom node install.
+    LTXTiledSampler = None  # type: ignore
+
+
+def _default_ltx_inputs() -> dict:
+    """Fallback schema used when the source LTX sampler is not importable."""
+    return {
+        "required": {
+            "noise": ("NOISE",),
+            "guider": ("GUIDER",),
+            "sampler": ("SAMPLER",),
+            "sigmas": ("SIGMAS",),
+            "latent_image": ("LATENT",),
+            "h_tiles": ("INT", {"default": 1, "min": 1, "max": 64}),
+            "w_tiles": ("INT", {"default": 1, "min": 1, "max": 64}),
+            "overlap": ("INT", {"default": 0, "min": 0, "max": 4096}),
+        },
+        "optional": {
+            "denoise_mask": ("MASK",),
+        },
+    }
+
+
+def _copy_base_inputs() -> dict:
+    if LTXTiledSampler is None:
+        return _default_ltx_inputs()
+    try:
+        return copy.deepcopy(LTXTiledSampler.INPUT_TYPES())
+    except Exception:
+        return _default_ltx_inputs()
+
+
+def _compute_tile_starts(total_size: int, n_tiles: int, overlap: int) -> Tuple[List[int], int]:
+    """Compute evenly spaced tile starts and tile size for one dimension."""
+    total_size = int(total_size)
+    n_tiles = max(1, int(n_tiles))
+    overlap = max(0, int(overlap))
+    if n_tiles <= 1 or total_size <= 1:
+        return [0], total_size
+    tile_size = math.ceil((total_size + (n_tiles - 1) * overlap) / n_tiles)
+    tile_size = min(tile_size, total_size)
+    if n_tiles == 2:
+        return [0, max(0, total_size - tile_size)], tile_size
+    stride = (total_size - tile_size) / (n_tiles - 1)
+    return [int(round(i * stride)) for i in range(n_tiles)], tile_size
+
+
+def _make_window_1d(size: int, fade_left: int, fade_right: int, dtype, device):
+    """Create a one-dimensional cosine edge window for tile blending."""
+    win = torch.ones(int(size), dtype=dtype, device=device)
+    if fade_left > 0:
+        fl = min(int(fade_left), int(size))
+        i = torch.arange(fl, dtype=dtype, device=device)
+        # Use fl rather than fl - 1 to preserve a nonzero center plateau for small fades.
+        win[:fl] = 0.5 * (1.0 - torch.cos(math.pi * i / max(1, fl)))
+    if fade_right > 0:
+        fr = min(int(fade_right), int(size))
+        i = torch.arange(fr, dtype=dtype, device=device)
+        win[int(size) - fr:] = 0.5 * (1.0 + torch.cos(math.pi * i / max(1, fr)))
+    return win
+
+
+def _pack_tensors(tensors: Dict[str, torch.Tensor]) -> str:
+    """Serialize tensors to a base64 safetensors payload for JSON transport."""
+    cpu_tensors = {
+        name: tensor.detach().contiguous().to("cpu")
+        for name, tensor in tensors.items()
+        if tensor is not None
+    }
+    payload = safetensors_save(cpu_tensors)
+    return base64.b64encode(payload).decode("ascii")
+
+
+def _unpack_tensors(payload_b64: str, device) -> Dict[str, torch.Tensor]:
+    payload = base64.b64decode(payload_b64.encode("ascii"))
+    return {name: tensor.to(device) for name, tensor in safetensors_load(payload).items()}
+
+
+def _extract_tensor_and_rebuilder(value) -> Tuple[torch.Tensor, Callable[[torch.Tensor], object]]:
+    """Extract the tensor from common Comfy latent containers and preserve shape."""
+    if isinstance(value, dict) and "samples" in value:
+        original = value
+
+        def rebuild(tensor):
+            updated = dict(original)
+            updated["samples"] = tensor
+            return updated
+
+        return value["samples"], rebuild
+
+    if isinstance(value, tuple) and value and torch.is_tensor(value[0]):
+        original = value
+
+        def rebuild(tensor):
+            return (tensor, *original[1:])
+
+        return value[0], rebuild
+
+    if isinstance(value, list) and value and torch.is_tensor(value[0]):
+        original = value
+
+        def rebuild(tensor):
+            updated = list(original)
+            updated[0] = tensor
+            return updated
+
+        return value[0], rebuild
+
+    if torch.is_tensor(value):
+        return value, lambda tensor: tensor
+
+    raise TypeError("Expected latent_image to be a tensor or latent container with a tensor samples field")
+
+
+def _extract_optional_mask(mask):
+    if mask is None:
+        return None
+    if isinstance(mask, dict) and "samples" in mask:
+        return mask["samples"]
+    return mask
+
+
+def _tile_latent(value: torch.Tensor, tile: dict) -> torch.Tensor:
+    return value[:, :, :, tile["h_start"]:tile["h_end"], tile["w_start"]:tile["w_end"]].contiguous()
+
+
+def _tile_mask(mask, tile: dict):
+    if mask is None:
+        return None
+    if torch.is_tensor(mask) and mask.ndim >= 5:
+        return mask[:, :, :, tile["h_start"]:tile["h_end"], tile["w_start"]:tile["w_end"]].contiguous()
+    if torch.is_tensor(mask) and mask.ndim >= 4:
+        return mask[:, :, tile["h_start"]:tile["h_end"], tile["w_start"]:tile["w_end"]].contiguous()
+    return mask
+
+
+def _sample_tile(guider, tile_noise, tile_latent, sampler, sigmas, tile_mask):
+    result = guider.sample(
+        tile_noise,
+        tile_latent,
+        sampler,
+        sigmas,
+        denoise_mask=tile_mask,
+        disable_pbar=True,
+    )
+    if isinstance(result, tuple):
+        tile_samples = result[0]
+        tile_denoised = result[1] if len(result) > 1 else None
+    else:
+        tile_samples = result
+        tile_denoised = None
+    return tile_samples, tile_denoised
+
+
+def _build_tile_queue(latent_shape: Iterable[int], n_h_tiles: int, n_w_tiles: int, overlap: int) -> List[dict]:
+    _batch, _channels, _frames, height, width = [int(v) for v in latent_shape]
+    starts_y, tile_h = _compute_tile_starts(height, n_h_tiles, overlap)
+    starts_x, tile_w = _compute_tile_starts(width, n_w_tiles, overlap)
+    queue = []
+    tile_idx = 0
+    for grid_y, y_start in enumerate(starts_y):
+        y_end = min(y_start + tile_h, height)
+        for grid_x, x_start in enumerate(starts_x):
+            x_end = min(x_start + tile_w, width)
+            queue.append(
+                {
+                    "tile_index": tile_idx,
+                    "h_start": y_start,
+                    "h_end": y_end,
+                    "w_start": x_start,
+                    "w_end": x_end,
+                    "grid_y": grid_y,
+                    "grid_x": grid_x,
+                }
+            )
+            tile_idx += 1
+    return queue
+
+
+def _blend_tile_into_sums(
+    tile: dict,
+    samples: torch.Tensor,
+    denoised: Optional[torch.Tensor],
+    output_sum: torch.Tensor,
+    weight_sum: torch.Tensor,
+    denoised_sum: Optional[torch.Tensor],
+    n_h_tiles: int,
+    n_w_tiles: int,
+    overlap: int,
+):
+    y_start = tile["h_start"]
+    y_end = tile["h_end"]
+    x_start = tile["w_start"]
+    x_end = tile["w_end"]
+    grid_y = tile["grid_y"]
+    grid_x = tile["grid_x"]
+    fade_top = overlap if grid_y > 0 else 0
+    fade_bottom = overlap if grid_y < n_h_tiles - 1 else 0
+    fade_left = overlap if grid_x > 0 else 0
+    fade_right = overlap if grid_x < n_w_tiles - 1 else 0
+    dtype = output_sum.dtype
+    device = output_sum.device
+    wy = _make_window_1d(y_end - y_start, fade_top, fade_bottom, dtype, device)
+    wx = _make_window_1d(x_end - x_start, fade_left, fade_right, dtype, device)
+    weights = wy.view(1, 1, 1, y_end - y_start, 1) * wx.view(1, 1, 1, 1, x_end - x_start)
+    output_sum[:, :, :, y_start:y_end, x_start:x_end] += samples.to(device=device, dtype=dtype) * weights
+    weight_sum[:, :, :, y_start:y_end, x_start:x_end] += weights
+    if denoised_sum is not None and denoised is not None:
+        denoised_sum[:, :, :, y_start:y_end, x_start:x_end] += denoised.to(device=device, dtype=dtype) * weights
+
+
+async def _get_next_ltx_tile_index(multi_job_id: str) -> Optional[int]:
+    prompt_server = ensure_tile_jobs_initialized()
+    pending_queue = None
+    async with prompt_server.distributed_tile_jobs_lock:
+        job_data = prompt_server.distributed_pending_tile_jobs.get(multi_job_id)
+        if isinstance(job_data, TileJobState):
+            pending_queue = job_data.pending_tasks
+    if pending_queue is None:
+        return None
+    try:
+        return await asyncio.wait_for(pending_queue.get(), timeout=0.1)
+    except asyncio.TimeoutError:
+        return None
+
+
+async def _get_ltx_completed_tasks(multi_job_id: str) -> dict:
+    prompt_server = ensure_tile_jobs_initialized()
+    async with prompt_server.distributed_tile_jobs_lock:
+        job_data = prompt_server.distributed_pending_tile_jobs.get(multi_job_id)
+        if isinstance(job_data, TileJobState):
+            return dict(job_data.completed_tasks)
+    return {}
+
+
+async def _check_ltx_job_status(multi_job_id: str, master_url: str) -> bool:
+    session = await get_client_session()
+    async with session.get(f"{master_url}/distributed/job_status?multi_job_id={multi_job_id}") as response:
+        if response.status != 200:
+            return False
+        data = await response.json()
+        return bool(data.get("ready", False))
+
+
+async def _request_tile_from_master(multi_job_id: str, master_url: str, worker_id: str):
+    session = await get_client_session()
+    async with session.post(
+        f"{master_url}/distributed/request_image",
+        json={"multi_job_id": multi_job_id, "worker_id": str(worker_id)},
+        timeout=aiohttp.ClientTimeout(total=TILE_WAIT_TIMEOUT),
+    ) as response:
+        if response.status == 200:
+            return await response.json()
+        if response.status == 404:
+            return None
+        response.raise_for_status()
+
+
+async def _submit_ltx_tile_to_master(
+    multi_job_id: str,
+    master_url: str,
+    worker_id: str,
+    tile_index: int,
+    payload_b64: Optional[str] = None,
+    is_last: bool = False,
+):
+    session = await get_client_session()
+    body = {
+        "multi_job_id": multi_job_id,
+        "worker_id": str(worker_id),
+        "tile_index": int(tile_index) if tile_index is not None else None,
+        "payload": payload_b64,
+        "is_last": bool(is_last),
+    }
+    async with session.post(f"{master_url}/distributed/ltx/submit_tile", json=body) as response:
+        response.raise_for_status()
+        return await response.json()
+
+
+async def _ltx_worker_process(
+    node,
+    noise,
+    guider,
+    sampler,
+    sigmas,
+    latent_tensor,
+    denoise_mask,
+    n_h_tiles,
+    n_w_tiles,
+    tile_overlap,
+):
+    job_id = node.multi_job_id
+    master_url = node.master_url
+    worker_id = node.worker_id or "ltx-worker"
+    tiles_by_index = {
+        tile["tile_index"]: tile
+        for tile in _build_tile_queue(latent_tensor.shape, n_h_tiles, n_w_tiles, tile_overlap)
+    }
+    for attempt in range(120):
+        if await _check_ltx_job_status(job_id, master_url):
+            break
+        await asyncio.sleep(1.0)
+    else:
+        log(f"LTX Dist Worker[{worker_id[:8]}]: job {job_id} was not ready; aborting")
+        return
+
+    full_noise = noise.generate_noise({"samples": latent_tensor})
+
+    last_heartbeat = 0.0
+    while True:
+        data = await _request_tile_from_master(job_id, master_url, worker_id)
+        if not data:
+            log(f"LTX Dist Worker[{worker_id[:8]}]: master job unavailable; stopping")
+            break
+        tile_index = data.get("tile_idx")
+        if tile_index is None:
+            await _submit_ltx_tile_to_master(job_id, master_url, worker_id, -1, is_last=True)
+            break
+        tile = tiles_by_index.get(int(tile_index))
+        if tile is None:
+            raise RuntimeError(f"Master assigned unknown LTX tile index {tile_index}")
+
+        tile_latent = _tile_latent(latent_tensor, tile)
+        tile_noise = _tile_latent(full_noise, tile)
+        tile_mask = _tile_mask(denoise_mask, tile)
+        tile_samples, tile_denoised = _sample_tile(guider, tile_noise, tile_latent, sampler, sigmas, tile_mask)
+        tensors = {"samples": tile_samples}
+        if tile_denoised is not None:
+            tensors["denoised"] = tile_denoised
+        await _submit_ltx_tile_to_master(
+            job_id,
+            master_url,
+            worker_id,
+            int(tile_index),
+            _pack_tensors(tensors),
+        )
+        now = asyncio.get_running_loop().time()
+        if now - last_heartbeat >= HEARTBEAT_INTERVAL:
+            await _send_heartbeat_to_master(job_id, master_url, worker_id)
+            last_heartbeat = now
+
+
+async def _ltx_master_process(
+    node,
+    noise,
+    guider,
+    sampler,
+    sigmas,
+    latent_tensor,
+    denoise_mask,
+    n_h_tiles,
+    n_w_tiles,
+    tile_overlap,
+):
+    job_id = node.multi_job_id
+    enabled_workers = node.enabled_worker_ids
+    tiles = _build_tile_queue(latent_tensor.shape, n_h_tiles, n_w_tiles, tile_overlap)
+    tiles_by_index = {tile["tile_index"]: tile for tile in tiles}
+    total_tiles = len(tiles)
+    await init_static_job_batched(job_id, batch_size=1, num_tiles_per_image=total_tiles, enabled_workers=enabled_workers)
+
+    full_noise = noise.generate_noise({"samples": latent_tensor})
+    output_sum = torch.zeros_like(latent_tensor)
+    weight_sum = torch.zeros_like(latent_tensor)
+    denoised_sum = torch.zeros_like(latent_tensor) if denoise_mask is not None else None
+    integrated = set()
+    last_timeout_check = 0.0
+
+    try:
+        while len(integrated) < total_tiles:
+            await _drain_results_queue(job_id)
+            completed = await _get_ltx_completed_tasks(job_id)
+            for tile_index, result in list(completed.items()):
+                if tile_index in integrated:
+                    continue
+                if not isinstance(result, dict) or "payload" not in result:
+                    # Locally completed tiles are blended immediately and then marked complete.
+                    continue
+                tile = tiles_by_index.get(int(tile_index))
+                if tile is None:
+                    continue
+                tensors = _unpack_tensors(result["payload"], latent_tensor.device)
+                _blend_tile_into_sums(
+                    tile,
+                    tensors["samples"],
+                    tensors.get("denoised"),
+                    output_sum,
+                    weight_sum,
+                    denoised_sum,
+                    n_h_tiles,
+                    n_w_tiles,
+                    tile_overlap,
+                )
+                integrated.add(int(tile_index))
+
+            if len(integrated) >= total_tiles:
+                break
+
+            tile_index = await _get_next_ltx_tile_index(job_id)
+            if tile_index is not None and int(tile_index) not in integrated:
+                tile = tiles_by_index[int(tile_index)]
+                tile_latent = _tile_latent(latent_tensor, tile)
+                tile_noise = _tile_latent(full_noise, tile)
+                tile_mask = _tile_mask(denoise_mask, tile)
+                tile_samples, tile_denoised = _sample_tile(guider, tile_noise, tile_latent, sampler, sigmas, tile_mask)
+                _blend_tile_into_sums(
+                    tile,
+                    tile_samples,
+                    tile_denoised,
+                    output_sum,
+                    weight_sum,
+                    denoised_sum,
+                    n_h_tiles,
+                    n_w_tiles,
+                    tile_overlap,
+                )
+                integrated.add(int(tile_index))
+                await _mark_task_completed(job_id, int(tile_index), {"tile_idx": int(tile_index), "source": "master"})
+                continue
+
+            now = asyncio.get_running_loop().time()
+            if now - last_timeout_check >= 10.0:
+                requeued = await _check_and_requeue_timed_out_workers(job_id, total_tiles)
+                if requeued:
+                    log(f"LTX Dist: Requeued {requeued} timed-out tile(s)")
+                last_timeout_check = now
+
+            completed_count = await _get_completed_count(job_id)
+            debug_log(f"LTX Dist: Progress {completed_count}/{total_tiles} completed, {len(integrated)} integrated")
+            await asyncio.sleep(0.1)
+    finally:
+        await _cleanup_job(job_id)
+
+    weight_sum = torch.where(weight_sum == 0, torch.ones_like(weight_sum), weight_sum)
+    samples = output_sum / weight_sum
+    denoised = denoised_sum / weight_sum if denoised_sum is not None else None
+    return samples, denoised
+
+
+def _ltx_local_tiled_process(
+    noise,
+    guider,
+    sampler,
+    sigmas,
+    latent_tensor,
+    denoise_mask,
+    n_h_tiles,
+    n_w_tiles,
+    tile_overlap,
+):
+    """Process all LTX latent tiles locally in the current ComfyUI execution context.
+
+    This path is used when no workers are enabled. Keeping it synchronous avoids
+    crossing ComfyUI's execution/inference context boundary while still using the
+    same tile extraction and blend logic as the distributed master/worker path.
+    """
+    tiles = _build_tile_queue(latent_tensor.shape, n_h_tiles, n_w_tiles, tile_overlap)
+    full_noise = noise.generate_noise({"samples": latent_tensor})
+    output_sum = torch.zeros_like(latent_tensor)
+    weight_sum = torch.zeros_like(latent_tensor)
+    denoised_sum = None
+
+    for tile in tiles:
+        tile_latent = _tile_latent(latent_tensor, tile)
+        tile_noise = _tile_latent(full_noise, tile)
+        tile_mask = _tile_mask(denoise_mask, tile)
+        tile_samples, tile_denoised = _sample_tile(guider, tile_noise, tile_latent, sampler, sigmas, tile_mask)
+        if denoised_sum is None and tile_denoised is not None:
+            denoised_sum = torch.zeros_like(latent_tensor)
+        _blend_tile_into_sums(
+            tile,
+            tile_samples,
+            tile_denoised,
+            output_sum,
+            weight_sum,
+            denoised_sum,
+            n_h_tiles,
+            n_w_tiles,
+            tile_overlap,
+        )
+
+    weight_sum = torch.where(weight_sum == 0, torch.ones_like(weight_sum), weight_sum)
+    samples = output_sum / weight_sum
+    denoised = denoised_sum / weight_sum if denoised_sum is not None else None
+    return samples, denoised
+
+
+class LTXTiledSamplerDistributed:
+    """Distributed wrapper around the LTX tiled sampler."""
+
+    @classmethod
+    def INPUT_TYPES(cls):
+        base_inputs = _copy_base_inputs()
+        base_inputs.setdefault("hidden", {})
+        base_inputs["hidden"].update(
+            {
+                "multi_job_id": ("STRING", {"default": ""}),
+                "is_worker": ("BOOLEAN", {"default": False}),
+                "master_url": ("STRING", {"default": ""}),
+                "worker_id": ("STRING", {"default": ""}),
+                "enabled_worker_ids": ("STRING", {"default": "[]"}),
+            }
+        )
+        return base_inputs
+
+    RETURN_TYPES = getattr(LTXTiledSampler, "RETURN_TYPES", ("LATENT", "LATENT", "LATENT"))
+    RETURN_NAMES = getattr(LTXTiledSampler, "RETURN_NAMES", ("samples", "denoised", "preview"))
+    FUNCTION = "sample_tiled_distributed"
+    CATEGORY = "latent/diffusion"
+
+    @classmethod
+    def IS_CHANGED(cls, **kwargs):
+        return float("nan")
+
+    def __init__(self):
+        self.base = LTXTiledSampler() if LTXTiledSampler is not None else None
+
+    def _run_base(self, noise, guider, sampler, sigmas, latent_image, denoise_mask, h_tiles, w_tiles, overlap):
+        if self.base is None:
+            raise RuntimeError(
+                "LTXTiledSamplerDistributed requires the base LTXTiledSampler node for local fallback. "
+                "Install the LTX tiled sampler dependency or run with distributed multi_job_id."
+            )
+        return self.base.sample_tiled(
+            noise,
+            guider,
+            sampler,
+            sigmas,
+            latent_image,
+            denoise_mask,
+            h_tiles,
+            w_tiles,
+            overlap,
+        )
+
+    def sample_tiled_distributed(
+        self,
+        noise,
+        guider,
+        sampler,
+        sigmas,
+        latent_image,
+        denoise_mask=None,
+        h_tiles: int = 1,
+        w_tiles: int = 1,
+        overlap: int = 0,
+        multi_job_id: str = "",
+        is_worker: bool = False,
+        master_url: str = "",
+        worker_id: str = "",
+        enabled_worker_ids: str = "[]",
+    ):
+        if not multi_job_id:
+            return self._run_base(noise, guider, sampler, sigmas, latent_image, denoise_mask, h_tiles, w_tiles, overlap)
+
+        latent_tensor, rebuild_latent = _extract_tensor_and_rebuilder(latent_image)
+        if latent_tensor.ndim != 5:
+            raise ValueError(
+                f"LTXTiledSamplerDistributed expects a 5D latent tensor [B,C,F,H,W]; got shape {tuple(latent_tensor.shape)}"
+            )
+        mask_tensor = _extract_optional_mask(denoise_mask)
+        self.multi_job_id = str(multi_job_id)
+        self.master_url = str(master_url or "").rstrip("/")
+        self.worker_id = str(worker_id or "")
+        if isinstance(enabled_worker_ids, str):
+            try:
+                import json
+
+                parsed_workers = json.loads(enabled_worker_ids) if enabled_worker_ids else []
+            except Exception:
+                parsed_workers = []
+        else:
+            parsed_workers = enabled_worker_ids or []
+        self.enabled_worker_ids = [str(worker) for worker in parsed_workers]
+
+        if is_worker:
+            run_async_in_server_loop(
+                _ltx_worker_process(
+                    self,
+                    noise,
+                    guider,
+                    sampler,
+                    sigmas,
+                    latent_tensor,
+                    mask_tensor,
+                    int(h_tiles),
+                    int(w_tiles),
+                    int(overlap),
+                ),
+                timeout=None,
+            )
+            placeholder = rebuild_latent(latent_tensor)
+            return (placeholder, placeholder, placeholder)
+
+        if not self.enabled_worker_ids:
+            samples, denoised = _ltx_local_tiled_process(
+                noise,
+                guider,
+                sampler,
+                sigmas,
+                latent_tensor,
+                mask_tensor,
+                int(h_tiles),
+                int(w_tiles),
+                int(overlap),
+            )
+        else:
+            samples, denoised = run_async_in_server_loop(
+                _ltx_master_process(
+                    self,
+                    noise,
+                    guider,
+                    sampler,
+                    sigmas,
+                    latent_tensor,
+                    mask_tensor,
+                    int(h_tiles),
+                    int(w_tiles),
+                    int(overlap),
+                ),
+                timeout=None,
+            )
+        samples_latent = rebuild_latent(samples)
+        denoised_latent = rebuild_latent(denoised if denoised is not None else samples)
+        return (samples_latent, denoised_latent, samples_latent)
+
+
+NODE_CLASS_MAPPINGS = {"LTXTiledSamplerDistributed": LTXTiledSamplerDistributed}
+NODE_DISPLAY_NAME_MAPPINGS = {"LTXTiledSamplerDistributed": "LTX Tiled Sampler (Distributed)"}

pyproject.toml

@@ -1,9 +1,9 @@
 [project]
 name = "ComfyUI-Distributed"
 description = "ComfyUI extension that enables multi-GPU processing locally, remotely and in the cloud"
-version = "1.3.1"
+version = "1.4.6"
 license = {file = "LICENSE"}
-dependencies  = []
+dependencies  = ["safetensors"]
 
 [project.urls]
 Repository = "https://github.com/robertvoy/ComfyUI-Distributed"

tests/api/test_distributed_queue.py

@@ -162,7 +162,7 @@ def _load_job_routes_module():
     sys.modules[f"{package_name}.utils.async_helpers"] = async_helpers_module
 
     queue_orchestration_module = types.ModuleType(f"{package_name}.api.queue_orchestration")
-    queue_orchestration_module.orchestrate_distributed_execution = AsyncMock(return_value=("prompt_dist", 1))
+    queue_orchestration_module.orchestrate_distributed_execution = AsyncMock(return_value=("prompt_dist", 7, 1, {}))
     sys.modules[f"{package_name}.api.queue_orchestration"] = queue_orchestration_module
 
     @dataclass(frozen=True)
@@ -221,7 +221,7 @@ job_routes = _load_job_routes_module()
 
 
 class DistributedQueueEndpointTests(unittest.IsolatedAsyncioTestCase):
-    async def test_distributed_queue_happy_path_returns_prompt_id(self):
+    async def test_distributed_queue_happy_path_returns_prompt_metadata(self):
         request = _FakeRequest(
             {
                 "prompt": {"1": {"class_type": "Node"}},
@@ -233,12 +233,14 @@ class DistributedQueueEndpointTests(unittest.IsolatedAsyncioTestCase):
         with patch.object(
             job_routes,
             "orchestrate_distributed_execution",
-            new=AsyncMock(return_value=("prompt_123", 2)),
+            new=AsyncMock(return_value=("prompt_123", 42, 2, {})),
         ):
             response = await job_routes.distributed_queue_endpoint(request)
 
         self.assertEqual(response.status, 200)
         self.assertEqual(response.payload.get("prompt_id"), "prompt_123")
+        self.assertEqual(response.payload.get("number"), 42)
+        self.assertEqual(response.payload.get("node_errors"), {})
         self.assertTrue(response.payload.get("auto_prepare_supported"))
 
     async def test_distributed_queue_missing_prompt_returns_400(self):

tests/test_async_helpers.py

@@ -0,0 +1,91 @@
+import importlib.util
+import sys
+import types
+import unittest
+from pathlib import Path
+
+
+class _PromptQueue:
+    def __init__(self):
+        self.items = []
+
+    def put(self, item):
+        self.items.append(item)
+
+
+def _load_async_helpers_module():
+    module_path = Path(__file__).resolve().parents[1] / "utils" / "async_helpers.py"
+    package_name = "dist_async_helpers_testpkg"
+
+    for mod_name in list(sys.modules):
+        if mod_name == package_name or mod_name.startswith(f"{package_name}."):
+            del sys.modules[mod_name]
+
+    root_pkg = types.ModuleType(package_name)
+    root_pkg.__path__ = []
+    sys.modules[package_name] = root_pkg
+
+    utils_pkg = types.ModuleType(f"{package_name}.utils")
+    utils_pkg.__path__ = []
+    sys.modules[f"{package_name}.utils"] = utils_pkg
+
+    execution_module = types.ModuleType("execution")
+
+    async def _validate_prompt(prompt_id, prompt, partial_execution_targets):
+        return (True, None, ["9"], {})
+
+    execution_module.validate_prompt = _validate_prompt
+    execution_module.SENSITIVE_EXTRA_DATA_KEYS = []
+    sys.modules["execution"] = execution_module
+
+    prompt_server = types.SimpleNamespace(
+        trigger_on_prompt=lambda payload: payload,
+        number=12,
+        prompt_queue=_PromptQueue(),
+    )
+    server_module = types.ModuleType("server")
+    server_module.PromptServer = types.SimpleNamespace(instance=prompt_server)
+    sys.modules["server"] = server_module
+
+    network_module = types.ModuleType(f"{package_name}.utils.network")
+    network_module.get_server_loop = lambda: None
+    sys.modules[f"{package_name}.utils.network"] = network_module
+
+    spec = importlib.util.spec_from_file_location(f"{package_name}.utils.async_helpers", module_path)
+    module = importlib.util.module_from_spec(spec)
+    assert spec is not None and spec.loader is not None
+    spec.loader.exec_module(module)
+    return module, prompt_server
+
+
+async_helpers, prompt_server = _load_async_helpers_module()
+
+
+class QueuePromptPayloadTests(unittest.IsolatedAsyncioTestCase):
+    async def test_queue_prompt_payload_includes_create_time_and_client_metadata(self):
+        result = await async_helpers.queue_prompt_payload(
+            {"1": {"class_type": "Node"}},
+            workflow_meta={"id": "workflow-1"},
+            client_id="client-1",
+            include_queue_metadata=True,
+        )
+
+        self.assertIsInstance(result["prompt_id"], str)
+        self.assertTrue(result["prompt_id"])
+        self.assertEqual(result["number"], 12)
+        self.assertEqual(result["node_errors"], {})
+
+        self.assertEqual(prompt_server.number, 13)
+        self.assertEqual(len(prompt_server.prompt_queue.items), 1)
+        queued_item = prompt_server.prompt_queue.items[0]
+        self.assertEqual(queued_item[0], 12)
+        extra_data = queued_item[3]
+        self.assertEqual(extra_data["client_id"], "client-1")
+        self.assertIn("create_time", extra_data)
+        self.assertIsInstance(extra_data["create_time"], int)
+        self.assertGreater(extra_data["create_time"], 0)
+        self.assertEqual(extra_data["extra_pnginfo"]["workflow"], {"id": "workflow-1"})
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_collector_list_inputs.py

@@ -0,0 +1,205 @@
+import importlib.util
+import sys
+import types
+import asyncio
+from pathlib import Path
+
+import torch
+
+
+def _load_collector_module():
+    module_path = Path(__file__).resolve().parents[1] / "nodes" / "collector.py"
+    package_name = "dist_collector_list_testpkg"
+
+    for mod_name in list(sys.modules):
+        if mod_name == package_name or mod_name.startswith(f"{package_name}."):
+            del sys.modules[mod_name]
+
+    root_pkg = types.ModuleType(package_name)
+    root_pkg.__path__ = []
+    sys.modules[package_name] = root_pkg
+
+    nodes_pkg = types.ModuleType(f"{package_name}.nodes")
+    nodes_pkg.__path__ = []
+    sys.modules[f"{package_name}.nodes"] = nodes_pkg
+
+    utils_pkg = types.ModuleType(f"{package_name}.utils")
+    utils_pkg.__path__ = []
+    sys.modules[f"{package_name}.utils"] = utils_pkg
+
+    class _Routes:
+        def post(self, _path):
+            return lambda fn: fn
+
+        def get(self, _path):
+            return lambda fn: fn
+
+    prompt_server = types.SimpleNamespace(
+        routes=_Routes(),
+        distributed_jobs_lock=None,
+        distributed_pending_jobs={},
+    )
+    server_module = types.ModuleType("server")
+    server_module.PromptServer = types.SimpleNamespace(instance=prompt_server)
+    sys.modules["server"] = server_module
+
+    comfy_module = types.ModuleType("comfy")
+    model_management = types.ModuleType("comfy.model_management")
+
+    class InterruptProcessingException(Exception):
+        pass
+
+    model_management.InterruptProcessingException = InterruptProcessingException
+    model_management.throw_exception_if_processing_interrupted = lambda: None
+    comfy_module.model_management = model_management
+
+    comfy_utils = types.ModuleType("comfy.utils")
+
+    class ProgressBar:
+        def __init__(self, _total):
+            self.total = _total
+            self.updates = []
+
+        def update(self, value):
+            self.updates.append(value)
+
+    comfy_utils.ProgressBar = ProgressBar
+    comfy_module.utils = comfy_utils
+    sys.modules["comfy"] = comfy_module
+    sys.modules["comfy.model_management"] = model_management
+    sys.modules["comfy.utils"] = comfy_utils
+
+    aiohttp_module = types.ModuleType("aiohttp")
+    aiohttp_module.ClientTimeout = lambda total: types.SimpleNamespace(total=total)
+    sys.modules["aiohttp"] = aiohttp_module
+
+    logging_module = types.ModuleType(f"{package_name}.utils.logging")
+    logging_module.debug_log = lambda *_args, **_kwargs: None
+    logging_module.log = lambda *_args, **_kwargs: None
+    sys.modules[f"{package_name}.utils.logging"] = logging_module
+
+    config_module = types.ModuleType(f"{package_name}.utils.config")
+    config_module.get_worker_timeout_seconds = lambda: 0.1
+    config_module.load_config = lambda: {"workers": []}
+    config_module.is_master_delegate_only = lambda: False
+    sys.modules[f"{package_name}.utils.config"] = config_module
+
+    constants_module = types.ModuleType(f"{package_name}.utils.constants")
+    constants_module.HEARTBEAT_INTERVAL = 1.0
+    sys.modules[f"{package_name}.utils.constants"] = constants_module
+
+    image_module = types.ModuleType(f"{package_name}.utils.image")
+    def _ensure_contiguous(tensor):
+        return tensor.contiguous() if hasattr(tensor, "contiguous") else tensor
+
+    image_module.ensure_contiguous = _ensure_contiguous
+    image_module.tensor_to_pil = lambda *_args, **_kwargs: None
+    image_module.pil_to_tensor = lambda value: value
+    sys.modules[f"{package_name}.utils.image"] = image_module
+
+    network_module = types.ModuleType(f"{package_name}.utils.network")
+    network_module.build_worker_url = lambda worker: "http://worker"
+    network_module.get_client_session = lambda: None
+    network_module.probe_worker = lambda *_args, **_kwargs: None
+    sys.modules[f"{package_name}.utils.network"] = network_module
+
+    audio_payload_module = types.ModuleType(f"{package_name}.utils.audio_payload")
+    audio_payload_module.encode_audio_payload = lambda audio: audio
+    sys.modules[f"{package_name}.utils.audio_payload"] = audio_payload_module
+
+    async_helpers_module = types.ModuleType(f"{package_name}.utils.async_helpers")
+    async_helpers_module.run_async_in_server_loop = lambda coro: asyncio.run(coro)
+    sys.modules[f"{package_name}.utils.async_helpers"] = async_helpers_module
+
+    spec = importlib.util.spec_from_file_location(f"{package_name}.nodes.collector", module_path)
+    module = importlib.util.module_from_spec(spec)
+    assert spec is not None and spec.loader is not None
+    spec.loader.exec_module(module)
+    return module
+
+
+def test_collector_opts_into_comfyui_list_inputs():
+    collector = _load_collector_module().DistributedCollectorNode
+
+    assert collector.INPUT_IS_LIST is True
+
+
+def test_pass_through_collapses_comfyui_image_list_to_batch_and_unwraps_hidden_inputs():
+    collector = _load_collector_module().DistributedCollectorNode()
+    first = torch.zeros(1, 2, 2, 3)
+    second = torch.ones(1, 2, 2, 3)
+
+    images, audio = collector.run(
+        images=[first, second],
+        load_balance=[False],
+        audio=[None],
+        multi_job_id=[""],
+        is_worker=[False],
+        master_url=[""],
+        enabled_worker_ids=["[]"],
+        worker_batch_size=[1],
+        worker_id=[""],
+        pass_through=[False],
+        delegate_only=[False],
+    )
+
+    assert tuple(images.shape) == (2, 2, 2, 3)
+    assert torch.equal(images[0:1], first)
+    assert torch.equal(images[1:2], second)
+    assert tuple(audio["waveform"].shape) == (1, 2, 1)
+
+
+def test_worker_list_input_sends_one_completion_sequence_with_last_only_on_final_item():
+    module = _load_collector_module()
+    collector = module.DistributedCollectorNode()
+    first = torch.zeros(1, 2, 2, 3)
+    second = torch.ones(1, 2, 2, 3)
+    posted_payloads = []
+
+    class _FakeImage:
+        def save(self, fp, format=None, compress_level=None):
+            fp.write(b"png-bytes")
+
+    class _FakeResponse:
+        async def __aenter__(self):
+            return self
+
+        async def __aexit__(self, exc_type, exc, tb):
+            return False
+
+        def raise_for_status(self):
+            return None
+
+    class _FakeSession:
+        def post(self, url, json, timeout):
+            posted_payloads.append(json)
+            return _FakeResponse()
+
+    async def _fake_get_client_session():
+        return _FakeSession()
+
+    module.tensor_to_pil = lambda *_args, **_kwargs: _FakeImage()
+    module.get_client_session = _fake_get_client_session
+    module.encode_audio_payload = lambda _audio: None
+
+    images, audio = collector.run(
+        images=[first, second],
+        load_balance=[False],
+        audio=[None],
+        multi_job_id=["job-list-1"],
+        is_worker=[True],
+        master_url=["http://master"],
+        enabled_worker_ids=["[]"],
+        worker_batch_size=[1],
+        worker_id=["worker-a"],
+        pass_through=[False],
+        delegate_only=[False],
+    )
+
+    assert tuple(images.shape) == (2, 2, 2, 3)
+    assert tuple(audio["waveform"].shape) == (1, 2, 1)
+    assert len(posted_payloads) == 2
+    assert [payload["batch_idx"] for payload in posted_payloads] == [0, 1]
+    assert [payload["is_last"] for payload in posted_payloads] == [False, True]
+    assert {payload["job_id"] for payload in posted_payloads} == {"job-list-1"}
+    assert {payload["worker_id"] for payload in posted_payloads} == {"worker-a"}

tests/test_ltx_tiled_sampler.py

@@ -0,0 +1,157 @@
+import asyncio
+import importlib.util
+import sys
+import types
+import unittest
+from pathlib import Path
+
+import torch
+
+
+def _load_ltx_module():
+    module_path = Path(__file__).resolve().parents[1] / "nodes" / "ltx_tiled_sampler.py"
+    package_name = "dist_ltx_testpkg"
+
+    for mod_name in list(sys.modules):
+        if mod_name == package_name or mod_name.startswith(f"{package_name}."):
+            del sys.modules[mod_name]
+
+    root_pkg = types.ModuleType(package_name)
+    root_pkg.__path__ = []
+    sys.modules[package_name] = root_pkg
+
+    for subpkg in ["nodes", "upscale", "utils"]:
+        pkg = types.ModuleType(f"{package_name}.{subpkg}")
+        pkg.__path__ = []
+        sys.modules[f"{package_name}.{subpkg}"] = pkg
+
+    job_models = types.ModuleType(f"{package_name}.upscale.job_models")
+
+    class TileJobState:
+        pass
+
+    job_models.TileJobState = TileJobState
+    sys.modules[f"{package_name}.upscale.job_models"] = job_models
+
+    async def _noop(*_args, **_kwargs):
+        return None
+
+    job_store = types.ModuleType(f"{package_name}.upscale.job_store")
+    job_store._cleanup_job = _noop
+    job_store._drain_results_queue = _noop
+    job_store._get_completed_count = _noop
+    job_store._mark_task_completed = _noop
+    job_store.ensure_tile_jobs_initialized = lambda: types.SimpleNamespace(
+        distributed_tile_jobs_lock=asyncio.Lock(),
+        distributed_pending_tile_jobs={},
+    )
+    job_store.init_static_job_batched = _noop
+    sys.modules[f"{package_name}.upscale.job_store"] = job_store
+
+    job_timeout = types.ModuleType(f"{package_name}.upscale.job_timeout")
+    job_timeout._check_and_requeue_timed_out_workers = _noop
+    sys.modules[f"{package_name}.upscale.job_timeout"] = job_timeout
+
+    async_helpers = types.ModuleType(f"{package_name}.utils.async_helpers")
+    async_helpers.run_async_in_server_loop = lambda coro, timeout=None: asyncio.run(coro)
+    sys.modules[f"{package_name}.utils.async_helpers"] = async_helpers
+
+    constants = types.ModuleType(f"{package_name}.utils.constants")
+    constants.HEARTBEAT_INTERVAL = 10.0
+    constants.TILE_WAIT_TIMEOUT = 1.0
+    sys.modules[f"{package_name}.utils.constants"] = constants
+
+    logging = types.ModuleType(f"{package_name}.utils.logging")
+    logging.debug_log = lambda *_args, **_kwargs: None
+    logging.log = lambda *_args, **_kwargs: None
+    sys.modules[f"{package_name}.utils.logging"] = logging
+
+    network = types.ModuleType(f"{package_name}.utils.network")
+
+    async def _missing_session():
+        raise RuntimeError("network not available in unit test")
+
+    network.get_client_session = _missing_session
+    sys.modules[f"{package_name}.utils.network"] = network
+
+    usdu = types.ModuleType(f"{package_name}.utils.usdu_managment")
+    usdu._send_heartbeat_to_master = _noop
+    sys.modules[f"{package_name}.utils.usdu_managment"] = usdu
+
+    spec = importlib.util.spec_from_file_location(f"{package_name}.nodes.ltx_tiled_sampler", module_path)
+    module = importlib.util.module_from_spec(spec)
+    assert spec is not None and spec.loader is not None
+    spec.loader.exec_module(module)
+    return module
+
+
+ltx = _load_ltx_module()
+
+
+class LtxTiledSamplerTests(unittest.TestCase):
+    def test_compute_tile_starts_with_overlap_covers_edges(self):
+        starts, tile_size = ltx._compute_tile_starts(total_size=10, n_tiles=3, overlap=2)
+
+        self.assertEqual(starts[0], 0)
+        self.assertEqual(starts[-1] + tile_size, 10)
+        self.assertGreaterEqual(tile_size, 4)
+
+    def test_pack_unpack_tensors_round_trips_with_safetensors(self):
+        tensor = torch.arange(12, dtype=torch.float32).reshape(1, 1, 3, 2, 2)
+
+        payload = ltx._pack_tensors({"samples": tensor})
+        restored = ltx._unpack_tensors(payload, device="cpu")
+
+        self.assertTrue(torch.equal(restored["samples"], tensor))
+
+    def test_latent_dict_rebuilder_preserves_metadata(self):
+        samples = torch.zeros((1, 2, 3, 4, 5))
+        latent = {"samples": samples, "noise_mask": "keep-me"}
+
+        extracted, rebuild = ltx._extract_tensor_and_rebuilder(latent)
+        rebuilt = rebuild(torch.ones_like(extracted))
+
+        self.assertIs(extracted, samples)
+        self.assertEqual(rebuilt["noise_mask"], "keep-me")
+        self.assertTrue(torch.equal(rebuilt["samples"], torch.ones_like(samples)))
+
+    def test_node_schema_includes_distributed_hidden_inputs(self):
+        schema = ltx.LTXTiledSamplerDistributed.INPUT_TYPES()
+
+        self.assertIn("multi_job_id", schema["hidden"])
+        self.assertIn("is_worker", schema["hidden"])
+        self.assertIn("master_url", schema["hidden"])
+        self.assertIn("enabled_worker_ids", schema["hidden"])
+        self.assertIn("LTXTiledSamplerDistributed", ltx.NODE_CLASS_MAPPINGS)
+
+
+    def test_local_tiled_process_runs_without_async_master_path(self):
+        class _Noise:
+            def generate_noise(self, latent):
+                return torch.ones_like(latent['samples']) * 0.25
+
+        class _Guider:
+            def sample(self, noise, latent_image, sampler, sigmas, denoise_mask=None, disable_pbar=True):
+                return latent_image + noise
+
+        latent = torch.zeros((1, 1, 2, 4, 4), dtype=torch.float32)
+
+        samples, denoised = ltx._ltx_local_tiled_process(
+            _Noise(),
+            _Guider(),
+            sampler=object(),
+            sigmas=torch.tensor([1.0, 0.0]),
+            latent_tensor=latent,
+            denoise_mask=None,
+            n_h_tiles=2,
+            n_w_tiles=2,
+            tile_overlap=0,
+        )
+
+        self.assertIsNone(denoised)
+        self.assertEqual(tuple(samples.shape), tuple(latent.shape))
+        self.assertTrue(torch.allclose(samples, torch.full_like(latent, 0.25)))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_network_helpers.py

@@ -90,14 +90,48 @@ class NetworkHelpersTests(unittest.TestCase):
             "https://master.example.com",
         )
 
+    def test_build_master_url_ignores_stale_saved_port_and_uses_runtime_port(self):
+        cfg = {"master": {"host": "192.168.68.56", "port": 8001}}
+        prompt_server = types.SimpleNamespace(address="127.0.0.1", port=8188)
+        self.assertEqual(
+            network.build_master_url(config=cfg, prompt_server_instance=prompt_server),
+            "http://192.168.68.56:8188",
+        )
+
+    def test_build_master_url_keeps_explicit_port_in_host(self):
+        cfg = {"master": {"host": "192.168.68.56:8001"}}
+        prompt_server = types.SimpleNamespace(address="127.0.0.1", port=8188)
+        self.assertEqual(
+            network.build_master_url(config=cfg, prompt_server_instance=prompt_server),
+            "http://192.168.68.56:8001",
+        )
+
     def test_build_master_url_falls_back_to_server_address(self):
-        cfg = {"master": {"host": ""}}
+        cfg = {"master": {"host": "", "port": 8001}}
         prompt_server = types.SimpleNamespace(address="0.0.0.0", port=8190)
         self.assertEqual(
             network.build_master_url(config=cfg, prompt_server_instance=prompt_server),
             "http://127.0.0.1:8190",
         )
 
+    def test_build_master_callback_url_uses_loopback_for_local_worker(self):
+        cfg = {"master": {"host": "192.168.68.56"}}
+        prompt_server = types.SimpleNamespace(address="127.0.0.1", port=8001)
+        worker = {"id": "w1", "type": "local", "host": "localhost", "port": 8189}
+        self.assertEqual(
+            network.build_master_callback_url(worker, config=cfg, prompt_server_instance=prompt_server),
+            "http://127.0.0.1:8001",
+        )
+
+    def test_build_master_callback_url_keeps_public_master_url_for_remote_worker(self):
+        cfg = {"master": {"host": "192.168.68.56"}}
+        prompt_server = types.SimpleNamespace(address="127.0.0.1", port=8001)
+        worker = {"id": "w2", "type": "remote", "host": "192.168.68.99", "port": 8189}
+        self.assertEqual(
+            network.build_master_callback_url(worker, config=cfg, prompt_server_instance=prompt_server),
+            "http://192.168.68.56:8001",
+        )
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_prompt_transform.py

@@ -296,6 +296,501 @@ class PrepareDelegateMasterPromptTests(unittest.TestCase):
         empty_nodes = [n for n in result.values() if n.get("class_type") == "DistributedEmptyImage"]
         self.assertEqual(len(empty_nodes), 2)
 
+    def test_preserves_primitive_string_for_downstream_required_input(self):
+        """Delegate-only master keeps primitive inputs needed by SaveImage."""
+        save_image = type(
+            "SaveImage",
+            (),
+            {
+                "INPUT_TYPES": classmethod(
+                    lambda cls: {
+                        "required": {
+                            "images": ("IMAGE",),
+                            "filename_prefix": ("STRING",),
+                        }
+                    }
+                )
+            },
+        )
+        previous_mappings = getattr(pt, "_DELEGATE_MASTER_NODE_CLASS_MAPPINGS", None)
+        pt._DELEGATE_MASTER_NODE_CLASS_MAPPINGS = {"SaveImage": save_image}
+        try:
+            prompt = {
+                "8": {"class_type": "KSampler", "inputs": {}},
+                "11": {"class_type": "DistributedCollector", "inputs": {"images": ["8", 0]}},
+                "15": {"class_type": "PrimitiveString", "inputs": {"value": "test/input_bug"}},
+                "9": {
+                    "class_type": "SaveImage",
+                    "inputs": {
+                        "images": ["11", 0],
+                        "filename_prefix": ["15", 0],
+                    },
+                },
+            }
+
+            result = pt.prepare_delegate_master_prompt(prompt, ["11"])
+        finally:
+            pt._DELEGATE_MASTER_NODE_CLASS_MAPPINGS = previous_mappings
+
+        self.assertIn("15", result)
+        self.assertEqual(result["15"], prompt["15"])
+        self.assertEqual(result["9"]["inputs"]["filename_prefix"], ["15", 0])
+
+    def test_does_not_preserve_non_primitive_upstream_for_collector(self):
+        prompt = _delegate_prompt()
+        result = pt.prepare_delegate_master_prompt(prompt, ["3"])
+
+        self.assertNotIn("2", result)
+        self.assertNotEqual(result["3"]["inputs"]["images"], ["2", 0])
+
+    def test_preserves_load_image_for_switch_alternate_required_input(self):
+        """Delegate-only master keeps LoadImage inputs needed by switches."""
+        prompt = {
+            "848": {"class_type": "LoadImage", "inputs": {"image": "input_bug2_00003_.png"}},
+            "862": {"class_type": "VAEDecode", "inputs": {}},
+            "854": {"class_type": "DistributedCollector", "inputs": {"images": ["862", 0]}},
+            "850": {
+                "class_type": "ComfySwitchNode",
+                "inputs": {
+                    "on_false": ["848", 0],
+                    "on_true": ["854", 0],
+                },
+            },
+            "851": {"class_type": "PreviewImage", "inputs": {"images": ["850", 0]}},
+        }
+
+        result = pt.prepare_delegate_master_prompt(prompt, ["854"])
+
+        self.assertIn("848", result)
+        self.assertEqual(result["850"]["inputs"]["on_false"], ["848", 0])
+        self.assertEqual(result["850"]["inputs"]["on_true"], ["854", 0])
+        self.assertNotIn("862", result)
+
+    def test_preserves_registered_string_utility_subgraph_for_downstream_required_input(self):
+        """Delegate-only master keeps scalar utility chains used by SaveImage."""
+        string_concat = type(
+            "StringConcatenate",
+            (),
+            {
+                "RETURN_TYPES": ("STRING",),
+                "INPUT_TYPES": classmethod(
+                    lambda cls: {
+                        "required": {
+                            "string_a": ("STRING",),
+                            "string_b": ("STRING",),
+                        }
+                    }
+                ),
+            },
+        )
+        save_image = type(
+            "SaveImage",
+            (),
+            {
+                "INPUT_TYPES": classmethod(
+                    lambda cls: {
+                        "required": {
+                            "images": ("IMAGE",),
+                            "filename_prefix": ("STRING",),
+                        }
+                    }
+                )
+            },
+        )
+        previous_mappings = getattr(pt, "_DELEGATE_MASTER_NODE_CLASS_MAPPINGS", None)
+        pt._DELEGATE_MASTER_NODE_CLASS_MAPPINGS = {
+            "SaveImage": save_image,
+            "StringConcatenate": string_concat,
+        }
+        try:
+            prompt = {
+                "8": {"class_type": "KSampler", "inputs": {}},
+                "11": {"class_type": "DistributedCollector", "inputs": {"images": ["8", 0]}},
+                "15": {"class_type": "PrimitiveString", "inputs": {"value": "test"}},
+                "16": {"class_type": "PrimitiveString", "inputs": {"value": "input_bug3"}},
+                "17": {
+                    "class_type": "StringConcatenate",
+                    "inputs": {
+                        "string_a": ["15", 0],
+                        "string_b": ["16", 0],
+                    },
+                },
+                "9": {
+                    "class_type": "SaveImage",
+                    "inputs": {
+                        "images": ["11", 0],
+                        "filename_prefix": ["17", 0],
+                    },
+                },
+            }
+
+            result = pt.prepare_delegate_master_prompt(prompt, ["11"])
+        finally:
+            pt._DELEGATE_MASTER_NODE_CLASS_MAPPINGS = previous_mappings
+
+        self.assertIn("15", result)
+        self.assertIn("16", result)
+        self.assertIn("17", result)
+        self.assertEqual(result["9"]["inputs"]["filename_prefix"], ["17", 0])
+        self.assertEqual(result["17"]["inputs"]["string_a"], ["15", 0])
+        self.assertEqual(result["17"]["inputs"]["string_b"], ["16", 0])
+
+    def test_preserves_registered_multi_string_join_subgraph_for_downstream_required_input(self):
+        """Delegate-only master keeps multi-input scalar utility chains."""
+        join_string_multi = type(
+            "JoinStringMulti",
+            (),
+            {
+                "RETURN_TYPES": ("STRING",),
+                "INPUT_TYPES": classmethod(
+                    lambda cls: {
+                        "required": {"string_1": ("STRING",)},
+                        "optional": {
+                            "string_2": ("STRING",),
+                            "string_3": ("STRING",),
+                        },
+                    }
+                ),
+            },
+        )
+        save_image = type(
+            "SaveImage",
+            (),
+            {
+                "INPUT_TYPES": classmethod(
+                    lambda cls: {
+                        "required": {
+                            "images": ("IMAGE",),
+                            "filename_prefix": ("STRING",),
+                        }
+                    }
+                )
+            },
+        )
+        previous_mappings = getattr(pt, "_DELEGATE_MASTER_NODE_CLASS_MAPPINGS", None)
+        pt._DELEGATE_MASTER_NODE_CLASS_MAPPINGS = {
+            "JoinStringMulti": join_string_multi,
+            "SaveImage": save_image,
+        }
+        try:
+            prompt = {
+                "8": {"class_type": "KSampler", "inputs": {}},
+                "11": {"class_type": "DistributedCollector", "inputs": {"images": ["8", 0]}},
+                "15": {"class_type": "PrimitiveString", "inputs": {"value": "test"}},
+                "16": {"class_type": "PrimitiveString", "inputs": {"value": "input"}},
+                "17": {"class_type": "PrimitiveString", "inputs": {"value": "bug4"}},
+                "18": {
+                    "class_type": "JoinStringMulti",
+                    "inputs": {
+                        "string_1": ["15", 0],
+                        "string_2": ["16", 0],
+                        "string_3": ["17", 0],
+                    },
+                },
+                "9": {
+                    "class_type": "SaveImage",
+                    "inputs": {
+                        "images": ["11", 0],
+                        "filename_prefix": ["18", 0],
+                    },
+                },
+            }
+
+            result = pt.prepare_delegate_master_prompt(prompt, ["11"])
+        finally:
+            pt._DELEGATE_MASTER_NODE_CLASS_MAPPINGS = previous_mappings
+
+        self.assertIn("15", result)
+        self.assertIn("16", result)
+        self.assertIn("17", result)
+        self.assertIn("18", result)
+        self.assertEqual(result["9"]["inputs"]["filename_prefix"], ["18", 0])
+
+    def test_does_not_preserve_scalar_utility_with_heavy_upstream_dependency(self):
+        """Scalar utility nodes are retained only when their full input branch is safe."""
+        string_concat = type(
+            "StringConcatenate",
+            (),
+            {
+                "RETURN_TYPES": ("STRING",),
+                "INPUT_TYPES": classmethod(
+                    lambda cls: {
+                        "required": {
+                            "string_a": ("STRING",),
+                            "string_b": ("STRING",),
+                        }
+                    }
+                ),
+            },
+        )
+        save_image = type(
+            "SaveImage",
+            (),
+            {
+                "INPUT_TYPES": classmethod(
+                    lambda cls: {
+                        "required": {
+                            "images": ("IMAGE",),
+                            "filename_prefix": ("STRING",),
+                        }
+                    }
+                )
+            },
+        )
+        previous_mappings = getattr(pt, "_DELEGATE_MASTER_NODE_CLASS_MAPPINGS", None)
+        pt._DELEGATE_MASTER_NODE_CLASS_MAPPINGS = {
+            "SaveImage": save_image,
+            "StringConcatenate": string_concat,
+        }
+        try:
+            prompt = {
+                "8": {"class_type": "KSampler", "inputs": {}},
+                "11": {"class_type": "DistributedCollector", "inputs": {"images": ["8", 0]}},
+                "15": {"class_type": "PrimitiveString", "inputs": {"value": "test"}},
+                "17": {
+                    "class_type": "StringConcatenate",
+                    "inputs": {
+                        "string_a": ["15", 0],
+                        "string_b": ["8", 0],
+                    },
+                },
+                "9": {
+                    "class_type": "SaveImage",
+                    "inputs": {
+                        "images": ["11", 0],
+                        "filename_prefix": ["17", 0],
+                    },
+                },
+            }
+
+            result = pt.prepare_delegate_master_prompt(prompt, ["11"])
+        finally:
+            pt._DELEGATE_MASTER_NODE_CLASS_MAPPINGS = previous_mappings
+
+        self.assertNotIn("8", result)
+        self.assertNotIn("17", result)
+        self.assertNotIn("filename_prefix", result["9"]["inputs"])
+
+    def test_does_not_preserve_scalar_output_for_non_scalar_downstream_input(self):
+        """Scalar outputs are retained only for scalar/config downstream inputs."""
+        string_provider = type("StringProvider", (), {"RETURN_TYPES": ("STRING",)})
+        image_consumer = type(
+            "ImageConsumer",
+            (),
+            {
+                "INPUT_TYPES": classmethod(
+                    lambda cls: {
+                        "required": {
+                            "images": ("IMAGE",),
+                            "mask": ("IMAGE",),
+                        }
+                    }
+                )
+            },
+        )
+        previous_mappings = getattr(pt, "_DELEGATE_MASTER_NODE_CLASS_MAPPINGS", None)
+        pt._DELEGATE_MASTER_NODE_CLASS_MAPPINGS = {
+            "ImageConsumer": image_consumer,
+            "StringProvider": string_provider,
+        }
+        try:
+            prompt = {
+                "8": {"class_type": "KSampler", "inputs": {}},
+                "11": {"class_type": "DistributedCollector", "inputs": {"images": ["8", 0]}},
+                "17": {"class_type": "StringProvider", "inputs": {}},
+                "9": {
+                    "class_type": "ImageConsumer",
+                    "inputs": {
+                        "images": ["11", 0],
+                        "mask": ["17", 0],
+                    },
+                },
+            }
+
+            result = pt.prepare_delegate_master_prompt(prompt, ["11"])
+        finally:
+            pt._DELEGATE_MASTER_NODE_CLASS_MAPPINGS = previous_mappings
+
+        self.assertNotIn("17", result)
+        self.assertNotIn("mask", result["9"]["inputs"])
+
+    def test_preserves_scalar_list_join_subgraph_for_downstream_required_input(self):
+        """Delegate-only master keeps list-shaped scalar config chains."""
+        create_list = type(
+            "CreateList",
+            (),
+            {
+                "RETURN_TYPES": ("LIST",),
+                "INPUT_TYPES": classmethod(
+                    lambda cls: {
+                        "required": {"inputs.input0": ("STRING",)},
+                        "optional": {
+                            "inputs.input1": ("STRING",),
+                            "inputs.input2": ("STRING",),
+                        },
+                    }
+                ),
+            },
+        )
+        string_data_list_join = type(
+            "StringDataListJoin",
+            (),
+            {
+                "RETURN_TYPES": ("STRING",),
+                "INPUT_TYPES": classmethod(
+                    lambda cls: {
+                        "required": {
+                            "strings": ("LIST",),
+                            "delimiter": ("STRING",),
+                        }
+                    }
+                ),
+            },
+        )
+        save_image = type(
+            "SaveImage",
+            (),
+            {
+                "INPUT_TYPES": classmethod(
+                    lambda cls: {
+                        "required": {
+                            "images": ("IMAGE",),
+                            "filename_prefix": ("STRING",),
+                        }
+                    }
+                )
+            },
+        )
+        previous_mappings = getattr(pt, "_DELEGATE_MASTER_NODE_CLASS_MAPPINGS", None)
+        pt._DELEGATE_MASTER_NODE_CLASS_MAPPINGS = {
+            "CreateList": create_list,
+            "SaveImage": save_image,
+            "StringDataListJoin": string_data_list_join,
+        }
+        try:
+            prompt = {
+                "8": {"class_type": "KSampler", "inputs": {}},
+                "11": {"class_type": "DistributedCollector", "inputs": {"images": ["8", 0]}},
+                "15": {"class_type": "PrimitiveString", "inputs": {"value": "test"}},
+                "17": {"class_type": "PrimitiveString", "inputs": {"value": "input_bug"}},
+                "29": {"class_type": "PrimitiveString", "inputs": {"value": "new"}},
+                "28": {
+                    "class_type": "CreateList",
+                    "inputs": {
+                        "inputs.input0": ["15", 0],
+                        "inputs.input1": ["17", 0],
+                        "inputs.input2": ["29", 0],
+                    },
+                },
+                "32": {
+                    "class_type": "StringDataListJoin",
+                    "inputs": {
+                        "strings": ["28", 0],
+                        "delimiter": "/",
+                    },
+                },
+                "9": {
+                    "class_type": "SaveImage",
+                    "inputs": {
+                        "images": ["11", 0],
+                        "filename_prefix": ["32", 0],
+                    },
+                },
+            }
+
+            result = pt.prepare_delegate_master_prompt(prompt, ["11"])
+        finally:
+            pt._DELEGATE_MASTER_NODE_CLASS_MAPPINGS = previous_mappings
+
+        self.assertIn("15", result)
+        self.assertIn("17", result)
+        self.assertIn("28", result)
+        self.assertIn("29", result)
+        self.assertIn("32", result)
+        self.assertEqual(result["9"]["inputs"]["filename_prefix"], ["32", 0])
+        self.assertEqual(result["32"]["inputs"]["strings"], ["28", 0])
+        self.assertEqual(result["28"]["inputs"]["inputs.input0"], ["15", 0])
+        self.assertNotIn("8", result)
+
+    def test_preserves_builtin_create_list_string_data_list_join_subgraph(self):
+        """Delegate-only master handles ComfyUI 0.23 CreateList data-list output."""
+        string_data_list_join = type(
+            "StringDataListJoin",
+            (),
+            {
+                "RETURN_TYPES": ("STRING",),
+                "INPUT_IS_LIST": True,
+                "INPUT_TYPES": classmethod(
+                    lambda cls: {
+                        "required": {
+                            "strings": ("STRING", {"forceInput": True}),
+                            "sep": ("STRING", {"default": " "}),
+                        }
+                    }
+                ),
+            },
+        )
+        save_image = type(
+            "SaveImage",
+            (),
+            {
+                "INPUT_TYPES": classmethod(
+                    lambda cls: {
+                        "required": {
+                            "images": ("IMAGE",),
+                            "filename_prefix": ("STRING",),
+                        }
+                    }
+                )
+            },
+        )
+        previous_mappings = getattr(pt, "_DELEGATE_MASTER_NODE_CLASS_MAPPINGS", None)
+        pt._DELEGATE_MASTER_NODE_CLASS_MAPPINGS = {
+            "Basic data handling: StringDataListJoin": string_data_list_join,
+            "SaveImage": save_image,
+        }
+        try:
+            prompt = {
+                "8": {"class_type": "KSampler", "inputs": {}},
+                "11": {"class_type": "DistributedCollector", "inputs": {"images": ["8", 0]}},
+                "15": {"class_type": "PrimitiveString", "inputs": {"value": "input_bug"}},
+                "17": {"class_type": "PrimitiveString", "inputs": {"value": "test"}},
+                "29": {"class_type": "PrimitiveString", "inputs": {"value": "new"}},
+                "28": {
+                    "class_type": "CreateList",
+                    "inputs": {
+                        "inputs.input0": ["17", 0],
+                        "inputs.input1": ["15", 0],
+                        "inputs.input2": ["29", 0],
+                    },
+                },
+                "32": {
+                    "class_type": "Basic data handling: StringDataListJoin",
+                    "inputs": {
+                        "strings": ["28", 0],
+                        "sep": "/",
+                    },
+                },
+                "9": {
+                    "class_type": "SaveImage",
+                    "inputs": {
+                        "images": ["11", 0],
+                        "filename_prefix": ["32", 0],
+                    },
+                },
+            }
+
+            result = pt.prepare_delegate_master_prompt(prompt, ["11"])
+        finally:
+            pt._DELEGATE_MASTER_NODE_CLASS_MAPPINGS = previous_mappings
+
+        for node_id in ("15", "17", "28", "29", "32"):
+            self.assertIn(node_id, result)
+        self.assertEqual(result["9"]["inputs"]["filename_prefix"], ["32", 0])
+        self.assertEqual(result["32"]["inputs"]["strings"], ["28", 0])
+        self.assertEqual(result["28"]["inputs"]["inputs.input0"], ["17", 0])
+        self.assertNotIn("8", result)
+
     def test_result_is_independent_copy(self):
         prompt = _delegate_prompt()
         result = pt.prepare_delegate_master_prompt(prompt, ["3"])

tests/test_worker_process_runtime.py

@@ -0,0 +1,134 @@
+import importlib.util
+import sys
+import types
+import unittest
+from argparse import Namespace
+from pathlib import Path
+from unittest.mock import patch
+
+
+def _load_process_module(module_filename: str):
+    module_path = Path(__file__).resolve().parents[1] / "workers" / "process" / module_filename
+    package_name = "dist_proc_testpkg"
+    module_name = module_filename[:-3]
+
+    for mod_name in list(sys.modules):
+        if mod_name == package_name or mod_name.startswith(f"{package_name}."):
+            del sys.modules[mod_name]
+
+    root_pkg = types.ModuleType(package_name)
+    root_pkg.__path__ = []
+    sys.modules[package_name] = root_pkg
+
+    workers_pkg = types.ModuleType(f"{package_name}.workers")
+    workers_pkg.__path__ = []
+    sys.modules[f"{package_name}.workers"] = workers_pkg
+
+    process_pkg = types.ModuleType(f"{package_name}.workers.process")
+    process_pkg.__path__ = []
+    sys.modules[f"{package_name}.workers.process"] = process_pkg
+
+    utils_pkg = types.ModuleType(f"{package_name}.utils")
+    utils_pkg.__path__ = []
+    sys.modules[f"{package_name}.utils"] = utils_pkg
+
+    logging_module = types.ModuleType(f"{package_name}.utils.logging")
+    logging_module.debug_log = lambda *_args, **_kwargs: None
+    logging_module.log = lambda *_args, **_kwargs: None
+    sys.modules[f"{package_name}.utils.logging"] = logging_module
+
+    process_module = types.ModuleType(f"{package_name}.utils.process")
+    process_module.get_python_executable = lambda: "/usr/bin/test-python"
+    sys.modules[f"{package_name}.utils.process"] = process_module
+
+    spec = importlib.util.spec_from_file_location(
+        f"{package_name}.workers.process.{module_name}",
+        module_path,
+    )
+    module = importlib.util.module_from_spec(spec)
+    assert spec is not None and spec.loader is not None
+    spec.loader.exec_module(module)
+    return module
+
+
+root_discovery_module = _load_process_module("root_discovery.py")
+launch_builder_module = _load_process_module("launch_builder.py")
+
+
+class ComfyRootDiscoveryTests(unittest.TestCase):
+    def test_prefers_loaded_comfyui_module_path(self):
+        discovery = root_discovery_module.ComfyRootDiscovery()
+        server_module = types.SimpleNamespace(__file__="/opt/ComfyUI/server.py")
+
+        def fake_exists(path):
+            return path == "/opt/ComfyUI/main.py"
+
+        with patch.dict(sys.modules, {"server": server_module}, clear=False), \
+             patch.object(root_discovery_module.os.path, "exists", side_effect=fake_exists), \
+             patch.dict(root_discovery_module.os.environ, {}, clear=True):
+            self.assertEqual(discovery.find_comfy_root(), "/opt/ComfyUI")
+
+
+class LaunchCommandBuilderTests(unittest.TestCase):
+    def test_inherits_runtime_layout_args_for_desktop(self):
+        builder = launch_builder_module.LaunchCommandBuilder()
+        runtime_args = Namespace(
+            listen="127.0.0.1",
+            base_directory="C:/Users/test/ComfyUI",
+            temp_directory=None,
+            input_directory="C:/Users/test/ComfyUI/input",
+            output_directory="C:/Users/test/ComfyUI/output",
+            user_directory="C:/Users/test/ComfyUI/user",
+            front_end_root="C:/Program Files/ComfyUI/web_custom_versions/desktop_app",
+            extra_model_paths_config=[["C:/Users/test/AppData/Roaming/ComfyUI/extra_models_config.yaml"]],
+            enable_manager=True,
+            disable_manager_ui=False,
+            enable_manager_legacy_ui=False,
+            windows_standalone_build=True,
+            log_stdout=True,
+            verbose="INFO",
+            enable_cors_header="*",
+        )
+        comfy_module = types.ModuleType("comfy")
+        comfy_cli_args = types.ModuleType("comfy.cli_args")
+        comfy_cli_args.args = runtime_args
+
+        worker_config = {
+            "port": 9001,
+            "extra_args": "--preview-method auto",
+        }
+
+        def fake_exists(path):
+            return path == "/desktop/ComfyUI/main.py"
+
+        with patch.dict(
+            sys.modules,
+            {"comfy": comfy_module, "comfy.cli_args": comfy_cli_args},
+            clear=False,
+        ), patch.object(launch_builder_module.os.path, "exists", side_effect=fake_exists):
+            cmd = builder.build_launch_command(worker_config, "/desktop/ComfyUI")
+
+        self.assertEqual(cmd[:2], ["/usr/bin/test-python", "/desktop/ComfyUI/main.py"])
+        self.assertIn("--listen", cmd)
+        self.assertIn("127.0.0.1", cmd)
+        self.assertIn("--base-directory", cmd)
+        self.assertIn("C:/Users/test/ComfyUI", cmd)
+        self.assertIn("--input-directory", cmd)
+        self.assertIn("--output-directory", cmd)
+        self.assertIn("--user-directory", cmd)
+        self.assertIn("--front-end-root", cmd)
+        self.assertIn("--extra-model-paths-config", cmd)
+        self.assertIn("C:/Users/test/AppData/Roaming/ComfyUI/extra_models_config.yaml", cmd)
+        self.assertIn("--enable-manager", cmd)
+        self.assertIn("--windows-standalone-build", cmd)
+        self.assertIn("--log-stdout", cmd)
+        self.assertIn("--disable-auto-launch", cmd)
+        self.assertIn("--enable-cors-header", cmd)
+        self.assertIn("*", cmd)
+        self.assertIn("--port", cmd)
+        self.assertIn("9001", cmd)
+        self.assertNotIn("--auto-launch", cmd)
+
+
+if __name__ == "__main__":
+    unittest.main()

utils/async_helpers.py

@@ -3,6 +3,7 @@ Async helper utilities for ComfyUI-Distributed.
 """
 import asyncio
 import threading
+import time
 import uuid
 import execution
 import server
@@ -104,7 +105,12 @@ class PromptValidationError(RuntimeError):
         super().__init__(f"Invalid prompt: {merged}")
 
 
-async def queue_prompt_payload(prompt_obj, workflow_meta=None, client_id=None):
+async def queue_prompt_payload(
+    prompt_obj,
+    workflow_meta=None,
+    client_id=None,
+    include_queue_metadata=False,
+):
     """Validate and queue a prompt via ComfyUI's prompt queue."""
     payload = {"prompt": prompt_obj}
     payload = prompt_server.trigger_on_prompt(payload)
@@ -117,7 +123,7 @@ async def queue_prompt_payload(prompt_obj, workflow_meta=None, client_id=None):
         node_errors = valid[3] if len(valid) > 3 else {}
         raise PromptValidationError(error_payload, node_errors)
 
-    extra_data = {}
+    extra_data = {"create_time": int(time.time() * 1000)}
     if workflow_meta:
         extra_data.setdefault("extra_pnginfo", {})["workflow"] = workflow_meta
     if client_id:
@@ -132,4 +138,12 @@ async def queue_prompt_payload(prompt_obj, workflow_meta=None, client_id=None):
     prompt_server.number = number + 1
     prompt_queue_item = (number, prompt_id, prompt, extra_data, valid[2], sensitive)
     prompt_server.prompt_queue.put(prompt_queue_item)
+
+    if include_queue_metadata:
+        return {
+            "prompt_id": prompt_id,
+            "number": number,
+            "node_errors": {},
+        }
+
     return prompt_id

utils/network.py

@@ -66,6 +66,25 @@ def normalize_host(value):
     return host.split("/")[0]
 
 
+def _split_host_and_port(host):
+    if not host:
+        return host, None
+
+    if host.startswith("["):
+        match = re.match(r"^(\[[^\]]+\])(?::(\d+))?$", host)
+        if match:
+            parsed_port = int(match.group(2)) if match.group(2) else None
+            return match.group(1), parsed_port
+        return host, None
+
+    if host.count(":") == 1:
+        candidate_host, candidate_port = host.rsplit(":", 1)
+        if candidate_port.isdigit():
+            return candidate_host, int(candidate_port)
+
+    return host, None
+
+
 def build_worker_url(worker, endpoint=""):
     """Construct the worker base URL with optional endpoint."""
     host = (worker.get("host") or "").strip()
@@ -126,12 +145,7 @@ def build_master_url(config=None, prompt_server_instance=None):
     prompt_server_instance = prompt_server_instance or server.PromptServer.instance
     master_cfg = (config or {}).get("master", {}) or {}
     configured_host = (master_cfg.get("host") or "").strip()
-    configured_port = master_cfg.get("port")
-    default_port = getattr(prompt_server_instance, "port", 8188) or 8188
-    try:
-        port = int(configured_port or default_port)
-    except (TypeError, ValueError):
-        port = int(default_port)
+    runtime_port = getattr(prompt_server_instance, "port", 8188) or 8188
 
     def _needs_https(hostname):
         hostname = hostname.lower()
@@ -149,10 +163,11 @@ def build_master_url(config=None, prompt_server_instance=None):
         if configured_host.startswith(("http://", "https://")):
             return configured_host.rstrip("/")
 
-        host = configured_host
+        host, explicit_port = _split_host_and_port(configured_host)
+        port = explicit_port if explicit_port is not None else int(runtime_port)
         scheme = "https" if _needs_https(host) or port == 443 else "http"
         default_port_for_scheme = 443 if scheme == "https" else 80
-        if configured_port is None and scheme == "https" and _needs_https(host):
+        if explicit_port is None and scheme == "https" and _needs_https(host):
             port = default_port_for_scheme
         port_part = "" if port == default_port_for_scheme else f":{port}"
         return f"{scheme}://{host}{port_part}"
@@ -160,7 +175,27 @@ def build_master_url(config=None, prompt_server_instance=None):
     address = getattr(prompt_server_instance, "address", "127.0.0.1") or "127.0.0.1"
     if address in ("0.0.0.0", "::"):
         address = "127.0.0.1"
+    port = int(runtime_port)
     scheme = "https" if port == 443 else "http"
     default_port_for_scheme = 443 if scheme == "https" else 80
     port_part = "" if port == default_port_for_scheme else f":{port}"
     return f"{scheme}://{address}{port_part}"
+
+
+def build_master_callback_url(worker, config=None, prompt_server_instance=None):
+    """Build the callback URL a specific worker should use to reach the master."""
+    prompt_server_instance = prompt_server_instance or server.PromptServer.instance
+
+    worker_type = str((worker or {}).get("type") or "").strip().lower()
+    worker_host = normalize_host((worker or {}).get("host"))
+    local_hosts = {"", "localhost", "127.0.0.1", "::1", "[::1]", "0.0.0.0"}
+
+    is_local_worker = worker_type == "local" or worker_host in local_hosts
+    if is_local_worker:
+        port = int(getattr(prompt_server_instance, "port", 8188) or 8188)
+        scheme = "https" if port == 443 else "http"
+        default_port_for_scheme = 443 if scheme == "https" else 80
+        port_part = "" if port == default_port_for_scheme else f":{port}"
+        return f"{scheme}://127.0.0.1{port_part}"
+
+    return build_master_url(config=config, prompt_server_instance=prompt_server_instance)

web/apiClient.js

@@ -178,11 +178,32 @@ export function createApiClient(baseUrl) {
                     return { ok: false, status: response.status, queueRemaining: null };
                 }
 
-                const data = await response.json().catch(() => ({}));
+                let data;
+                try {
+                    data = await response.json();
+                } catch {
+                    return { ok: false, status: response.status, queueRemaining: null };
+                }
+
+                if (!data || typeof data !== "object" || Array.isArray(data)) {
+                    return { ok: false, status: response.status, queueRemaining: null };
+                }
+
+                const execInfo = data.exec_info;
+                if (!execInfo || typeof execInfo !== "object" || Array.isArray(execInfo)) {
+                    return { ok: false, status: response.status, queueRemaining: null };
+                }
+
+                const rawQueueRemaining = execInfo.queue_remaining;
+                const queueRemaining = Number(rawQueueRemaining);
+                if (!Number.isFinite(queueRemaining)) {
+                    return { ok: false, status: response.status, queueRemaining: null };
+                }
+
                 return {
                     ok: true,
                     status: response.status,
-                    queueRemaining: data.exec_info?.queue_remaining || 0,
+                    queueRemaining: Math.max(0, queueRemaining),
                 };
             } finally {
                 clearTimeout(timeoutId);

web/executionUtils.js

@@ -4,7 +4,7 @@ import { TIMEOUTS, NODE_CLASSES, generateUUID } from './constants.js';
 import { checkAllWorkerStatuses, getWorkerUrl } from './workerLifecycle.js';
 
 export function setupInterceptor(extension) {
-    api.queuePrompt = async (number, prompt) => {
+    api.queuePrompt = async (number, prompt, ...rest) => {
         if (extension.isEnabled) {
             const hasCollector = findNodesByClass(prompt.output, NODE_CLASSES.DISTRIBUTED_COLLECTOR).length > 0;
             const hasDistUpscale = findNodesByClass(prompt.output, NODE_CLASSES.UPSCALE_DISTRIBUTED).length > 0;
@@ -18,7 +18,7 @@ export function setupInterceptor(extension) {
                 return result;
             }
         }
-        return extension.originalQueuePrompt(number, prompt);
+        return extension.originalQueuePrompt(number, prompt, ...rest);
     };
 }
 

web/tests/apiClient.test.js

@@ -0,0 +1,95 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+
+import { createApiClient } from "../apiClient.js";
+
+describe("apiClient probeWorker", () => {
+    let originalFetch;
+
+    beforeEach(() => {
+        originalFetch = globalThis.fetch;
+        globalThis.fetch = vi.fn();
+    });
+
+    afterEach(() => {
+        globalThis.fetch = originalFetch;
+        vi.restoreAllMocks();
+    });
+
+    it("returns ok=true when /prompt returns valid exec_info payload", async () => {
+        globalThis.fetch.mockResolvedValue({
+            ok: true,
+            status: 200,
+            json: vi.fn().mockResolvedValue({ exec_info: { queue_remaining: 2 } }),
+        });
+
+        const client = createApiClient("http://127.0.0.1:8188");
+        const result = await client.probeWorker("http://worker.local:8190", 1000);
+
+        expect(result).toEqual({ ok: true, status: 200, queueRemaining: 2 });
+    });
+
+    it("returns ok=false on non-200 responses", async () => {
+        globalThis.fetch.mockResolvedValue({
+            ok: false,
+            status: 503,
+            json: vi.fn(),
+        });
+
+        const client = createApiClient("http://127.0.0.1:8188");
+        const result = await client.probeWorker("http://worker.local:8190", 1000);
+
+        expect(result).toEqual({ ok: false, status: 503, queueRemaining: null });
+    });
+
+    it("returns ok=false when response JSON is invalid", async () => {
+        globalThis.fetch.mockResolvedValue({
+            ok: true,
+            status: 200,
+            json: vi.fn().mockRejectedValue(new Error("invalid json")),
+        });
+
+        const client = createApiClient("http://127.0.0.1:8188");
+        const result = await client.probeWorker("http://worker.local:8190", 1000);
+
+        expect(result).toEqual({ ok: false, status: 200, queueRemaining: null });
+    });
+
+    it("returns ok=false when exec_info is missing", async () => {
+        globalThis.fetch.mockResolvedValue({
+            ok: true,
+            status: 200,
+            json: vi.fn().mockResolvedValue({}),
+        });
+
+        const client = createApiClient("http://127.0.0.1:8188");
+        const result = await client.probeWorker("http://worker.local:8190", 1000);
+
+        expect(result).toEqual({ ok: false, status: 200, queueRemaining: null });
+    });
+
+    it("returns ok=false when queue_remaining is not numeric", async () => {
+        globalThis.fetch.mockResolvedValue({
+            ok: true,
+            status: 200,
+            json: vi.fn().mockResolvedValue({ exec_info: { queue_remaining: "n/a" } }),
+        });
+
+        const client = createApiClient("http://127.0.0.1:8188");
+        const result = await client.probeWorker("http://worker.local:8190", 1000);
+
+        expect(result).toEqual({ ok: false, status: 200, queueRemaining: null });
+    });
+
+    it("clamps negative queue_remaining to zero", async () => {
+        globalThis.fetch.mockResolvedValue({
+            ok: true,
+            status: 200,
+            json: vi.fn().mockResolvedValue({ exec_info: { queue_remaining: -5 } }),
+        });
+
+        const client = createApiClient("http://127.0.0.1:8188");
+        const result = await client.probeWorker("http://worker.local:8190", 1000);
+
+        expect(result).toEqual({ ok: true, status: 200, queueRemaining: 0 });
+    });
+});

workers/process/launch_builder.py

@@ -10,6 +10,62 @@ from ...utils.process import get_python_executable
 class LaunchCommandBuilder:
     """Build command-lines for launching worker ComfyUI processes."""
 
+    def _extend_arg(self, cmd, flag, value):
+        if value in (None, "", [], ()):
+            return
+        cmd.extend([flag, str(value)])
+
+    def _extend_grouped_args(self, cmd, flag, values):
+        for group in values or []:
+            flattened = [str(item) for item in group if item]
+            if flattened:
+                cmd.append(flag)
+                cmd.extend(flattened)
+
+    def _get_runtime_args(self):
+        try:
+            from comfy.cli_args import args
+            return args
+        except Exception as exc:
+            debug_log(f"Could not read current ComfyUI CLI args for worker launch: {exc}")
+            return None
+
+    def _build_runtime_launch_args(self):
+        args = self._get_runtime_args()
+        if args is None:
+            return []
+
+        inherited = []
+        self._extend_arg(inherited, "--listen", getattr(args, "listen", None))
+        self._extend_arg(inherited, "--base-directory", getattr(args, "base_directory", None))
+        self._extend_arg(inherited, "--temp-directory", getattr(args, "temp_directory", None))
+        self._extend_arg(inherited, "--input-directory", getattr(args, "input_directory", None))
+        self._extend_arg(inherited, "--output-directory", getattr(args, "output_directory", None))
+        self._extend_arg(inherited, "--user-directory", getattr(args, "user_directory", None))
+        self._extend_arg(inherited, "--front-end-root", getattr(args, "front_end_root", None))
+        self._extend_grouped_args(
+            inherited,
+            "--extra-model-paths-config",
+            getattr(args, "extra_model_paths_config", None),
+        )
+
+        if getattr(args, "enable_manager", False):
+            inherited.append("--enable-manager")
+        if getattr(args, "disable_manager_ui", False):
+            inherited.append("--disable-manager-ui")
+        if getattr(args, "enable_manager_legacy_ui", False):
+            inherited.append("--enable-manager-legacy-ui")
+        if getattr(args, "windows_standalone_build", False):
+            inherited.append("--windows-standalone-build")
+        if getattr(args, "log_stdout", False):
+            inherited.append("--log-stdout")
+
+        verbose = getattr(args, "verbose", None)
+        if verbose and verbose != "INFO":
+            inherited.extend(["--verbose", str(verbose)])
+
+        return inherited
+
     def _find_windows_terminal(self):
         """Find Windows Terminal executable."""
         possible_paths = [
@@ -39,10 +95,19 @@ class LaunchCommandBuilder:
             cmd = [
                 get_python_executable(),
                 main_py,
-                "--port",
-                str(worker_config["port"]),
-                "--enable-cors-header",
             ]
+            cmd.extend(self._build_runtime_launch_args())
+            cmd.extend(["--port", str(worker_config["port"])])
+
+            current_args = self._get_runtime_args()
+            current_cors = getattr(current_args, "enable_cors_header", None) if current_args else None
+            cmd.append("--enable-cors-header")
+            if current_cors is not None:
+                cmd.append(str(current_cors))
+
+            if "--disable-auto-launch" not in cmd:
+                cmd.append("--disable-auto-launch")
+
             debug_log(f"Using main.py: {main_py}")
         else:
             error_msg = f"Could not find main.py in {comfy_root}\n"

workers/process/lifecycle.py

@@ -33,6 +33,7 @@ class ProcessLifecycle:
         env["CUDA_VISIBLE_DEVICES"] = str(worker_config.get("cuda_device", 0))
         env["PYTORCH_CUDA_ALLOC_CONF"] = "expandable_segments:True"
         env["COMFYUI_MASTER_PID"] = str(os.getpid())
+        env["COMFYUI_IS_WORKER"] = "1"
 
         cmd = self._manager.build_launch_command(worker_config, comfy_root)
         cwd = comfy_root

workers/process/root_discovery.py

@@ -1,4 +1,5 @@
 import os
+import sys
 
 from ...utils.logging import debug_log, log
 
@@ -6,6 +7,21 @@ from ...utils.logging import debug_log, log
 class ComfyRootDiscovery:
     """Resolve the ComfyUI root directory across local and container layouts."""
 
+    def _find_root_from_loaded_modules(self):
+        """Use already-imported ComfyUI modules to locate the runtime root."""
+        for module_name in ("server", "folder_paths", "main"):
+            module = sys.modules.get(module_name)
+            module_file = getattr(module, "__file__", None)
+            if not module_file:
+                continue
+
+            candidate = os.path.dirname(os.path.abspath(module_file))
+            if os.path.exists(os.path.join(candidate, "main.py")):
+                debug_log(f"Found ComfyUI root via loaded module {module_name}: {candidate}")
+                return candidate
+
+        return None
+
     def find_comfy_root(self):
         # Start from current file location.
         current_dir = os.path.dirname(os.path.abspath(__file__))
@@ -17,12 +33,17 @@ class ComfyRootDiscovery:
             debug_log(f"Found ComfyUI root via COMFYUI_ROOT environment variable: {env_root}")
             return env_root
 
-        # Method 2: Try going up from custom_nodes directory.
+        # Method 2: Inspect the already-loaded ComfyUI runtime modules.
+        runtime_root = self._find_root_from_loaded_modules()
+        if runtime_root:
+            return runtime_root
+
+        # Method 3: Try going up from custom_nodes directory.
         if os.path.exists(os.path.join(potential_root, "main.py")):
             debug_log(f"Found ComfyUI root via directory traversal: {potential_root}")
             return potential_root
 
-        # Method 3: Look for common Docker paths.
+        # Method 4: Look for common Docker paths.
         docker_paths = [
             "/basedir",
             "/ComfyUI",
@@ -37,7 +58,7 @@ class ComfyRootDiscovery:
                 debug_log(f"Found ComfyUI root in Docker path: {path}")
                 return path
 
-        # Method 4: Search upwards for main.py.
+        # Method 5: Search upwards for main.py.
         search_dir = current_dir
         for _ in range(5):
             if os.path.exists(os.path.join(search_dir, "main.py")):
@@ -48,7 +69,7 @@ class ComfyRootDiscovery:
                 break
             search_dir = parent
 
-        # Method 5: Try to import and use folder_paths.
+        # Method 6: Try to import and use folder_paths.
         try:
             import folder_paths