fix: carry image-list values through End context

Preserve source values without extra wires, collect explicit per-item overrides, and expose ordered bundles through Get Value. Keep legacy sockets and no-override transport compatible.

api/image_list_routes.py

@@ -36,8 +36,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",
     })
 

api/orchestration/dispatch.py

@@ -188,7 +188,7 @@ async def dispatch_worker_prompt(
         return prompt_id if return_prompt_id else None
 
 
-async def preflight_image_list_workers(workers, require_idle=True, timeout=3.0):
+async def preflight_image_list_workers(workers, require_idle=True, timeout=3.0, *, 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
@@ -242,6 +242,12 @@ async def preflight_image_list_workers(workers, require_idle=True, timeout=3.0):
                         "upgrade that worker before running this feature; 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")
                 async with session.get(queue_url, timeout=aiohttp.ClientTimeout(total=timeout)) as response:

api/queue_orchestration.py

@@ -343,7 +343,12 @@ async def orchestrate_distributed_execution(
             raise RuntimeError(
                 "No selected direct worker is reachable for Distributed Image List in orchestrator-only mode"
             )
-        idle_ids = (await preflight_image_list_workers(active_workers, require_idle=delegate_master)
+        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)
+        value_preflight = {"require_value_results": True} if requires_values else {}
+        idle_ids = (await preflight_image_list_workers(active_workers, require_idle=delegate_master,
+                                                       **value_preflight)
                     if active_workers else {})
         active_workers = [worker for worker in active_workers if worker["id"] in idle_ids]
         if not active_workers and delegate_master:

docs/image-list-queue.md

@@ -52,7 +52,31 @@ Start outputs, in order: `image`, `mask`, `item_index` (zero-based), `context`,
 
 Values that cannot be transported exactly (models, latents, conditioning and other objects) are rejected. Load models inside the region from a loader connected directly to the processing nodes, as described below.
 
-End accepts `images`, optional `masks` / `texts`, and `context`. It returns `image_list`, `masks`, and `texts` in original item order. Each job must produce exactly one image, at most one mask, and at most one text value. Unconnected optional outputs contain `None` entries.
+End accepts `images`, optional `masks` / `texts`, and `context`. Its first three outputs remain `image_list`, `masks`, and `texts` in original item order. Each job must produce exactly one image, at most one mask, and at most one text value. Unconnected optional mask/text outputs contain `None` entries.
+
+### Keep values through End without extra wires
+
+Start's existing **context** connection now also lets End recover each item's original `value_1` … `value_10`. You do not need to wire those values individually into End. Context remains a lightweight execution reference, not a general-purpose data pipe.
+
+End appends a **values** output (`DISTRIBUTED_IMAGE_LIST_VALUES`): one value bundle per image, in the same source order. Connect it to **Distributed Image List Get Value**, then choose `value_name` to extract the field you need. The getter preserves its type and ComfyUI maps it across the bundles, producing a list aligned with End's images. Missing fields fail clearly instead of producing empty placeholders.
+
+For example, to preserve filenames:
+
+```text
+filenames → Start.value_1
+                 Start.context ─────────────→ End.context
+                 Start.image → processing → End.images
+                                               ├─ image_list → Save.images
+                                               └─ values → Get Value [value_1] → Save.filename_prefix
+```
+
+**New or changed per-item values:** connect the calculated result to End's optional `value_N` input. It replaces the original field with that number, or adds it if Start did not supply it. Every connected override must produce exactly one supported value for each item. A tensor batch remains one value, just as it does at Start. Image, mask, string, integer, float and boolean values are supported; arbitrary objects/models/latents and non-finite numbers are not.
+
+End shows override inputs progressively. Use **Add value override** to expose a later slot without connecting the earlier ones. Disconnecting an override never renumbers the other End fields: `value_3` always means `value_3`. Unlike Start's native Autogrow inputs, End overrides must keep these identities stable. Existing image/mask/text/context sockets and output indices are unchanged. If the frontend extension is unavailable, Python still declares all optional override inputs and the values output.
+
+Original values are retained on the master; workers only return explicitly connected overrides, avoiding a redundant return trip for unchanged reference images or other values. Overrides use the existing exact value transport and size limits. Selected workers must advertise `value_results: true` when overrides are requested, checked before execution and again at dispatch. Original-value passthrough alone does not require that additional capability.
+
+Values leaving the region must come from **End**, not directly from Start. To chain regions, extract an End value and connect it to the next Start's corresponding input. Values remain available downstream after End cleans up the execution context; they do not depend on a later lookup of expired job state.
 
 Models are loaded and cached on each processing participant, not serialized between machines. Connect loader outputs directly to the processing nodes. The master needs the region's model files when participating, as does each selected worker. The master still validates the prompt: it needs the relevant node definitions and any model-name/file availability required by their validation (including combo lists). This is not a model-free-master guarantee.
 

image_list/execution_plan.py

@@ -6,6 +6,7 @@ import sys
 
 from .runtime import (
     END_CLASS,
+    END_RESULT_INPUTS,
     _ancestors_for_inputs,
     _downstream_until,
     _image_list_pairs,
@@ -49,7 +50,7 @@ def prepare_image_list_execution(prompt: dict) -> tuple[dict, list[str]]:
     required = _reachable(boundaries, upstream)
     discarded = set()
     for start_id, end_id in pairs:
-        region_nodes = _ancestors_for_inputs(prepared, end_id, start_id, ("images", "masks", "texts"))
+        region_nodes = _ancestors_for_inputs(prepared, end_id, start_id, END_RESULT_INPUTS)
         outside = _downstream_until(prepared, start_id, end_id) - region_nodes
         # Preserve bypass joins and their whole chain into post-End computation.
         # They must fail boundary validation, not disappear with debug previews.

image_list/runtime.py

@@ -29,6 +29,7 @@ VALUE_OUT_BASE = 4
 MAX_VALUE_SLOTS = 10
 VALUE_NAMES = tuple(f"value_{i}" for i in range(1, MAX_VALUE_SLOTS + 1))
 VALUES_INPUT = "values"
+END_RESULT_INPUTS = ("images", "masks", "texts") + VALUE_NAMES
 MAX_PAYLOAD_BYTES = 50 * 1024 * 1024
 MAX_REQUEST_BYTES = MAX_PAYLOAD_BYTES * 2 + 1024 * 1024
 MAX_ITEMS = 4096
@@ -179,6 +180,50 @@ def _normalize_values(values: Any, count: int) -> dict[str, list[Any]]:
     return aligned
 
 
+def end_override_names(inputs: dict[str, Any]) -> tuple[str, ...]:
+    """Fixed keys stay stable even when an intermediate socket disconnects."""
+    for name in inputs:
+        if (name.startswith("value_") or name.startswith("values.")) and name not in VALUE_NAMES:
+            raise ValueError(f"Unknown End value slot {name!r}")
+    return tuple(name for name in VALUE_NAMES if name in inputs)
+
+
+def _validate_result_values(values: Any) -> dict[str, Any]:
+    """Validate flat override values; tensors are values, never frame lists."""
+    if not isinstance(values, dict):
+        raise ValueError("End result values must be a mapping of value_N slots")
+    for name, value in values.items():
+        if name not in VALUE_NAMES:
+            raise ValueError(f"Unknown End value slot {name!r}")
+        kind = value_kind(value, name)
+        if kind == "TENSOR":
+            _tensor_dtype_name(value)
+            if not _torch().isfinite(value).all().item():
+                raise ValueError(f"End {name} tensor values must be finite")
+    return values
+
+
+def normalize_value_overrides(values: Any) -> dict[str, Any]:
+    if not isinstance(values, dict):
+        raise ValueError("End overrides must be a mapping of value_N slots")
+    normalized = {}
+    for name, raw in values.items():
+        items = list(raw) if isinstance(raw, (list, tuple)) else [raw]
+        if len(items) != 1:
+            raise ValueError(f"Worker End override {name} expects exactly one value, received {len(items)}")
+        normalized[name] = items[0]
+    return _validate_result_values(normalized)
+
+
+def _checked_overrides(result: dict[str, Any], expected: tuple[str, ...]) -> dict[str, Any]:
+    values = _validate_result_values(result.get("values", {}))
+    missing, extra = set(expected) - values.keys(), values.keys() - set(expected)
+    if missing or extra:
+        raise ValueError(f"End result overrides do not match connected slots: "
+                         f"missing {sorted(missing)}, unexpected {sorted(extra)}")
+    return values
+
+
 def normalize_image_list_inputs(images: Any, mask: Any = None, values: Any = None) -> list[dict[str, Any]]:
     """Split images into items and pair masks/generic values by the 1-or-N rule."""
     frames = _image_frames(images, "images")
@@ -544,10 +589,11 @@ def validate_image_list_regions(prompt: dict[str, Any], *, explicit_targets: boo
         if "images" not in prompt[end_id].get("inputs", {}):
             raise ValueError(f"End node {end_id}.images must be connected to the per-item result branch")
         end_inputs = prompt[end_id].get("inputs", {})
-        for name in ("images", "masks", "texts"):
+        end_override_names(end_inputs)
+        for name in END_RESULT_INPUTS:
             if name in end_inputs:
                 _direct_link(end_inputs[name], prompt, end_id, name)
-        region_nodes = _ancestors_for_inputs(prompt, end_id, start_id, ("images", "masks", "texts"))
+        region_nodes = _ancestors_for_inputs(prompt, end_id, start_id, END_RESULT_INPUTS)
         if start_id not in region_nodes:
             raise ValueError(f"Start node {start_id} is not an ancestor of End node {end_id}")
         forbidden = {n for n in region_nodes if n != start_id and n != end_id and
@@ -713,6 +759,7 @@ def set_master_result(context_id: str, item_index: int, token: str, result: dict
             raise ValueError("Image-list result token does not match an assigned item")
         if item_index in state.received_indexes:
             raise ValueError(f"Duplicate image-list result for item {item_index}")
+        _checked_overrides(result, end_override_names(state.original_prompt[state.end_id].get("inputs", {})))
         state.received_indexes.add(item_index)
         state.results[item_index] = result
         event = state.event
@@ -744,10 +791,24 @@ def decode_result(result: Any) -> dict[str, Any]:
     if mask is not None and (mask.ndim != 3 or mask.shape[0] != 1):
         raise ValueError("result.mask must contain exactly one MASK frame")
     text = decode_value(result.get("text"), "STRING")
-    return {"image": image, "mask": mask, "text": text}
+    decoded = {"image": image, "mask": mask, "text": text}
+    if "values" in result:
+        raw_values = result["values"]
+        if not isinstance(raw_values, dict):
+            raise ValueError("result.values must be an object")
+        values = {}
+        for name, encoded in raw_values.items():
+            if name not in VALUE_NAMES:
+                raise ValueError(f"result.values has unknown slot {name!r}")
+            if not isinstance(encoded, dict) or encoded.get("type") not in _VALUE_KINDS:
+                raise ValueError(f"result.values.{name} has an unsupported type")
+            values[name] = decode_value(encoded, encoded["type"])
+        decoded["values"] = _validate_result_values(values)
+    return decoded
 
 
-def normalize_worker_result(images: Any, masks: Any = None, texts: Any = None) -> dict[str, Any]:
+def normalize_worker_result(images: Any, masks: Any = None, texts: Any = None,
+                            values: Any = None) -> dict[str, Any]:
     def one(value, name, kind):
         if value is None:
             return None
@@ -770,17 +831,29 @@ def normalize_worker_result(images: Any, masks: Any = None, texts: Any = None) -
     text = one(texts, "texts", "STRING")
     if text is not None and not isinstance(text, str):
         raise TypeError("Worker End text result must be a string")
-    return {"image": image, "mask": mask, "text": text}
+    result = {"image": image, "mask": mask, "text": text}
+    if values is not None:
+        overrides = normalize_value_overrides(values)
+        if overrides:
+            result["values"] = overrides
+    return result
 
 
 def encode_result(result: dict[str, Any]) -> dict[str, Any]:
     normalized = normalize_worker_result(result.get("image"), result.get("mask"), result.get("text"))
-    return {"image": encode_value(normalized["image"], "IMAGE"),
-            "mask": encode_value(normalized["mask"], "MASK"),
-            "text": encode_value(normalized["text"], "STRING")}
+    encoded = {"image": encode_value(normalized["image"], "IMAGE"),
+               "mask": encode_value(normalized["mask"], "MASK"),
+               "text": encode_value(normalized["text"], "STRING")}
+    if "values" in result:
+        values = _validate_result_values(result["values"])
+        if values:
+            encoded["values"] = {name: encode_value(value, value_kind(value, name))
+                                 for name, value in values.items()}
+    return encoded
 
 
-def _collect_output_lists(results: list[dict[str, Any]]) -> tuple[list[Any], list[Any], list[Any]]:
+def _collect_output_lists(results: list[dict[str, Any]], *, state: ImageListJob | None = None,
+                          include_values: bool = False) -> tuple[list[Any], ...]:
     if not results:
         raise RuntimeError("Image-list execution returned no results")
     images, masks, texts = [], [], []
@@ -790,10 +863,20 @@ def _collect_output_lists(results: list[dict[str, Any]]) -> tuple[list[Any], lis
         images.append(result["image"])
         masks.append(result.get("mask"))
         texts.append(result.get("text"))
-    return images, masks, texts
+    if not include_values:
+        return images, masks, texts
+    if state is None or len(results) != len(state.items):
+        raise RuntimeError("Image-list values require exactly one result per source item")
+    expected = end_override_names(state.original_prompt[state.end_id].get("inputs", {}))
+    # Reconstruct unchanged values from the master source, not worker transport.
+    # Do this before the context is removed, and only after all items succeeded.
+    values = [{**item.get("values", {}), **_checked_overrides(result, expected)}
+              for item, result in zip(state.items, results)]
+    return images, masks, texts, values
 
 
-async def run_image_list_job(context_id: str, *, model_callback=None) -> tuple[list[Any], list[Any], list[Any]]:
+async def run_image_list_job(context_id: str, *, model_callback=None,
+                             include_values: bool = False) -> tuple[list[Any], ...]:
     """Schedule a shared queue across the optional master and direct workers."""
     state = get_master_job(context_id)
     if state is None:
@@ -807,7 +890,7 @@ async def run_image_list_job(context_id: str, *, model_callback=None) -> tuple[l
         timer = asyncio.timeout(deadline if deadline > 0 else None)
         try:
             async with timer:
-                return await _run_image_list_job(state)
+                return await _run_image_list_job(state, include_values=include_values)
         except TimeoutError:
             if timer.expired():
                 raise TimeoutError(f"Image-list total deadline exceeded ({deadline:g}s); cleanup attempted") from None
@@ -820,7 +903,7 @@ async def run_image_list_job(context_id: str, *, model_callback=None) -> tuple[l
         remove_master_job(context_id)
 
 
-async def _run_image_list_job(state: ImageListJob):
+async def _run_image_list_job(state: ImageListJob, *, include_values: bool = False):
     context_id = state.context_id
     state.event = asyncio.Event()
     state.event_loop = asyncio.get_running_loop()
@@ -838,7 +921,10 @@ async def _run_image_list_job(state: ImageListJob):
     if not workers and not state.master_participates:
         raise RuntimeError("Distributed Image List requires enabled direct workers; serverless pools are unsupported")
     # Recheck worker capability and idle state immediately before the first item.
-    worker_statuses = await preflight_image_list_workers(workers, require_idle=not state.master_participates) if workers else {}
+    override_names = end_override_names(state.original_prompt[state.end_id].get("inputs", {}))
+    value_preflight = {"require_value_results": True} if override_names else {}
+    worker_statuses = (await preflight_image_list_workers(
+        workers, require_idle=not state.master_participates, **value_preflight) if workers else {})
     workers = [worker for worker in workers if worker["id"] in worker_statuses]
     if not workers and not state.master_participates:
         raise RuntimeError("No free, reachable, feature-compatible direct workers are available; master execution is disabled")
@@ -881,6 +967,13 @@ async def _run_image_list_job(state: ImageListJob):
         if worker is None:
             raise RuntimeError(f"Worker {worker_id} was removed before dispatch")
         assert session is not None
+        if override_names:
+            # A worker can restart/downgrade between items. Check before upload,
+            # not merely once at queue preparation or at the start of the run.
+            ready = await preflight_image_list_workers([worker], require_idle=False,
+                                                       require_value_results=True)
+            if worker_id not in ready:
+                raise RuntimeError(f"Image-list worker {worker_id} is unavailable for value overrides")
         token = create_worker_task_token()
         register_assignment(context_id, index, worker_id, token)
         item_payload = {"context_id": context_id, "item": encode_item(state.items[index])}
@@ -1041,7 +1134,7 @@ async def _run_image_list_job(state: ImageListJob):
         slots = ([master_slot] if local_executor is not None else []) + [worker["id"] for worker in workers]
         results = await dynamic_schedule(len(state.items), slots,
                                          dispatch, wait_next, cleanup, interrupted)
-        return _collect_output_lists(results)
+        return _collect_output_lists(results, state=state, include_values=include_values)
     finally:
         if local_executor is not None:
             # No timeout here: Python cannot safely kill a synchronous node.

nodes/image_list.py

@@ -10,6 +10,7 @@ from ..image_list.runtime import (
     MAX_VALUE_SLOTS,
     VALUE_NAMES,
     VALUES_INPUT,
+    end_override_names,
     create_master_job,
     normalize_image_list_inputs,
     normalize_worker_result,
@@ -149,17 +150,23 @@ class DistributedImageListEnd(io.ComfyNode):
             node_id="DistributedImageListEnd",
             display_name="Distributed Image List End",
             category="image/distributed",
-            description="Closes a paired image-list region and returns ordered image, mask, and text lists.",
+            description=("Closes a paired image-list region and returns ordered image, mask, text and values lists. "
+                         "Start values are retained; connected value_N inputs replace or add that key per item."),
             inputs=[
                 io.Image.Input("images", lazy=True),
                 io.Mask.Input("masks", optional=True, lazy=True),
                 io.String.Input("texts", optional=True, lazy=True, force_input=True),
                 io.Custom("DISTRIBUTED_IMAGE_LIST_CONTEXT").Input("context"),
+                *[io.MultiType.Input(name, types=_VALUE_TYPES, optional=True, lazy=True,
+                                     tooltip="Replace or add this value for each item; exactly one value per item.")
+                  for name in VALUE_NAMES],
             ],
             outputs=[
                 io.Image.Output(display_name="image_list", is_output_list=True),
                 io.Mask.Output(display_name="masks", is_output_list=True),
                 io.String.Output(display_name="texts", is_output_list=True),
+                io.Custom("DISTRIBUTED_IMAGE_LIST_VALUES").Output(
+                    id="values", display_name="values", is_output_list=True),
             ],
             is_input_list=True,
             is_output_node=True,
@@ -181,7 +188,8 @@ class DistributedImageListEnd(io.ComfyNode):
                                and node.get("class_type") == "DistributedImageListEnd"
                                and node.get("inputs", {}).get("worker_result_token") == worker_result_token), {})
             required = []
-            for name, value in (("images", images), ("masks", masks), ("texts", texts)):
+            for name, value in (("images", images), ("masks", masks), ("texts", texts),
+                                *((name, kwargs.get(name)) for name in end_override_names(end_inputs))):
                 connected = name == "images" or name in end_inputs
                 missing = value is None or (isinstance(value, (list, tuple)) and all(v is None for v in value))
                 if connected and missing:
@@ -195,7 +203,7 @@ class DistributedImageListEnd(io.ComfyNode):
     @classmethod
     def execute(cls, images=None, masks=None, texts=None, context=None,
                 worker_result_token="", worker_item_index=-1, master_callback_url="",
-                image_list_local_execution=False):
+                image_list_local_execution=False, **values):
         worker_result_token = unwrap_single(worker_result_token)
         master_callback_url = unwrap_single(master_callback_url)
         if worker_result_token:
@@ -203,7 +211,16 @@ class DistributedImageListEnd(io.ComfyNode):
                               if isinstance(item, str) and item]
             if not context_values or len(set(context_values)) != 1:
                 raise RuntimeError("Worker Image List End is missing its paired context")
-            result = normalize_worker_result(images, masks, texts)
+            prompt = getattr(cls.hidden, "prompt", {}) or {}
+            end_inputs = next((node.get("inputs", {}) for node in prompt.values()
+                               if isinstance(node, dict) and node.get("class_type") == "DistributedImageListEnd"
+                               and node.get("inputs", {}).get("worker_result_token") == worker_result_token), {})
+            expected = end_override_names(end_inputs)
+            # Missing connected lazy values must fail instead of becoming an
+            # implicit fallback to the original Start value.
+            supplied = {name: values.get(name) for name in expected}
+            supplied.update(values)
+            result = normalize_worker_result(images, masks, texts, supplied)
             if unwrap_single(image_list_local_execution):
                 from ..image_list.runtime import set_master_result
                 set_master_result(context_values[0], int(unwrap_single(worker_item_index)),
@@ -214,7 +231,8 @@ class DistributedImageListEnd(io.ComfyNode):
                     submit_worker_result(str(master_callback_url), str(worker_result_token),
                                          context_values[0], int(unwrap_single(worker_item_index)), payload), timeout=60,
                 )
-            return io.NodeOutput([result["image"]], [result.get("mask")], [result.get("text")])
+            return io.NodeOutput([result["image"]], [result.get("mask")], [result.get("text")],
+                                 [result.get("values", {})])
 
         context_values = [item for item in (context if isinstance(context, (list, tuple)) else [context])
                           if isinstance(item, str) and item]
@@ -229,14 +247,40 @@ class DistributedImageListEnd(io.ComfyNode):
             # Add them to its existing tracker; never start/end another tracker
             # that could unpin a model still needed by the parent.
             model_callback = getattr(execution_list, "output_link_callback", None)
-            result_lists = run_async_in_server_loop(run_image_list_job(context_id, model_callback=model_callback))
+            result_lists = run_async_in_server_loop(run_image_list_job(
+                context_id, model_callback=model_callback, include_values=True))
         except ImageListCancelledError:
             import comfy.model_management
             raise comfy.model_management.InterruptProcessingException()
-        images_out, masks_out, texts_out = result_lists
-        if len(images_out) != len(masks_out) or len(images_out) != len(texts_out):
-            raise RuntimeError("Image-list collector received mismatched image, mask, or text result counts")
-        return io.NodeOutput(images_out, masks_out, texts_out)
+        images_out, masks_out, texts_out, values_out = result_lists
+        if len({len(output) for output in result_lists}) != 1:
+            raise RuntimeError("Image-list collector received mismatched image, mask, text or values result counts")
+        return io.NodeOutput(images_out, masks_out, texts_out, values_out)
+
+
+class DistributedImageListGetValue(io.ComfyNode):
+    """Ordinary scalar execution lets core map over End's ordered values list."""
+
+    @classmethod
+    def define_schema(cls):
+        return io.Schema(
+            node_id="DistributedImageListGetValue",
+            display_name="Distributed Image List Get Value",
+            category="image/distributed",
+            inputs=[io.Custom("DISTRIBUTED_IMAGE_LIST_VALUES").Input("values"),
+                    io.Combo.Input("value_name", options=list(VALUE_NAMES), default="value_1")],
+            outputs=[io.AnyType.Output(id="value", display_name="value")],
+        )
+
+    @classmethod
+    def execute(cls, values, value_name="value_1"):
+        if value_name not in VALUE_NAMES:
+            raise ValueError(f"Unknown image-list value slot {value_name!r}")
+        if not isinstance(values, dict):
+            raise TypeError("Image List Get Value expects one values mapping from Image List End")
+        if value_name not in values:
+            raise ValueError(f"Image List Get Value: {value_name} is absent from this item's values")
+        return io.NodeOutput(values[value_name])
 
 
 async def submit_worker_result(callback_url, item_token, context_id, item_index, result):

nodes/v3.py

@@ -10,7 +10,7 @@ from . import utilities as _utilities
 from .collector import DistributedCollectorNode as _CollectorRuntime
 from ..utils import transfer_codec as _transfer_codec
 from .distributed_upscale import UltimateSDUpscaleDistributed as _UpscaleRuntime
-from .image_list import DistributedImageListEnd, DistributedImageListStart
+from .image_list import DistributedImageListEnd, DistributedImageListStart, DistributedImageListGetValue
 
 
 class DistributedSeed(io.ComfyNode):
@@ -212,4 +212,4 @@ class DistributedWorkerSink(io.ComfyNode):
 
 NODES = [DistributedCollector, DistributedSeed, DistributedModelName, DistributedValue,
          ImageBatchDivider, AudioBatchDivider, DistributedEmptyImage, UltimateSDUpscaleDistributed,
-         DistributedWorkerSink, DistributedImageListStart, DistributedImageListEnd]
+         DistributedWorkerSink, DistributedImageListStart, DistributedImageListEnd, DistributedImageListGetValue]

tests/helpers/image_list_network_check.py

@@ -37,14 +37,14 @@ import aiohttp
 
 torch.set_num_threads(1)
 SOURCE_CALLS, BRANCH_CALLS, CAPTURED, MODEL_CALLS = [], [], [], []
-UTILITY_CALLS, ITEM_EXECUTIONS = [], []
+UTILITY_CALLS, ITEM_EXECUTIONS, VALUE_CAPTURES = [], [], []
 ALLOW_MASTER = False
 COUNT = 7
 
 
 def source_rows():
     return [{'image': torch.full((1, 5 + i, 7, 3 if i % 2 == 0 else 4), i / 16, dtype=torch.float32),
-             'reference_image': torch.full((1, 2, 3, 3), (i + 1) / 16),
+             'reference_image': torch.full((2, 2, 3, 3), (i + 1) / 16),
              'mask': torch.full((1, 5 + i, 7), (i + 1) / 16),
              'text': f'item-{i}', 'strength': (i + 1) / 16, 'seed': 100 + i, 'index': i}
             for i in range(COUNT)]
@@ -160,17 +160,32 @@ class TestImageOperation:
         return image + strength, mask * 0.5, f'{text}:seed-{seed}'
 
 
+class TestValueOverride:
+    @classmethod
+    def INPUT_TYPES(cls):
+        return {'required': {'text': ('STRING', {'forceInput': True}),
+                             'index': ('INT', {'forceInput': True})}}
+    RETURN_TYPES = ('STRING', 'INT')
+    FUNCTION = 'run'
+    def run(self, text, index):
+        assert WORKER or ALLOW_MASTER, 'override branch executed on delegate-only master'
+        return (text + ':override', 200 + index)
+
+
 class TestCapture:
     @classmethod
     def INPUT_TYPES(cls):
-        return {'required': {'images': ('IMAGE',), 'masks': ('MASK',), 'texts': ('STRING', {'forceInput': True})}}
+        return {'required': {'images': ('IMAGE',), 'masks': ('MASK',), 'texts': ('STRING', {'forceInput': True})},
+                'optional': {'values': ('DISTRIBUTED_IMAGE_LIST_VALUES',),
+                             'picked': ('STRING', {'forceInput': True})}}
     INPUT_IS_LIST = True
     OUTPUT_NODE = True
     RETURN_TYPES = ()
     FUNCTION = 'run'
-    def run(self, images, masks, texts):
+    def run(self, images, masks, texts, values=None, picked=None):
         assert not WORKER, 'post-End capture leaked into a worker prompt'
         CAPTURED.append((images, masks, texts))
+        VALUE_CAPTURES.append((values, picked))
         return ()
 
 
@@ -179,11 +194,19 @@ async def setup():
     ps = server.PromptServer(asyncio.get_running_loop(), assets)
     assert await nodes.load_custom_node(str(ROOT))
     module = sys.modules[str(ROOT).replace('.', '_x_')]
+    boundary = importlib.import_module(module.__name__ + '.nodes.image_list')
+    encode_result = boundary.encode_result
+    def checked_encode(result):
+        encoded = encode_result(result)
+        if 'values' in encoded:
+            assert set(encoded['values']) == {f'value_{n}' for n in (2, 3, 4, 5, 7, 8, 9, 10)}
+        return encoded
+    boundary.encode_result = checked_encode
     nodes.NODE_CLASS_MAPPINGS.update(TestImageSource=TestImageSource,
                                    TestImageOperation=TestImageOperation, TestCapture=TestCapture,
                                    TestCachedModel=TestCachedModel, TestSharedFlag=TestSharedFlag,
                                    TestRequiredOutputUtility=TestRequiredOutputUtility,
-                                   TestDiscardedPreview=TestDiscardedPreview)
+                                   TestDiscardedPreview=TestDiscardedPreview, TestValueOverride=TestValueOverride)
     # Match the CORS-enabled direct-worker launch contract for canvas probes.
     app = web.Application(middlewares=[server.create_cors_middleware('*')])
     # Register the actual feature routes, not test replacements.
@@ -271,21 +294,45 @@ def graph(callback, worker_ids):
             'images': ['1', 0], 'mask': ['1', 2],
             'values.value_1': ['1', 1], 'values.value_2': ['1', 3],
             'values.value_3': ['1', 4], 'values.value_4': ['1', 5], 'values.value_5': ['7', 0],
+            'values.value_6': ['1', 2],
             'image_list_enabled_worker_ids': json.dumps(worker_ids), 'image_list_master_callback': callback}},
         '3': {'class_type': 'TestImageOperation', 'inputs': {
             key: ['2', index] for key, index in [('image', 0), ('mask', 1), ('index', 2),
                 ('reference_image', 4), ('text', 5), ('strength', 6), ('seed', 7), ('flag', 8)]}
               | {'image': ['8', 0], 'model': ['6', 0]}},
         '4': {'class_type': 'DistributedImageListEnd', 'inputs': {
-            'images': ['3', 0], 'masks': ['3', 1], 'texts': ['3', 2], 'context': ['2', 3]}},
-        '5': {'class_type': 'TestCapture', 'inputs': {'images': ['4', 0], 'masks': ['4', 1], 'texts': ['4', 2]}},
+            'images': ['3', 0], 'masks': ['3', 1], 'texts': ['3', 2], 'context': ['2', 3],
+            'value_2': ['10', 0], 'value_3': ['2', 6], 'value_4': ['2', 7], 'value_5': ['2', 8],
+            'value_7': ['3', 1], 'value_8': ['2', 4], 'value_9': ['2', 9], 'value_10': ['10', 1]}},
+        '5': {'class_type': 'TestCapture', 'inputs': {'images': ['4', 0], 'masks': ['4', 1], 'texts': ['4', 2],
+                                                     'values': ['4', 3], 'picked': ['get', 0]}},
         '6': {'class_type': 'TestCachedModel', 'inputs': {}},
         '7': {'class_type': 'TestSharedFlag', 'inputs': {}},
         '8': {'class_type': 'TestRequiredOutputUtility', 'inputs': {'image': ['2', 0], 'index': ['2', 2]}},
         '9': {'class_type': 'TestDiscardedPreview', 'inputs': {'image': ['8', 0]}},
