feat: integrate native image-list caching and exact serverless value overrides

api/image_list_routes.py

@@ -37,8 +37,9 @@ async def image_list_capabilities_endpoint(_request):
     return web.json_response({
         "supported": True,
         "execution_targets": True,
+        "value_results": True,
         "protocol_version": PROTOCOL_VERSION,
-        "node_ids": ["DistributedImageListStart", "DistributedImageListEnd"],
+        "node_ids": ["DistributedImageListStart", "DistributedImageListEnd", "DistributedImageListGetValue"],
         "transport": "typed-float-tensor-json-v1",
         "image_transfer_formats": list(IMAGE_TRANSFER_FORMATS),
     })
@@ -54,8 +55,9 @@ async def image_list_item_endpoint(request):
         if not isinstance(context_id, str) or not context_id:
             raise ValueError("context_id must be a non-empty string")
         item = data.get("item")
-        store_worker_item(data.get("token"), item)
-        return web.json_response({"status": "stored", "item_index": item["index"]})
+        identity = store_worker_item(data.get("token"), item)
+        return web.json_response({"status": "stored", "item_index": item["index"],
+                                  "item_identity": identity})
     except (ValueError, TypeError) as exc:
         return web.json_response({"error": str(exc)}, status=400)
     except Exception as exc:

api/job_routes.py

@@ -37,6 +37,8 @@ prompt_server = server.PromptServer.instance
 #   envelope: JSON { "job_id", "worker_id", "format", "start_index", "count", "width", "height",
 #                    "is_last", "audio"?: {...} }
 #   media:    one part per frame for still formats, or one video file for video formats
+#   format "exact" adds envelope "descriptors" (one tensor-le-v1 descriptor per
+#   frame) and sends one raw tensor part per frame; values are not quantized.
 
 
 def _decode_image_sync(image_path):
@@ -299,6 +301,8 @@ async def _enqueue_worker_items(multi_job_id, items):
 
 def _decode_compressed_chunk_sync(envelope, parts):
     """Decode a compressed chunk into per-frame IMAGE tensors (threadpool)."""
+    if envelope.get("format") == transfer_codec.EXACT_FORMAT:
+        return _decode_exact_chunk_sync(envelope, parts)
     frames = transfer_codec.decode_chunk(envelope, parts)
     return [
         ensure_contiguous(torch.from_numpy(transfer_codec.uint8_frame_to_tensor_array(frame)))
@@ -306,6 +310,35 @@ def _decode_compressed_chunk_sync(envelope, parts):
     ]
 
 
+def _decode_exact_chunk_sync(envelope, parts):
+    """Decode raw exact IMAGE frames, checking each against the envelope geometry."""
+    from ..serverless_bridge import codec as bridge_codec
+    count, width, height = envelope.get("count"), envelope.get("width"), envelope.get("height")
+    for name, value in (("count", count), ("width", width), ("height", height)):
+        if not isinstance(value, int) or isinstance(value, bool) or value <= 0:
+            raise transfer_codec.TransferCodecError(f"envelope.{name}: expected positive integer")
+    descriptors = envelope.get("descriptors")
+    if not isinstance(descriptors, list) or len(descriptors) != count or len(parts) != count:
+        raise transfer_codec.TransferCodecError(
+            f"exact chunk expected {count} descriptors and frames, got "
+            f"{len(descriptors) if isinstance(descriptors, list) else 'no'} descriptors and {len(parts)} frames"
+        )
+    tensors = []
+    for descriptor, payload in zip(descriptors, parts):
+        if not isinstance(descriptor, dict) or descriptor.get("encoding") != "tensor-le-v1":
+            raise transfer_codec.TransferCodecError("exact frames require tensor-le-v1 encoding")
+        shape = descriptor.get("shape") if isinstance(descriptor, dict) else None
+        if not isinstance(shape, list) or len(shape) != 4 or shape[:3] != [1, height, width]:
+            raise transfer_codec.TransferCodecError(
+                f"exact frame shape {shape} does not match the envelope's 1x{height}x{width} frames"
+            )
+        try:
+            tensors.append(bridge_codec.decode_payload("IMAGE", descriptor, payload))
+        except ValueError as exc:
+            raise transfer_codec.TransferCodecError(f"exact frame: {exc}") from None
+    return tensors
+
+
 def _validate_compressed_envelope(envelope):
     errors = []
     if not isinstance(envelope, dict):