+        '10': {'class_type': 'TestValueOverride', 'inputs': {'text': ['2', 5], 'index': ['2', 2]}},
+        'get': {'class_type': 'DistributedImageListGetValue', 'inputs': {'values': ['4', 3], 'value_name': 'value_2'}},
     }
 
 
+def check_carried_values():
+    values, picked = VALUE_CAPTURES[-1]
+    assert len(values) == len(picked) == COUNT
+    for index, (actual, source) in enumerate(zip(values, source_rows())):
+        assert set(actual) == {f'value_{n}' for n in range(1, 11)}, actual.keys()
+        assert torch.equal(actual['value_1'], source['reference_image'])
+        assert actual['value_2'] == picked[index] == f'item-{index}:override'
+        assert actual['value_3'] == source['strength']
+        assert actual['value_4'] == source['seed']
+        assert actual['value_5'] is True
+        assert torch.equal(actual['value_6'], source['mask'])
+        assert torch.equal(actual['value_7'], source['mask'] * 0.5)
+        assert torch.equal(actual['value_8'], source['reference_image'])
+        assert actual['value_8'].shape[0] == 2, 'value tensor batch was split'
+        assert torch.equal(actual['value_9'], source['mask'])
+        assert actual['value_10'] == 200 + index
+
+
 async def coordinator():
     global ALLOW_MASTER
     ps, assets, module, app = await setup()