@@ -384,6 +417,7 @@ async def _job_complete_compressed(request):
             "image_index": start_index + pos,
             "is_last": bool(is_last and pos == last_pos),
             "audio": decoded_audio if pos == last_pos else None,
+            "format": envelope["format"],
         }
         for pos, tensor in enumerate(tensors)
     ]
@@ -462,6 +496,7 @@ async def job_complete_endpoint(request):
                     "image_index": int(batch_idx),
                     "is_last": is_last,
                     "audio": decoded_audio,
+                    "format": transfer_codec.LEGACY_FORMAT,
                 }
             ],
         )

api/orchestration/dispatch.py

@@ -189,7 +189,7 @@ async def dispatch_worker_prompt(
 
 
 async def preflight_image_list_workers(workers, require_idle=True, timeout=3.0, *,
-                                       required_transfer_formats=()):
+                                       required_transfer_formats=(), require_value_results=False):
     """Require protocol/node support and report which direct workers are idle.
 
     This performs read-only GETs only; it never submits a prompt or contacts a
@@ -261,6 +261,11 @@ async def preflight_image_list_workers(workers, require_idle=True, timeout=3.0,
                             f"{sorted(required_formats)}; upgrade that worker before compressed IMAGE returns; "
                             "no work was submitted"
                         )
+                if require_value_results and capability.get("value_results") is not True:
+                    raise RuntimeError(
+                        f"Image-list worker {worker_id} lacks explicit value_results support; "
+                        "upgrade that worker before using End overrides; no work was submitted"
+                    )
 
                 check = "queue"
                 queue_url = build_worker_url(worker, "/queue")

api/orchestration/prompt_transform.py

@@ -401,20 +401,6 @@ def prune_prompt_for_worker(prompt_obj, use_worker_sink=False):
 
 def prepare_delegate_master_prompt(prompt_obj, collector_ids):
     """Prune master prompt so it only executes post-collector nodes in delegate mode."""
-    # Typed collectors need no fabricated IMAGE placeholder. Sever all media
-    # links, including STRING scalar branches, before retained-config analysis.
-    # Their expected_media control was frozen by participant overrides.
-    if any((prompt_obj.get(cid) or {}).get("class_type") == "DistributedMediaCollector"
-           for cid in collector_ids):
-        from ...serverless_bridge.contract import SOCKET_KINDS
-        prompt_obj = json.loads(json.dumps(prompt_obj))
-        for cid in collector_ids:
-            node = prompt_obj.get(cid) or {}
-            if node.get("class_type") == "DistributedMediaCollector":
-                inputs = node.setdefault("inputs", {})
-                inputs.setdefault("expected_media", json.dumps([name for name in SOCKET_KINDS if name in inputs]))
-                for name in SOCKET_KINDS:
-                    inputs.pop(name, None)
     downstream = _find_downstream_nodes(prompt_obj, collector_ids)
     nodes_to_keep = set(collector_ids)
     nodes_to_keep.update(downstream)
@@ -519,7 +505,6 @@ def generate_job_id_map(prompt_index, prefix):
     job_map = {}
     distributed_nodes = (
         prompt_index.nodes_for_class("DistributedCollector")
-        + prompt_index.nodes_for_class("DistributedMediaCollector")
         + prompt_index.nodes_for_class("UltimateSDUpscaleDistributed")
         + prompt_index.nodes_for_class("DistributedImageListStart")
     )
@@ -552,10 +537,8 @@ def _override_collector_nodes(
     enabled_json,
     delegate_master,
 ):
-    """Configure legacy and typed collectors without changing their media."""
-    collector_ids = (prompt_index.nodes_for_class("DistributedCollector")
-                     + prompt_index.nodes_for_class("DistributedMediaCollector"))
-    for node_id in collector_ids:
+    """Configure DistributedCollector nodes for master or worker role."""
+    for node_id in prompt_index.nodes_for_class("DistributedCollector"):
         node = prompt_copy.get(node_id)
         if not isinstance(node, dict):
             continue
@@ -568,9 +551,6 @@ def _override_collector_nodes(
         inputs["multi_job_id"] = job_id_map.get(node_id, node_id)
         inputs["is_worker"] = not is_master
         inputs["enabled_worker_ids"] = enabled_json
-        if node.get("class_type") == "DistributedMediaCollector":
-            from ...serverless_bridge.contract import SOCKET_KINDS
-            inputs["expected_media"] = json.dumps([socket for socket in SOCKET_KINDS if socket in inputs])
         if is_master:
             inputs["delegate_only"] = bool(delegate_master)
             inputs.pop("master_url", None)

api/orchestration/serverless_media_prompt.py

@@ -1,8 +1,9 @@
-"""Pure compilation of prepared typed-media collectors for serverless workers.
+"""Pure compilation of prepared Distributed Collectors for comfy-media-v1 pools.
 
 The legacy image compiler remains separate. This compiler emits one MediaExportV1
-execution target per connected socket, retaining only its upstream computation.
-It never imports ComfyUI or inspects/executes runtime node classes.
+execution target per connected Collector socket (``images`` and/or ``audio``),
+retaining only its upstream computation. It never imports ComfyUI or
+inspects/executes runtime node classes.
 """
 
 from __future__ import annotations