@@ -403,6 +450,7 @@ async def coordinator():
             assert executor.success, executor.status_messages
             assert len(SOURCE_CALLS) == 1 and not BRANCH_CALLS and not UTILITY_CALLS
             assert len(CAPTURED) == 1
+            check_carried_values()
             images, masks, texts = CAPTURED[0]
             assert len(images) == len(masks) == len(texts) == COUNT
             for i, row in enumerate(source_rows()):
@@ -430,15 +478,20 @@ async def coordinator():
             # left unconnected. Region N+1 must not race the prior worker End.
             chained = {'1': {'class_type': 'EmptyImage', 'inputs': {
                 'height': 8, 'width': 8, 'batch_size': 2, 'color': 0}}}
+            chained['constant'] = {'class_type': 'TestSharedFlag', 'inputs': {}}
             previous = '1'
+            value_source = 'constant'
             for start, end in [('2', '3'), ('4', '5'), ('6', '7')]:
                 chained[start] = {'class_type': 'DistributedImageListStart', 'inputs': {
-                    'images': [previous, 0]}}
+                    'images': [previous, 0], 'values.value_1': [value_source, 0]}}
                 chained[end] = {'class_type': 'DistributedImageListEnd', 'inputs': {
                     'images': [start, 0], 'context': [start, 3]}}