@@ -12,8 +13,8 @@ import hashlib
 import json
 from graphlib import CycleError, TopologicalSorter
 
-from ...serverless_bridge.contract import SOCKET_KINDS, identifier, validate_bindings
-from ...serverless_bridge.image_transfer import DEFAULT_FORMAT, DEFAULT_QUALITY, validate_options
+from ...serverless_bridge.contract import SOCKET_KINDS, VALUE_SOCKETS, identifier, validate_bindings
+from ...serverless_bridge.image_transfer import DEFAULT_FORMAT, validate_options
 from .serverless_prompt import (
     _PRIMITIVE_NODES,
     _REMOVED_SCALAR_NODES,
@@ -24,7 +25,24 @@ from .serverless_prompt import (
 )
 
 
-_COLLECTOR = "DistributedMediaCollector"
+_COLLECTOR = "DistributedCollector"
+# Collector inputs, in output-slot order, that a typed pool returns.
+COLLECTOR_SOCKETS = ("images", "audio")
+# Collector widget defaults (the saved-workflow contract is legacy_png / 90).
+COLLECTOR_DEFAULT_FORMAT = "legacy_png"
+COLLECTOR_DEFAULT_QUALITY = 90
+
+
+def collector_image_policy(inputs):
+    """Return the endpoint IMAGE (format, quality) for one Collector.
+
+    legacy_png is the Collector's saved default and names a JSON/base64 framing
+    that only applies to direct workers; serverless returns the same 8-bit
+    pixels as compressed png. exact stays opt-in. No other format is altered.
+    """
+    fmt, quality = validate_options(inputs.get("transfer_format", COLLECTOR_DEFAULT_FORMAT),
+                                    inputs.get("transfer_quality", COLLECTOR_DEFAULT_QUALITY))
+    return ("png" if fmt == "legacy_png" else fmt), quality
 
 
 def _validate_identifier(value, context):
@@ -36,7 +54,7 @@ def _validate_identifier(value, context):
         raise ValueError(f"{context} must be a valid nonempty media identifier") from None
 
 
-def _validate_prompt(prepared_prompt):
+def _validate_prompt(prepared_prompt, *, require_collectors=True):
     """Validate the entire source graph, including branches later discarded."""
     if not isinstance(prepared_prompt, dict) or not prepared_prompt:
         raise ValueError("prepared prompt must be a nonempty API graph mapping")
@@ -51,7 +69,7 @@ def _validate_prompt(prepared_prompt):
             raise ValueError(f"prompt node {node_id!r} must have a nonempty class_type")
         if class_type == _COLLECTOR:
             collectors.append(node_id)
-        elif class_type == "DistributedCollector" or _is_unsupported_distributed(class_type):
+        elif _is_unsupported_distributed(class_type):
             raise ValueError(f"unsupported distributed or batch node: {class_type}")
         inputs = node.get("inputs", {})
         if not isinstance(inputs, dict):
@@ -59,8 +77,8 @@ def _validate_prompt(prepared_prompt):
         if any(not isinstance(name, str) or not name for name in inputs):
             raise ValueError(f"prompt node {node_id!r} input names must be nonempty strings")
 
-    if not collectors:
-        raise ValueError("prepared prompt contains no DistributedMediaCollector nodes")
+    if require_collectors and not collectors:
+        raise ValueError("prepared prompt contains no DistributedCollector nodes")
 
     dependencies = {}
     for node_id, node in prepared_prompt.items():
@@ -81,18 +99,17 @@ def _validate_prompt(prepared_prompt):
             if source_class in _REMOVED_SCALAR_NODES or source_class in _PRIMITIVE_NODES:
                 if slot != 0:
                     raise ValueError(f"invalid output slot {slot} for scalar node {source_id!r}")
-            elif source_class == _COLLECTOR and slot >= len(SOCKET_KINDS):
+            elif source_class == _COLLECTOR and slot >= len(COLLECTOR_SOCKETS):
                 raise ValueError(f"invalid output slot {slot} for collector {source_id!r}")
             parents.add(source_id)
         dependencies[node_id] = parents
 
     for collector_id in collectors:
         inputs = prepared_prompt[collector_id].get("inputs", {})
-        validate_options(inputs.get("transfer_format", DEFAULT_FORMAT),
-                         inputs.get("transfer_quality", DEFAULT_QUALITY))
+        collector_image_policy(inputs)
         for field in ("multi_job_id", "worker_id"):
             _validate_identifier(inputs.get(field), f"collector {collector_id!r} {field}")
-        connected = [socket for socket in SOCKET_KINDS if socket in inputs]
+        connected = [socket for socket in COLLECTOR_SOCKETS if socket in inputs]
         if not connected:
             raise ValueError(f"collector {collector_id!r} requires at least one connected media socket")
         for socket in connected:
@@ -123,61 +140,84 @@ def _validate_prompt(prepared_prompt):
     return sorted(collectors)
 
 
-def _build_exports(source_nodes, collector_ids, request_id):
-    bindings = []
-    exports = {}
+def _build_explicit_exports(source_nodes, specs, request_id):
+    bindings, exports = [], {}
     reserved = set(source_nodes)
-    for collector_id in collector_ids:
-        inputs = source_nodes[collector_id].get("inputs", {})
-        for socket, kind in SOCKET_KINDS.items():
-            if socket not in inputs:
-                continue
-            # Hashing avoids separator ambiguity and keeps long valid source
-            # identities within the bridge's identifier length bound. Identity
-            # is stable across request IDs and dictionary insertion orders.
-            identity = json.dumps([collector_id, socket], separators=(",", ":"))
-            digest = hashlib.sha256(identity.encode("utf-8")).hexdigest()
-            base_id = f"serverless:media:{digest}"
-            node_id = base_id
-            suffix = 0
-            while node_id in reserved:
-                suffix += 1
-                node_id = f"{base_id}:{suffix}"
-            reserved.add(node_id)
-            binding = {
-                "binding_id": f"b:{digest}",
-                "collector_id": collector_id,
-                "job_id": inputs["multi_job_id"],
-                "participant_id": inputs["worker_id"],
-                "node_id": node_id,
-                "socket": socket,
-                "kind": kind,
-            }
-            bindings.append(binding)
-            exports[node_id] = {
-                "class_type": "MediaExportV1",
-                "inputs": {
-                    "request_id": request_id,
-                    "binding_json": json.dumps(binding),
-                    socket: copy.deepcopy(inputs[socket]),
-                },
-            }
-            if kind == "IMAGE":
-                fmt, quality = validate_options(inputs.get("transfer_format", DEFAULT_FORMAT),
-                                                inputs.get("transfer_quality", DEFAULT_QUALITY))
-                if fmt != DEFAULT_FORMAT:
-                    exports[node_id]["inputs"].update(transfer_format=fmt, transfer_quality=quality)
+    for spec in specs:
+        if not isinstance(spec, dict) or set(spec) != {
+                "collector_id", "job_id", "participant_id", "socket", "source",
+                "transfer_format", "transfer_quality"}:
+            raise ValueError("Invalid explicit media export specification")
+        collector_id, socket = spec["collector_id"], spec["socket"]
+        kind = "VALUE" if socket in VALUE_SOCKETS else SOCKET_KINDS.get(socket)
+        if kind is None:
+            raise ValueError("Unsupported explicit media export socket")
+        link = spec["source"]
+        if (not _is_link(link) or link[0] not in source_nodes
+                or type(link[1]) is not int or link[1] < 0):
+            raise ValueError("Explicit media export requires a valid source link")
+        fmt, quality = validate_options(spec["transfer_format"], spec["transfer_quality"])
+        if kind != "IMAGE" and fmt != DEFAULT_FORMAT:
+            raise ValueError("Image transfer controls apply only to IMAGE exports")
+        # Same stable identities as Collector compilation, including collisions.
+        identity = json.dumps([collector_id, socket], separators=(",", ":"))
+        digest = hashlib.sha256(identity.encode("utf-8")).hexdigest()
+        base_id = f"serverless:media:{digest}"
+        node_id, suffix = base_id, 0
+        while node_id in reserved:
+            suffix += 1
+            node_id = f"{base_id}:{suffix}"
+        reserved.add(node_id)
+        binding = {"binding_id": f"b:{digest}", "collector_id": collector_id,
+                   "job_id": spec["job_id"], "participant_id": spec["participant_id"],
+                   "node_id": node_id, "socket": socket, "kind": kind}
+        bindings.append(binding)
+        exports[node_id] = {"class_type": "MediaExportV1", "inputs": {
+            "request_id": request_id, "binding_json": json.dumps(binding),
+            socket: copy.deepcopy(link)}}
+        if fmt != DEFAULT_FORMAT:
+            exports[node_id]["inputs"].update(transfer_format=fmt, transfer_quality=quality)
     validate_bindings(bindings)
     return bindings, exports
 
 
+def _build_exports(source_nodes, collector_ids, request_id):
+    specs = []
+    for collector_id in collector_ids:
+        inputs = source_nodes[collector_id].get("inputs", {})
+        for socket in COLLECTOR_SOCKETS:
+            if socket not in inputs:
+                continue
+            fmt, quality = collector_image_policy(inputs) if socket == "images" else (DEFAULT_FORMAT, 90)
+            specs.append({"collector_id": collector_id, "job_id": inputs["multi_job_id"],
+                          "participant_id": inputs["worker_id"], "socket": socket,
+                          "source": inputs[socket], "transfer_format": fmt, "transfer_quality": quality})
+    return _build_explicit_exports(source_nodes, specs, request_id)
+
+
+def compile_export_prompt(prepared_prompt, request_id, specs):
+    """Compile explicit internal exports, without fabricating a Collector node.
+
+    Callers supply linked source sockets and literal binding identities; only
+    exporters and required ancestors survive. This does not add Collector
+    sockets: its public compiler remains images/audio only.
+    """
+    _validate_identifier(request_id, "request_id")
+    if _validate_prompt(prepared_prompt, require_collectors=False):
+        raise ValueError("Explicit exports cannot contain Collector nodes")
+    bindings, exports = _build_explicit_exports(prepared_prompt, specs, request_id)
+    compiled = _prune_graph(prepared_prompt, bindings, exports, _make_scalar_resolver(prepared_prompt))
+    return {node_id: compiled[node_id] for node_id in sorted(compiled)}, bindings
+
+
 def compile_media_prompt(prepared_prompt: dict, request_id: str) -> tuple[dict, list[dict]]:
     """Return a worker graph and canonical typed-media export bindings.
 
-    Each DistributedMediaCollector must have prepared literal ``multi_job_id``
-    and ``worker_id`` identities and at least one linked media input. Present
-    media inputs (including texts) must be API links; literal text belongs in a
-    PrimitiveString node. Nonmedia collector controls are not exported.
+    Each DistributedCollector must have prepared literal ``multi_job_id`` and
+    ``worker_id`` identities and at least one linked ``images``/``audio`` input.
+    Present media inputs must be API links. Collector transport controls become
+    the IMAGE export policy (``legacy_png`` is sent as ``png``); other controls
+    are not exported.
 
     Only MediaExportV1 targets and their ancestors survive. Required ordinary
     output/utility nodes are preserved regardless of their runtime OUTPUT_NODE

api/queue_orchestration.py

@@ -153,44 +153,36 @@ def _is_load_balance_enabled(value):
 
 
 def _prompt_requests_load_balance(prompt_index):
-    collector_ids = (prompt_index.nodes_for_class("DistributedCollector")
-                     + prompt_index.nodes_for_class("DistributedMediaCollector"))
-    for node_id in collector_ids:
+    for node_id in prompt_index.nodes_for_class("DistributedCollector"):
         inputs = prompt_index.inputs_by_node.get(node_id, {})
         if _is_load_balance_enabled(inputs.get("load_balance", False)):
             return True
     return False
 
 
-def _validate_media_workers(prompt_index, workers):
-    """Reject incompatible selections before probing, pruning or dispatching.
+def _validate_media_workers(workers, prompt_index=None):
+    """Reject unknown serverless media profiles before probing or dispatching.
 
-    Typed collectors currently have only a master/serverless transport. Check
-    all selected workers, not just reachable or load-balanced participants, so
-    a bad selection cannot silently turn into a different execution mode.
+    A Distributed Collector accepts direct workers, legacy-image-v1 pools and
+    comfy-media-v1 pools in the same run; each pool's profile selects its
+    endpoint contract. Unknown profiles fail rather than guessing a contract.
+    legacy-image-v1 endpoints return 8-bit files, so a Collector set to exact
+    cannot use them.
     """
-    typed = bool(prompt_index.nodes_for_class("DistributedMediaCollector"))
-    if typed and (prompt_index.nodes_for_class("DistributedCollector")
-                  or prompt_index.nodes_for_class("UltimateSDUpscaleDistributed")):
-        raise ValueError(
-            "DistributedMediaCollector cannot be mixed with DistributedCollector or "
-            "UltimateSDUpscaleDistributed in one prompt"
-        )
+    exact = prompt_index is not None and any(
+        str(prompt_index.inputs_by_node.get(node_id, {}).get("transfer_format", "")).strip().lower() == "exact"
+        for node_id in prompt_index.nodes_for_class("DistributedCollector")
+    )
     for worker in workers:
         if worker.get("type") != "serverless":
-            if typed:
-                raise ValueError(
-                    "DistributedMediaCollector does not support selected direct workers; "
-                    "use master execution or comfy-media-v1 serverless pools"
-                )
             continue
         profile = worker.get("media_profile", "legacy-image-v1")
         if profile not in ("legacy-image-v1", "comfy-media-v1"):
             raise ValueError("Unsupported serverless media_profile")
-        if typed != (profile == "comfy-media-v1"):
+        if exact and profile == "legacy-image-v1":
             raise ValueError(
-                "DistributedMediaCollector requires comfy-media-v1 serverless pools; "
-                "legacy collectors require legacy-image-v1"
+                "transfer_format exact requires comfy-media-v1 serverless pools; "
+                "legacy-image-v1 endpoints return 8-bit images"
             )
 
 
@@ -209,7 +201,7 @@ async def _prepare_worker_payload(
 ):
     """Prepare one worker prompt payload with bounded concurrency and media-sync timeout."""
     async with worker_prep_semaphore:
-        _validate_media_workers(prompt_index, [worker])
+        _validate_media_workers([worker], prompt_index)
         worker_prompt = prompt_index.copy_prompt()
         if worker.get("type") == "serverless":
             from .orchestration.runpod_client import serialize_request
@@ -217,9 +209,8 @@ async def _prepare_worker_payload(
                 from ..serverless_bridge.contract import PROFILE, VERSION
                 from .orchestration.serverless_media_prompt import compile_media_prompt
                 from .orchestration.serverless_typed_inputs import prepare_media_inputs
-                # The legacy pruner only recognizes Collector/Upscale roots.
                 # The typed compiler validates the complete private graph and
-                # prunes it itself, retaining every connected typed branch.
+                # prunes it itself, retaining every connected Collector branch.
                 worker_prompt = apply_participant_overrides(
                     worker_prompt, worker["id"], enabled_ids, job_id_map,
                     "", delegate_master, prompt_index,
@@ -332,34 +323,20 @@ async def orchestrate_distributed_execution(
                           for node in prompt_obj.values())
     target_kwargs = {}
     if image_list_mode:
-        if any(isinstance(node, dict) and node.get("class_type") in {
-                "DistributedCollector", "DistributedMediaCollector", "UltimateSDUpscaleDistributed"
-        } for node in prompt_obj.values()):
-            raise ValueError("Distributed Image List cannot be mixed with other distributed collectors")
+        # Check the source graph before execution-root pruning can remove an
+        # incompatible mode (or mask it with an unrelated boundary error).
+        if any(isinstance(node, dict) and node.get("class_type") in (
+                "DistributedCollector", "UltimateSDUpscaleDistributed") for node in prompt_obj.values()):
+            raise ValueError(
+                "Distributed Image List v1 cannot be mixed with Distributed Collector or "
+                "Ultimate SD Upscale nodes in one prompt"
+            )
         prompt_obj, outer_targets = prepare_image_list_execution(prompt_obj)
         target_kwargs["partial_execution_targets"] = outer_targets
     prompt_index = PromptIndex(prompt_obj)
-    typed_media_mode = bool(prompt_index.nodes_for_class("DistributedMediaCollector"))
-    if typed_media_mode:
-        from ..serverless_bridge.contract import identifier
-        # PromptIndex's JSON copy would otherwise turn integer keys into strings.
-        for node_id in prompt_obj:
-            identifier(node_id)
-    image_list_start_ids = prompt_index.nodes_for_class("DistributedImageListStart")
-    image_list_mode = bool(image_list_start_ids)
     if image_list_mode:
         # Validate before probing or queueing anything so malformed boundaries and
         # old worker releases fail before any upstream image is materialized.
-        other_distributed = (
-            prompt_index.nodes_for_class("DistributedCollector")
-            + prompt_index.nodes_for_class("DistributedMediaCollector")
-            + prompt_index.nodes_for_class("UltimateSDUpscaleDistributed")
-        )
-        if other_distributed:
-            raise ValueError(
-                "Distributed Image List v1 cannot be mixed with Distributed Collector, "
-                "DistributedMediaCollector or Ultimate SD Upscale nodes in one prompt"
-            )
         validate_image_list_regions(prompt_obj, explicit_targets=True)
         from ..serverless_bridge.image_transfer import DEFAULT_FORMAT, DEFAULT_QUALITY, validate_options
         required_transfer_formats = set()
@@ -381,14 +358,14 @@ async def orchestrate_distributed_execution(
                 "or comfy-media-v1 serverless pool"
             )
     else:
-        if typed_media_mode and requested_ids is not None:
-            configured_ids = {worker.get("id") for worker in config.get("workers", [])}
-            if any(worker_id not in configured_ids for worker_id in requested_ids):
-                raise ValueError("Unknown explicitly selected media worker ID")
         workers = _resolve_enabled_workers(config, requested_ids)
-        _validate_media_workers(prompt_index, workers)
-        if typed_media_mode and delegate_master and not workers:
-            raise RuntimeError("Delegate-only DistributedMediaCollector requires a selected comfy-media-v1 pool")
+        _validate_media_workers(workers, prompt_index)
+        if any(worker.get("media_profile") == "comfy-media-v1" for worker in workers):
+            from ..serverless_bridge.contract import identifier
+            # Validate the source, not PromptIndex's JSON copy (which coerces
+            # integer keys to strings), before probing or dispatching workers.
+            for node_id in prompt_obj:
+                identifier(node_id)
     load_balance_requested = _prompt_requests_load_balance(prompt_index)
     trace_debug(
         execution_trace_id,
@@ -404,8 +381,8 @@ async def orchestrate_distributed_execution(
         ),
     )
 
-    # Keep the legacy Collector fallback. Image-list and typed-media delegate
-    # modes must not silently enable master participation.
+    # Keep the legacy Collector fallback, but never override image-list
+    # orchestrator-only mode merely because its direct workers are unavailable.
     if not image_list_mode and not workers and delegate_master:
         trace_debug(
             execution_trace_id,
@@ -420,23 +397,24 @@ async def orchestrate_distributed_execution(
         trace_execution_id=execution_trace_id,
         probe_concurrency=worker_probe_concurrency,
     )
-    if typed_media_mode and delegate_master and not active_workers:
-        raise RuntimeError("No selected comfy-media-v1 worker is reachable for delegate-only DistributedMediaCollector")
     if image_list_mode:
         from .orchestration.dispatch import preflight_image_list_workers
         if not active_workers and delegate_master:
             raise RuntimeError(
                 "No selected worker is reachable for Distributed Image List in orchestrator-only mode"
             )
+        from ..image_list.runtime import END_CLASS, end_override_names
+        requires_values = any(end_override_names(node.get("inputs", {}))
+                              for node in prompt_obj.values() if node.get("class_type") == END_CLASS)
         direct_workers = [worker for worker in active_workers if worker.get("type") != "serverless"]
-        transfer_kwargs = ({"required_transfer_formats": sorted(required_transfer_formats)}
-                           if required_transfer_formats else {})
-        idle_ids = (await preflight_image_list_workers(direct_workers, require_idle=delegate_master, **transfer_kwargs)
-                    if direct_workers else {})
-        active_workers = [worker for worker in active_workers
-                          if worker.get("type") == "serverless" or worker["id"] in idle_ids]
-        if not active_workers and delegate_master:
-            raise RuntimeError("No feature-compatible worker is available in orchestrator-only mode")
+        preflight_kwargs = ({"required_transfer_formats": sorted(required_transfer_formats)}
+                            if required_transfer_formats else {})
+        if requires_values:
+            preflight_kwargs["require_value_results"] = True
+        # Preserve busy direct candidates for queued runs; runtime rechecks idle
+        # availability before assignment. Never direct-probe serverless pools.
+        if direct_workers:
+            await preflight_image_list_workers(direct_workers, require_idle=False, **preflight_kwargs)
     else:
         delegate_master = selected_delegate_master
 
@@ -525,8 +503,7 @@ async def orchestrate_distributed_execution(
     )
 
     if delegate_master and not image_list_mode:
-        collector_ids = (find_nodes_by_class(master_prompt, "DistributedCollector")
-                         + find_nodes_by_class(master_prompt, "DistributedMediaCollector"))
+        collector_ids = find_nodes_by_class(master_prompt, "DistributedCollector")
         upscale_nodes = find_nodes_by_class(master_prompt, "UltimateSDUpscaleDistributed")
         if upscale_nodes:
             debug_log(