+                value_source = f'get-{end}'
+                chained[value_source] = {'class_type': 'DistributedImageListGetValue', 'inputs': {
+                    'values': [end, 3], 'value_name': 'value_1'}}
                 previous = end
             chained['8'] = {'class_type': 'TestCapture', 'inputs': {
-                'images': ['7', 0], 'masks': ['7', 1], 'texts': ['7', 2]}}
+                'images': ['7', 0], 'masks': ['7', 1], 'texts': ['7', 2], 'values': ['7', 3]}}
             saved_chained = copy.deepcopy(chained)
             with patch.object(config_mod, 'load_config', return_value=config), \
                  patch.object(orchestration, 'load_config', return_value=config), \
@@ -451,6 +504,7 @@ async def coordinator():
             assert len(CAPTURED) == 2 and len(CAPTURED[1][0]) == 2
             assert all(torch.equal(image, torch.zeros((1, 8, 8, 3))) for image in CAPTURED[1][0])
             assert CAPTURED[1][1:] == ([None, None], [None, None])
+            assert VALUE_CAPTURES[1][0] == [{'value_1': True}, {'value_1': True}]
             assert not runtime._MASTER_JOBS
             print('IMAGE_LIST_SEQUENTIAL_OK three regions; one worker; optional outputs omitted')
             # Same public queue path, now with master participation enabled. The
@@ -504,6 +558,7 @@ async def coordinator():
                 else:
                     assert not remote_calls and len(local_calls) == COUNT
                 assert Counter(local_calls + remote_calls) == Counter(range(COUNT)), (local_calls, remote_calls)
+                check_carried_values()
                 images, masks, texts = CAPTURED[-1]
                 assert len(images) == len(masks) == len(texts) == COUNT
                 for i, row in enumerate(source_rows()):
@@ -513,6 +568,7 @@ async def coordinator():
                 assert not runtime._MASTER_JOBS and not runtime._WORKER_ITEMS
                 participation_counts[mode] = {'master': len(local_calls), 'workers': len(remote_calls)}
             print('IMAGE_LIST_MASTER_OK ' + json.dumps(participation_counts))
+            print('IMAGE_LIST_VALUES_OK passthrough all six types; sparse override/add; Get; sequential Get-to-Start; direct/mixed/master-only')
             assert len(outer_executions) == 4 and all(row['captures'] == 1 for row in outer_executions)
             print('IMAGE_LIST_TARGETS_OK ' + json.dumps({
                 'modes': ['delegate-only', 'mixed', 'master-only'],
@@ -527,6 +583,12 @@ async def coordinator():
             del direct_prompt['8'], direct_prompt['9']
             direct_prompt['3']['inputs']['image'] = ['2', 0]
             # Real Comfy user-interrupt flag while a real worker executor is active.
+            rows = source_rows()
+            items = runtime.normalize_image_list_inputs(
+                [row['image'] for row in rows], [row['mask'] for row in rows],
+                {f'value_{n}': [row[key] for row in rows]
+                 for n, key in ((1, 'reference_image'), (2, 'text'), (3, 'strength'), (4, 'seed'), (6, 'mask'))}
+                | {'value_5': True})
             slow = workers[1]
             async with aiohttp.ClientSession() as session:
                 async def slow_stats():
@@ -540,11 +602,11 @@ async def coordinator():
                         await asyncio.sleep(0.02)
                     raise AssertionError('worker state transition timed out')
                 before = await until(lambda state: not state['running'] and not state['pending'])
-                context = runtime.create_master_job('2', '4', direct_prompt, source_rows(), ['slow'],
+                context = runtime.create_master_job('2', '4', direct_prompt, items, ['slow'],
                                                   f'http://127.0.0.1:{port}')
                 with patch.object(config_mod, 'load_config', return_value=config), \
                      patch.object(config_mod, 'get_worker_timeout_seconds', return_value=15):
-                    task = asyncio.create_task(runtime.run_image_list_job(context))
+                    task = asyncio.create_task(runtime.run_image_list_job(context, include_values=True))
                     active = await until(lambda state: bool(state['running']))
                     owned_id = active['running'][0]
                     comfy.model_management.interrupt_current_processing(True)
@@ -562,14 +624,14 @@ async def coordinator():
                 assert not runtime._MASTER_JOBS
             print('IMAGE_LIST_CANCEL_OK owned prompt stopped; no further dispatch; context removed')
             # A worker exception before End must fail, not hang or return a partial list.
-            bad_rows = source_rows()
-            bad_rows[0]['seed'] = -1
+            bad_rows = copy.deepcopy(items)
+            bad_rows[0]['values']['value_4'] = -1
             context = runtime.create_master_job('2', '4', direct_prompt, bad_rows, ['fast'],
                                               f'http://127.0.0.1:{port}')
             with patch.object(config_mod, 'load_config', return_value=config), \
                  patch.object(config_mod, 'get_worker_timeout_seconds', return_value=15):
                 try:
-                    await asyncio.wait_for(runtime.run_image_list_job(context), 5)
+                    await asyncio.wait_for(runtime.run_image_list_job(context, include_values=True), 5)
                     raise AssertionError('worker exception was ignored')
                 except RuntimeError as exc:
                     assert 'failed image-list item' in str(exc), str(exc)

tests/helpers/v3_runtime_check.py

@@ -45,7 +45,8 @@ ADDED_OPTIONAL_INPUTS = {
 
 # Nodes added after the V1 baseline; not part of saved-workflow contracts.
 # DistributedWorkerSink is internal (dev-only) and only inserted by the master.
-ADDED_NODES = {'DistributedWorkerSink', 'DistributedImageListStart', 'DistributedImageListEnd'}
+ADDED_NODES = {'DistributedWorkerSink', 'DistributedImageListStart', 'DistributedImageListEnd',
+               'DistributedImageListGetValue'}
 
 
 def normalized_input(value):
@@ -273,7 +274,7 @@ async def check_image_list_lazy_executor(mapping, module, prompt_server, asset_m
     original_contexts = set(runtime._MASTER_JOBS)
     image_list_nodes = sys.modules[module.__name__ + '.nodes.image_list']
     workflow_meta = {'nodes': [{'id': 2, 'type': 'DistributedImageListStart'}], 'extra': {'acceptance': True}}
-    async def fake_worker_results(_context_id, *, model_callback=None):
+    async def fake_worker_results(_context_id, *, model_callback=None, include_values=False):
         state = runtime.get_master_job(_context_id)
         assert state.prompt_id == 'image-list-lazy', state.prompt_id
         assert state.workflow_meta == workflow_meta, state.workflow_meta
@@ -281,6 +282,7 @@ async def check_image_list_lazy_executor(mapping, module, prompt_server, asset_m
         return (
             [torch.full((1, 8, 8, 3), value, dtype=torch.float32) for value in (0.125, 0.5, 0.875)],
             [None, None, None], [None, None, None],
+            [{}, {}, {}],
         )
 
     graph = {
@@ -366,7 +368,7 @@ async def main():
     extension = await module.comfy_entrypoint()
     classes = await extension.get_node_list()
     mapping = {cls.GET_SCHEMA().node_id: cls for cls in classes}
-    assert len(classes) == len(mapping) == len(BASELINE) + len(ADDED_NODES) == 11
+    assert len(classes) == len(mapping) == len(BASELINE) + len(ADDED_NODES) == 12
     assert set(mapping) == set(BASELINE) | ADDED_NODES
     for node_id in ADDED_NODES:
         assert nodes.NODE_CLASS_MAPPINGS[node_id] is mapping[node_id]
@@ -423,6 +425,37 @@ async def main():
     assert start_schema.display_name == 'Distributed Image List Start'
     assert end_schema.node_id == 'DistributedImageListEnd'
     assert end_schema.display_name == 'Distributed Image List End'
+    runtime = sys.modules[module.__name__ + '.image_list.runtime']
+    assert [input.id for input in end_schema.inputs] == ['images', 'masks', 'texts', 'context'] + list(runtime.VALUE_NAMES)
+    assert [output.io_type for output in end_schema.outputs] == [
+        'IMAGE', 'MASK', 'STRING', 'DISTRIBUTED_IMAGE_LIST_VALUES']
+    assert end_schema.outputs[3].id == end_schema.outputs[3].display_name == 'values'
+    assert all(output.is_output_list for output in end_schema.outputs)
+    for override in end_schema.inputs[4:]:
+        assert override.optional and override.lazy
+        assert override.get_io_type() == 'IMAGE,MASK,STRING,INT,FLOAT,BOOLEAN'
+    getter = mapping['DistributedImageListGetValue']
+    getter_schema = getter.GET_SCHEMA()
+    assert getter_schema.display_name == 'Distributed Image List Get Value'
+    assert not getter_schema.is_input_list
+    assert getter_schema.outputs[0].id == getter_schema.outputs[0].display_name == 'value'
+    assert getter_schema.outputs[0].io_type == '*'
+    assert getter_schema.inputs[1].options == list(runtime.VALUE_NAMES)
+    assert getter_schema.inputs[1].default == 'value_1'
+    # Real core maps scalar mappings, never arbitrary tensor batches.
+    tensor = torch.ones(3, 2, 2, 3)
+    data, missing, hidden = execution.get_input_data({'value_name': 'value_1'}, getter, 'get-value')
+    data['values'] = [{'value_1': tensor}, {'value_1': 17}, {'value_1': False}]
+    mapped = await execution._async_map_node_over_list('get-value', 'get-value', getter, data,
+                                                       getter.FUNCTION, v3_data=hidden)
+    mapped = await execution.resolve_map_node_over_list_results(mapped)
+    assert len(mapped) == 3 and mapped[0].result[0] is tensor
+    assert mapped[1].result == (17,) and mapped[2].result == (False,)
+    try:
+        getter.execute({}, 'value_1')
+        raise AssertionError('missing Get Value key accepted')
+    except ValueError as exc:
+        assert 'absent' in str(exc)
     texts_input = next(input for input in end_schema.inputs if input.id == 'texts')
     texts_schema = texts_input.as_dict()
     assert texts_schema.get('forceInput') is True, texts_schema
@@ -466,7 +499,6 @@ async def main():
     rows = runtime_mod.normalize_image_list_inputs(prepared['images'], None, seen['values'])
     assert [row['values'] for row in rows] == [{'value_1': 'shared', 'value_3': n} for n in (4, 5, 6)]
     print('IMAGE_LIST_AUTOGROW_OK real parser; value_N slots shared/per-item')
-    assert [out.io_type for out in end_schema.outputs] == ['IMAGE', 'MASK', 'STRING']
     assert all(out.is_output_list for out in end_schema.outputs)
     assert end_schema.is_input_list and end_schema.is_output_node
     print('IMAGE_LIST_SCHEMA_OK exact IDs, names, typed list outputs')

tests/test_dispatch_selection.py

@@ -472,6 +472,26 @@ class DispatchSelectionTests(unittest.IsolatedAsyncioTestCase):
                     with self.assertRaisesRegex(RuntimeError, "execution_targets.*upgrade"):
                         await dispatch.preflight_image_list_workers([{"id": "worker"}])
 
+    async def test_image_list_value_results_requires_exact_true_only_for_overrides(self):
+        for value in (None, False, 1, "true", [], True):
+            capability = _valid_capability()
+            if value is not None:
+                capability['value_results'] = value
+            session = _FakeSession({('worker', 'capabilities'): (200, capability),
+                                    ('worker', 'queue'): (200, _valid_queue())})
+            async def fake_session():
+                return session
+            with patch.object(dispatch, 'get_client_session', side_effect=fake_session), patch.object(
+                dispatch, 'build_worker_url', side_effect=lambda worker, path: f"http://{worker['id']}{path}"
+            ):
+                self.assertEqual(await dispatch.preflight_image_list_workers([{'id': 'worker'}]), {'worker': True})
+                if value is True:
+                    self.assertEqual(await dispatch.preflight_image_list_workers(
+                        [{'id': 'worker'}], require_value_results=True), {'worker': True})
+                else:
+                    with self.assertRaisesRegex(RuntimeError, 'value_results.*upgrade'):
+                        await dispatch.preflight_image_list_workers([{'id': 'worker'}], require_value_results=True)
+
     async def test_image_list_preflight_fails_closed_for_malformed_capabilities(self):
         malformed = {"supported": True, "protocol_version": 2, "node_ids": None}
         session = _FakeSession({("worker", "capabilities"): (200, malformed)})

tests/test_image_list_master_runtime.py

@@ -199,6 +199,73 @@ class MasterRuntimeTests(unittest.IsolatedAsyncioTestCase):
         with self.assertRaisesRegex(RuntimeError, 'worker|fallback'):
             self.rt.create_master_job('start', 'end', graph(), self.items, [], '')
 
+    async def test_values_reconstructed_in_source_order_without_transporting_originals(self):
+        self.workers.append({'id': 'remote', 'type': 'remote'})
+        context = self.create(count=6)
+        tensor = torch.ones(4, 2, 2, 3, dtype=torch.float64)
+        for index, item in enumerate(self.items):
+            item['values'].update(value_2=tensor, value_3=True, value_4=index / 8)
+        self.local_delay = 0.1  # Remote results arrive ahead of master item zero.
+        with patch.object(self.rt, 'encode_result', side_effect=AssertionError('originals must not return')):
+            result = await self.rt.run_image_list_job(context, include_values=True)
+        self.assertFalse(self.rt._MASTER_JOBS)
+        self.assertEqual(len(result), 4)
+        for index, row in enumerate(result[3]):
+            self.assertEqual(row['value_1'], str(index))
+            self.assertIs(row['value_2'], tensor)
+            self.assertIs(row['value_3'], True)
+            self.assertEqual(row['value_4'], index / 8)
+        self.assertTrue(self.dispatched and self.local_calls)
+
+    async def test_overrides_merge_add_and_recheck_capability_before_every_item(self):
+        self.workers.append({'id': 'remote', 'type': 'remote'})
+        context = self.create(participates=False, count=3)
+        self.state.original_prompt['end']['inputs'].update(value_1=['start', 4], value_10=['start', 2])
+        original_set = self.rt.set_master_result
+        def override(context, index, token, result):
+            result['values'] = {'value_1': f'new-{index}', 'value_10': index}
+            return original_set(context, index, token, result)
+        with patch.object(self.rt, 'set_master_result', side_effect=override):
+            result = await self.rt.run_image_list_job(context, include_values=True)
+        self.assertEqual(result[3], [{'value_1': f'new-{index}', 'value_10': index} for index in range(3)])
+        self.assertEqual(self.preflight.await_count, 4)
+        self.assertTrue(all(call.kwargs['require_value_results'] is True
+                            for call in self.preflight.await_args_list))
+        self.assertEqual(self.items[0]['values'], {'value_1': '0'})
+
+    async def test_override_capability_downgrade_stops_before_upload(self):
+        self.workers.append({'id': 'remote', 'type': 'remote'})
+        context = self.create(participates=False)
+        self.state.original_prompt['end']['inputs']['value_1'] = ['start', 4]
+        self.preflight.side_effect = [{'remote': True}, RuntimeError('value_results unsupported')]
+        with patch.object(self, 'post', wraps=self.post) as upload:
+            with self.assertRaisesRegex(RuntimeError, 'value_results'):
+                await self.rt.run_image_list_job(context, include_values=True)
+            upload.assert_not_called()
+        self.assertFalse(self.dispatched or self.rt._MASTER_JOBS)
+
+    async def test_missing_expected_override_fails_without_partial_output(self):
+        self.workers.append({'id': 'remote', 'type': 'remote'})
+        context = self.create(participates=False)
+        self.state.original_prompt['end']['inputs']['value_1'] = ['start', 4]
+        with self.assertRaisesRegex(ValueError, 'missing'):
+            await self.rt.run_image_list_job(context, include_values=True)
+        self.assertFalse(self.rt._MASTER_JOBS)
+
+    async def test_values_failure_and_cancellation_return_no_partial_lists(self):
+        self.local_error = RuntimeError('branch failed')
+        with self.assertRaisesRegex(RuntimeError, 'branch failed'):
+            await self.rt.run_image_list_job(self.create(), include_values=True)
+        self.local_error = None
+        self.local_started.clear()
+        self.local_delay = 0.1
+        task = asyncio.create_task(self.rt.run_image_list_job(self.create(), include_values=True))
+        await self.local_started.wait()
+        task.cancel()
+        with self.assertRaises(self.rt.ImageListCancelledError):
+            await task
+        self.assertFalse(self.rt._MASTER_JOBS)
+
     async def test_local_failure_surfaces_and_reclaims_staged_token(self):
         self.local_error = RuntimeError('local branch failed')
         with self.assertRaisesRegex(RuntimeError, 'local branch failed'):

tests/test_image_list_master_selection.py

@@ -62,6 +62,16 @@ class MasterSelectionTests(unittest.IsolatedAsyncioTestCase):
         self.q.dispatch_worker_prompt.assert_not_awaited()
         self.assertEqual(self.ps.distributed_pending_jobs, {})
 
+    async def test_end_overrides_require_value_capability_before_master_queue(self):
+        self.prompt['3']['inputs']['value_10'] = ['2', 2]
+        await self.queue()
+        self.assertIs(self.preflight.call_args.kwargs['require_value_results'], True)
+        self.q.queue_prompt_payload.reset_mock()
+        self.preflight.side_effect = RuntimeError('value_results unsupported')
+        with self.assertRaisesRegex(RuntimeError, 'value_results'):
+            await self.queue()
+        self.q.queue_prompt_payload.assert_not_awaited()
+
     async def test_orchestrator_setting_excludes_master_processing(self):
         self.config["settings"]["master_delegate_only"] = True
         await self.queue()

tests/test_image_list_values.py

@@ -0,0 +1,165 @@
+"""Context-carried values and sparse, per-item End overrides."""
+import copy
+from unittest.mock import patch
+
+import pytest
+import torch
+
+from image_list import runtime as rt
+from image_list.execution_plan import prepare_image_list_execution
+
+
+def value_graph():
+    # Use a minimal, closed region with an override-only ancestor.
+    return {
+        'source': {'class_type': 'Source', 'inputs': {}},
+        'start': {'class_type': rt.START_CLASS, 'inputs': {'images': ['source', 0]}},
+        'override': {'class_type': 'Transform', 'inputs': {'value': ['start', 4]}},
+        'end': {'class_type': rt.END_CLASS, 'inputs': {
+            'images': ['start', 0], 'context': ['start', 3], 'value_1': ['override', 0],
+            'value_10': ['start', 5]}},
+    }
+
+
+def values():
+    return {'value_1': torch.arange(72, dtype=torch.float64).reshape(2, 3, 4, 3),
+            'value_2': torch.ones(3, 2, 2, dtype=torch.bfloat16), 'value_3': 'name',
+            'value_4': 42, 'value_5': 0.125, 'value_6': True}
+
+
+def assert_values(actual, expected):
+    assert actual.keys() == expected.keys()
+    for key, value in expected.items():
+        assert type(actual[key]) is type(value)
+        if isinstance(value, torch.Tensor):
+            assert actual[key].dtype == value.dtype
+            assert torch.equal(actual[key], value)
+        else:
+            assert actual[key] == value
+
+
+def test_normalize_and_roundtrip_all_override_kinds_keeps_batches_whole():
+    expected = values()
+    result = rt.normalize_worker_result(torch.zeros(1, 2, 2, 3),
+                                       values={key: [value] for key, value in expected.items()})
+    assert_values(result['values'], expected)
+    wire = rt.encode_result(result)
+    assert_values(rt.decode_result(wire)['values'], expected)
+    assert wire['values']['value_1']['shape'][0] == 2
+    assert wire['values']['value_2']['shape'][0] == 3
+
+
+@pytest.mark.parametrize('bad', [None, [], [None], [1, 2], [[1]], object(),
+                                     float('nan'), float('inf'), torch.tensor([float('nan')]),
+                                     torch.tensor([float('inf')]), torch.ones(1, dtype=torch.int32)])
+def test_invalid_override_rejected(bad):
+    with pytest.raises((ValueError, TypeError)):
+        rt.normalize_worker_result(torch.zeros(1, 2, 2, 3), values={'value_1': bad})
+
+
+def test_unknown_override_names_and_wire_types_rejected():
+    for name in ('value_0', 'value_11', 'values.value_1'):
+        with pytest.raises(ValueError, match='slot'):
+            rt.normalize_worker_result(torch.zeros(1, 2, 2, 3), values={name: 1})
+    base = rt.encode_result({'image': torch.zeros(1, 2, 2, 3)})
+    for payload in (None, [], {'value_0': {'type': 'INT', 'value': 1}},
+                    {'value_1': {'type': 'LATENT', 'value': {}}},
+                    {'value_1': {'type': 'FLOAT', 'value': float('nan')}},
+                    {'value_1': {'type': 'STRING', 'value': None}},
+                    {'value_1': rt.encode_value(torch.tensor([float('inf')]), 'TENSOR')}):
+        with pytest.raises((ValueError, TypeError)):
+            rt.decode_result({**base, 'values': payload})
+
+
+def test_no_overrides_keeps_legacy_wire_and_does_not_encode_original_tensors():
+    image = torch.zeros(1, 2, 2, 3)
+    result = rt.normalize_worker_result(image)
+    assert set(result) == {'image', 'mask', 'text'}
+    with patch.object(rt, 'encode_value', wraps=rt.encode_value) as encode:
+        wire = rt.encode_result(result)
+    assert set(wire) == {'image', 'mask', 'text'}
+    assert [call.args[1] for call in encode.call_args_list] == ['IMAGE', 'MASK', 'STRING']
+    assert set(rt.decode_result(wire)) == {'image', 'mask', 'text'}
+
+
+def test_override_ancestors_survive_plan_and_worker_prompt():
+    prompt = value_graph()
+    frozen = copy.deepcopy(prompt)
+    prepared, targets = prepare_image_list_execution(prompt)
+    assert prepared == frozen and prompt == frozen
+    assert targets == ['end']
+    region = rt.validate_image_list_regions(prepared)[0]
+    assert 'override' in region['branch_ids']
+    worker = rt.build_worker_prompt(prepared, region, 'token', 'context', 0, 'http://master')
+    assert worker['end']['inputs']['value_1'] == ['override', 0]
+    assert worker['end']['inputs']['value_10'] == ['start', 5]
+    assert 'source' not in worker
+    for bad in ('value_0', 'value_11', 'values.value_1'):
+        invalid = copy.deepcopy(prompt)
+        invalid['end']['inputs'][bad] = ['start', 4]
+        with pytest.raises(ValueError, match='slot'):
+            rt.validate_image_list_regions(invalid)
+
+
+def test_missing_or_unexpected_result_override_fails_before_recording():
+    prompt = value_graph()
+    items = rt.normalize_image_list_inputs(torch.zeros(1, 2, 2, 3), values={'value_1': 'original'})
+    context = rt.create_master_job('start', 'end', prompt, items, ['worker'], '')
+    try:
+        rt.register_assignment(context, 0, 'worker', 'token')
+        for result_values in ({}, {'value_1': 'updated'},
+                              {'value_1': 'updated', 'value_10': 10, 'value_9': False}):
+            with pytest.raises(ValueError, match='override'):
+                rt.set_master_result(context, 0, 'token', {'image': items[0]['image'], 'values': result_values})
+            assert rt.get_master_job(context).received_indexes == set()
+        rt.set_master_result(context, 0, 'token', {'image': items[0]['image'],
+                                                  'values': {'value_1': 'updated', 'value_10': 10}})
+        assert rt.get_master_job(context).received_indexes == {0}
+    finally:
+        rt.remove_master_job(context)
+
+
+def test_override_return_request_budget_is_aggregate():
+    result = rt.normalize_worker_result(torch.zeros(1, 2, 2, 3), values={'value_1': ['x' * 500]})
+    with patch.object(rt, 'MAX_REQUEST_BYTES', 400):
+        with pytest.raises(ValueError, match='request'):
+            rt.encode_request_payload({'result': rt.encode_result(result)})
+
+
+class TestResultHTTP:
+    def test_missing_and_invalid_overrides_fail_through_real_http(self):
+        import asyncio
+        from aiohttp import web
+        from aiohttp.test_utils import TestClient, TestServer
+        from test_image_list_integration_gaps import load_routes
+        async def check():
+            _module, routes = load_routes()
+            app = web.Application()
+            app.add_routes(routes)
+            prompt = value_graph()
+            items = rt.normalize_image_list_inputs(torch.zeros(1, 2, 2, 3), values={'value_1': 'original'})
+            context = rt.create_master_job('start', 'end', prompt, items, ['worker'], '')
+            token = rt.create_worker_task_token()
+            rt.register_assignment(context, 0, 'worker', token)
+            try:
+                async with TestClient(TestServer(app)) as client:
+                    caps = await (await client.get('/distributed/image-list/capabilities')).json()
+                    assert caps['value_results'] is True and caps['protocol_version'] == 2
+                    result = rt.encode_result({'image': items[0]['image']})
+                    envelope = {'context_id': context, 'token': token, 'item_index': 0, 'result': result}
+                    response = await client.post('/distributed/image-list/result', json=envelope)
+                    assert response.status == 400
+                    assert 'missing' in (await response.json())['error']
+                    assert not rt.get_master_job(context).received_indexes
+                    result['values'] = {'value_1': {'type': 'INT', 'value': [1, 2]},
+                                        'value_10': {'type': 'INT', 'value': 10}}
+                    response = await client.post('/distributed/image-list/result', json=envelope)
+                    assert response.status == 400
+                    assert not rt.get_master_job(context).received_indexes
+                    result['values']['value_1'] = {'type': 'STRING', 'value': 'new'}
+                    response = await client.post('/distributed/image-list/result', json=envelope)
+                    assert response.status == 200
+                    assert rt.get_master_job(context).results[0]['values'] == {'value_1': 'new', 'value_10': 10}
+            finally:
+                rt.remove_master_job(context)
+        asyncio.run(check())

tests/test_v3_collector_transfer.py

@@ -69,7 +69,8 @@ def v3(monkeypatch):
     module(f"{PKG}.nodes.utilities")
     module(f"{PKG}.nodes.distributed_upscale", UltimateSDUpscaleDistributed=object)
     module(f"{PKG}.nodes.collector", DistributedCollectorNode=CollectorRuntime)
-    module(f"{PKG}.nodes.image_list", DistributedImageListStart=object, DistributedImageListEnd=object)
+    module(f"{PKG}.nodes.image_list", DistributedImageListStart=object, DistributedImageListEnd=object,
+           DistributedImageListGetValue=object)
     module("comfy", __path__=[])
     module("comfy.samplers", KSampler=types.SimpleNamespace(SAMPLERS=["euler"], SCHEDULERS=["normal"]))
     module("comfy_api", __path__=[])

tests/test_v3_migration.py

@@ -34,6 +34,7 @@ def test_image_list_network_runtime():
     )
     assert result.returncode == 0, result.stdout + '\n' + result.stderr
     assert 'IMAGE_LIST_NETWORK_OK' in result.stdout
+    assert 'IMAGE_LIST_VALUES_OK' in result.stdout
     assert 'IMAGE_LIST_CANCEL_OK' in result.stdout
     assert 'IMAGE_LIST_MASTER_OK' in result.stdout
     assert 'IMAGE_LIST_PREFLIGHT_OK' in result.stdout

web/imageListValues.js

@@ -0,0 +1,174 @@
+import { app } from "/scripts/app.js";
+
+// End's Python schema owns all ten optional override slots. Only presentation
+// grows here: unlike Start's Autogrow inputs, End keys must never compact, since
+// value_3 always overrides source value_3, even if value_2 is disconnected.
+const START = "DistributedImageListStart";
+const END = "DistributedImageListEnd";
+const GET = "DistributedImageListGetValue";
+const PIPE = "DISTRIBUTED_IMAGE_LIST_VALUES";
+const VALUE_TYPES = "IMAGE,MASK,STRING,INT,FLOAT,BOOLEAN";
+const MAX_VALUES = 10;
+const pendingGraphs = new WeakSet();
+
+function ordinal(name) {
+    const match = /^value_([1-9]|10)$/.exec(name ?? "");
+    return match ? Number(match[1]) : null;
+}
+
+function getLink(graph, id) {
+    if (id == null) return undefined;
+    if (typeof graph?.getLink === "function") return graph.getLink(id);
+    return graph?.links?.get?.(id) ?? graph?.links?.[id];
+}
+
+function sourceOf(node, inputName) {
+    const input = node.inputs?.find(slot => slot.name === inputName);
+    const link = getLink(node.graph, input?.link);
+    const source = link && node.graph?.getNodeById?.(link.origin_id);
+    return source ? { node: source, slot: link.origin_slot, link } : null;
+}
+
+function isReroute(node) {
+    return [node?.comfyClass, node?.type].includes("Reroute");
+}
+
+function throughReroutes(source) {
+    const seen = new Set();
+    while (source && isReroute(source.node)) {
+        if (seen.has(source.node)) return null;
+        seen.add(source.node);
+        source = sourceOf(source.node, source.node.inputs?.[0]?.name);
+    }
+    return source;
+}
+
+function sourceType(source, seen) {
+    source = throughReroutes(source);
+    if (!source) return "*";
+    const { node, slot, link } = source;
+    if (node.comfyClass === GET) return getterType(node, seen);
+    const type = node.outputs?.[slot]?.type;
+    return type && type !== "*" ? type : (link.type || "*");
+}
+
+function getterType(node, seen = new Set()) {
+    if (seen.has(node)) return "*";
+    seen.add(node);
+    const name = node.widgets?.find(widget => widget.name === "value_name")?.value;
+    if (!ordinal(name)) return "*";
+    const endSource = throughReroutes(sourceOf(node, "values"));
+    if (endSource?.node.comfyClass !== END || endSource.slot !== 3) return "*";
+    const end = endSource.node;
+    const override = sourceOf(end, name);
+    if (override) return sourceType(override, seen);
+    const startSource = throughReroutes(sourceOf(end, "context"));
+    if (startSource?.node.comfyClass !== START || startSource.slot !== 3) return "*";
+    return sourceType(sourceOf(startSource.node, `values.${name}`), seen);
+}
+
+function compatible(outputType, inputType) {
+    if (!outputType || !inputType || outputType === "*" || inputType === "*") return true;
+    return String(outputType).split(",").some(type => String(inputType).split(",").includes(type));
+}
+
+function syncGetter(node) {
+    const output = node.outputs?.[0];
+    if (!output) return;
+    const type = getterType(node);
+    if (output.type === type) return;
+    output.type = type;
+    for (const id of [...(output.links ?? [])]) {
+        const link = getLink(node.graph, id);
+        if (!link) continue;
+        const target = node.graph?.getNodeById?.(link.target_id);
+        if (target && !compatible(type, target.inputs?.[link.target_slot]?.type)) {
+            target.disconnectInput(link.target_slot);
+        } else {
+            link.type = type;
+        }
+    }
+    node.setDirtyCanvas?.(true, true);
+}
+
+function syncEnd(node) {
+    if (!Array.isArray(node.inputs) || !Array.isArray(node.outputs)) return;
+    // Old saved End nodes have only the three original outputs. Append rather
+    // than replacing any socket, preserving every existing output index/link.
+    if (node.outputs.length === 3) {
+        const shape = node.outputs[0]?.shape;
+        node.addOutput("values", PIPE, shape === undefined ? undefined : { shape });
+    }
+    const connected = node.inputs.filter(input => ordinal(input.name) && input.link != null);
+    const highest = Math.max(0, ...connected.map(input => ordinal(input.name)));
+    const visible = Math.min(MAX_VALUES, Math.max(highest + 1, node._distributedVisibleOverrides ?? 1));
+    for (let i = node.inputs.length - 1; i >= 0; i--) {
+        const input = node.inputs[i];
+        if (ordinal(input.name) > visible && input.link == null) node.removeInput(i);
+    }
+    for (let i = 1; i <= visible; i++) {
+        if (!node.inputs.some(input => input.name === `value_${i}`)) {
+            node.addInput(`value_${i}`, VALUE_TYPES);
+        }
+    }
+    node.setDirtyCanvas?.(true, true);
+}
+
+function schedule(graph) {
+    if (!graph || pendingGraphs.has(graph)) return;
+    pendingGraphs.add(graph);
+    const run = () => {
+        // Keep the guard while syncing, since disconnectInput can re-enter the
+        // connection hook. A fresh user event schedules a fresh traversal.
+        try {
+            for (const node of graph._nodes ?? []) if (node.comfyClass === END) syncEnd(node);
+            for (const node of graph._nodes ?? []) if (node.comfyClass === GET) syncGetter(node);
+        } catch (error) {
+            console.error("[Distributed] Image List values presentation failed:", error);
+        } finally {
+            pendingGraphs.delete(graph);
+        }
+    };
+    // Let native Start Autogrow and its existing output-sync finish first.
+    if (typeof requestAnimationFrame === "function") requestAnimationFrame(() => setTimeout(run, 0));
+    else setTimeout(run, 0);
+}
+
+app.registerExtension({
+    name: "Distributed.ImageListEndValues",
+    nodeCreated(node) {
+        if (![START, END, GET].includes(node.comfyClass) && !isReroute(node)) return;
+        if (node.comfyClass === END) {
+            node.addWidget("button", "Add value override", null, () => {
+                const shown = Math.max(0, ...(node.inputs ?? []).map(input => ordinal(input.name) ?? 0));
+                node._distributedVisibleOverrides = Math.min(MAX_VALUES, shown + 1);
+                schedule(node.graph);
+            }, { serialize: false });
+        }
+        for (const hook of ["onConnectionsChange", "onConfigure", "onAdded", "onWidgetChanged"]) {
+            const original = node[hook];
+            node[hook] = function (...args) {
+                const result = original?.apply(this, args);
+                schedule(this.graph);
+                return result;
+            };
+        }
+        // Widget callbacks are also used by frontend versions that do not call
+        // the node's onWidgetChanged hook for a combo selection.
+        if (node.comfyClass === GET) {
+            const widget = node.widgets?.find(widget => widget.name === "value_name");
+            if (widget) {
+                const callback = widget.callback;
+                widget.callback = function (...args) {
+                    const result = callback?.apply(this, args);
+                    schedule(node.graph);
+                    return result;
+                };
+            }
+        }
+        schedule(node.graph);
+    },
+    loadedGraphNode(node) {
+        schedule(node.graph);
+    },
+});

web/tests/imageListValues.test.js

@@ -0,0 +1,190 @@
+import { readFileSync } from "node:fs";
+import vm from "node:vm";
+import { describe, expect, it } from "vitest";
+
+const source = readFileSync(new URL("../imageListValues.js", import.meta.url), "utf8")
+    .replace(/^import .*;\r?\n/gm, "");
+const PIPE = "DISTRIBUTED_IMAGE_LIST_VALUES";
+
+function setup() {
+    let extension;
+    let nextId = 1;
+    const tasks = [];
+    const graph = {
+        _nodes: [], links: new Map(),
+        getNodeById(id) { return this._nodes.find(node => node.id === id); },
+        getLink(id) { return this.links.get(id); },
+    };
+    vm.runInNewContext(source, {
+        app: { registerExtension: ext => { extension = ext; } },
+        setTimeout: callback => tasks.push(callback), console,
+    });
+    function node(comfyClass, inputs, outputs, widgets = []) {
+        const n = {
+            id: nextId++, comfyClass, graph, widgets,
+            inputs: inputs.map(name => ({ name, type: "*", link: null })),
+            outputs: outputs.map(([name, type]) => ({ name, type, links: [], shape: 6 })),
+            addWidget(type, name, value, callback, options) { this.widgets.push({ type, name, value, callback, options }); },
+            addInput(name, type) { this.inputs.push({ name, type, link: null }); },
+            addOutput(name, type, extra) { this.outputs.push({ name, type, links: [], ...extra }); },
+            removeInput(slot) {
+                if (this.inputs[slot].link != null) throw new Error("Removed wired input");
+                this.inputs.splice(slot, 1);
+                this.inputs.forEach((input, index) => {
+                    if (input.link != null) graph.links.get(input.link).target_slot = index;
+                });
+            },
+            disconnectInput(slot) {
+                const input = this.inputs[slot];
+                const link = graph.links.get(input.link);
+                if (link) {
+                    const out = graph.getNodeById(link.origin_id).outputs[link.origin_slot];
+                    out.links = out.links.filter(id => id !== link.id);
+                    graph.links.delete(link.id);
+                }
+                input.link = null;
+                this.onConnectionsChange?.();
+            },
+            setDirtyCanvas() {},
+        };
+        graph._nodes.push(n);
+        extension.nodeCreated(n);
+        return n;
+    }
+    const flush = () => { while (tasks.length) tasks.shift()(); };
+    const connect = (from, slot, to, name) => {
+        const targetSlot = to.inputs.findIndex(input => input.name === name);
+        if (targetSlot < 0) throw new Error(`Missing input ${name}`);
+        to.disconnectInput(targetSlot);
+        const id = nextId++;
+        graph.links.set(id, { id, origin_id: from.id, origin_slot: slot, target_id: to.id,
+            target_slot: targetSlot, type: from.outputs[slot].type });
+        from.outputs[slot].links.push(id);
+        to.inputs[targetSlot].link = id;
+        from.onConnectionsChange?.();
+        to.onConnectionsChange?.();
+        return id;
+    };
+    const start = node("DistributedImageListStart", ["images", "mask", "values.value_1", "values.value_2", "values.value_3"],
+        [["image", "IMAGE"], ["mask", "MASK"], ["item_index", "INT"], ["context", "DISTRIBUTED_IMAGE_LIST_CONTEXT"]]);
+    const end = node("DistributedImageListEnd", ["images", "masks", "texts", "context", ...Array.from({ length: 10 }, (_, i) => `value_${i + 1}`)],
+        [["image_list", "IMAGE"], ["masks", "MASK"], ["texts", "STRING"], ["values", PIPE]]);
+    const get = node("DistributedImageListGetValue", ["values"], [["value", "*"]], [{ name: "value_name", value: "value_1" }]);
+    connect(start, 3, end, "context");
+    connect(end, 3, get, "values");
+    flush();
+    return { node, graph, start, end, get, flush, connect, extension };
+}
+
+const valueNames = node => node.inputs.filter(input => input.name.startsWith("value_")).map(input => input.name);
+const addOverride = h => { h.end.widgets.find(widget => widget.name === "Add value override").callback(); h.flush(); };
+
+describe("Image List End context values", () => {
+    it("keeps original sockets and shows only the first optional override initially", () => {
+        const h = setup();
+        expect(h.end.inputs.map(input => input.name)).toEqual(["images", "masks", "texts", "context", "value_1"]);
+        expect(h.end.outputs.map(output => output.name)).toEqual(["image_list", "masks", "texts", "values"]);
+        expect(h.end.widgets[0].options.serialize).toBe(false);
+    });
+
+    it("appends the values output when loading an old three-output End", () => {
+        const h = setup();
+        const old = h.node("DistributedImageListEnd", ["images", "masks", "texts", "context"],
+            [["image_list", "IMAGE"], ["masks", "MASK"], ["texts", "STRING"]]);
+        h.flush();
+        expect(old.outputs[3]).toMatchObject({ name: "values", type: PIPE, shape: 6 });
+        expect(valueNames(old)).toEqual(["value_1"]);
+    });
+
+    it("grows overrides without compacting field identities or existing links", () => {
+        const h = setup();
+        addOverride(h);
+        addOverride(h);
+        const src = h.node("PrimitiveInt", [], [["INT", "INT"]]);
+        const thirdLink = h.connect(src, 0, h.end, "value_3");
+        h.connect(src, 0, h.end, "value_2");
+        h.flush();
+        h.end.disconnectInput(h.end.inputs.findIndex(input => input.name === "value_2"));
+        h.flush();
+        expect(valueNames(h.end)).toEqual(["value_1", "value_2", "value_3", "value_4"]);
+        const link = h.graph.links.get(thirdLink);
+        expect(h.end.inputs[link.target_slot].name).toBe("value_3");
+        for (let i = 0; i < 20; i++) addOverride(h);
+        expect(valueNames(h.end)).toHaveLength(10);
+    });
+
+    it.each(["IMAGE", "MASK", "STRING", "INT", "FLOAT", "BOOLEAN"])("infers %s from the original Start value through context", type => {
+        const h = setup();
+        const source = h.node("Source", [], [["value", type]]);
+        h.connect(source, 0, h.start, "values.value_1");
+        h.flush();
+        expect(h.get.outputs[0].type).toBe(type);
+    });
+
+    it("prefers the End override type, then returns to the original after disconnect", () => {
+        const h = setup();
+        h.connect(h.node("Source", [], [["value", "INT"]]), 0, h.start, "values.value_1");
+        h.connect(h.node("Source", [], [["value", "STRING"]]), 0, h.end, "value_1");
+        h.flush();
+        expect(h.get.outputs[0].type).toBe("STRING");
+        h.end.disconnectInput(h.end.inputs.findIndex(input => input.name === "value_1"));
+        h.flush();
+        expect(h.get.outputs[0].type).toBe("INT");
+    });
+
+    it("updates a selected field and disconnects incompatible consumers", () => {
+        const h = setup();
+        h.connect(h.node("Source", [], [["value", "STRING"]]), 0, h.start, "values.value_1");
+        h.connect(h.node("Source", [], [["value", "INT"]]), 0, h.start, "values.value_2");
+        h.flush();
+        const consumer = h.node("Save", ["filename"], []);
+        consumer.inputs[0].type = "STRING";
+        h.connect(h.get, 0, consumer, "filename");
+        h.get.widgets[0].value = "value_2";
+        h.get.widgets[0].callback();
+        h.flush();
+        expect(h.get.outputs[0].type).toBe("INT");
+        expect(consumer.inputs[0].link).toBe(null);
+    });
+
+    it("keeps compatible consumer links and uses wildcard for unavailable graph metadata", () => {
+        const h = setup();
+        h.connect(h.node("Source", [], [["value", "INT"]]), 0, h.start, "values.value_1");
+        h.flush();
+        const consumer = h.node("Consumer", ["value"], []);
+        consumer.inputs[0].type = "INT,FLOAT";
+        const id = h.connect(h.get, 0, consumer, "value");
+        h.get.disconnectInput(0);
+        h.flush();
+        expect(h.get.outputs[0].type).toBe("*");
+        expect(h.graph.links.has(id)).toBe(true);
+    });
+
+    it("follows reroutes for context and values, and terminates on a cycle", () => {
+        const h = setup();
+        h.connect(h.node("Source", [], [["value", "FLOAT"]]), 0, h.start, "values.value_1");
+        const reroute = h.node("Reroute", ["input"], [["output", "*"]]);
+        h.connect(h.end, 3, reroute, "input");
+        h.connect(reroute, 0, h.get, "values");
+        h.flush();
+        expect(h.get.outputs[0].type).toBe("FLOAT");
+        h.connect(reroute, 0, reroute, "input");
+        h.flush();
+        expect(h.get.outputs[0].type).toBe("*");
+    });
+
+    it("recomputes types across sequential regions", () => {
+        const h = setup();
+        const secondStart = h.node("DistributedImageListStart", ["values.value_1"],
+            [["image", "IMAGE"], ["mask", "MASK"], ["item_index", "INT"], ["context", "DISTRIBUTED_IMAGE_LIST_CONTEXT"]]);
+        const secondEnd = h.node("DistributedImageListEnd", ["images", "masks", "texts", "context"],
+            [["image_list", "IMAGE"], ["masks", "MASK"], ["texts", "STRING"], ["values", PIPE]]);
+        const secondGet = h.node("DistributedImageListGetValue", ["values"], [["value", "*"]], [{ name: "value_name", value: "value_1" }]);
+        h.connect(h.node("Source", [], [["value", "BOOLEAN"]]), 0, h.start, "values.value_1");
+        h.connect(h.get, 0, secondStart, "values.value_1");
+        h.connect(secondStart, 3, secondEnd, "context");
+        h.connect(secondEnd, 3, secondGet, "values");
+        h.flush();
+        expect(secondGet.outputs[0].type).toBe("BOOLEAN");
+    });
+});