Merge 9211284444 into 51bf508a0b

* fix(Video): don't crash on videos with undecodable audio streams

Signed-off-by: bigcat88 <bigcat88@icloud.com>

* Update comfy_api_nodes/util/upload_helpers.py

---------

Signed-off-by: bigcat88 <bigcat88@icloud.com>
Co-authored-by: Alexis Rolland <alexisrolland@hotmail.com>

.github/workflows/cla.yml

@@ -0,0 +1,91 @@
+name: CLA Assistant
+
+on:
+  issue_comment:
+    types: [created]
+  pull_request_target:
+    types: [opened, synchronize, closed]
+
+permissions:
+  actions: write
+  contents: read # 'read' is enough because signatures live in a REMOTE repo
+  pull-requests: write
+  statuses: write
+
+jobs:
+  cla-assistant:
+    runs-on: ubuntu-latest
+    steps:
+      # The CLA action normally requires every commit author in a PR to sign.
+      # We only want the PR author to sign, so we allowlist all other committers
+      # by computing them from the PR's commits and excluding the PR author.
+      - name: Build author-only allowlist
+        id: allowlist
+        if: >
+          github.event_name == 'pull_request_target' ||
+          (github.event_name == 'issue_comment' && github.event.issue.pull_request && (
+            github.event.comment.body == 'recheck' ||
+            github.event.comment.body == 'I have read and agree to the Contributor License Agreement'
+          ))
+        env:
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+          PR_NUMBER: ${{ github.event.pull_request.number || github.event.issue.number }}
+          PR_AUTHOR: ${{ github.event.pull_request.user.login || github.event.issue.user.login }}
+          BASE_ALLOWLIST: action@github.com,actions-user,ampagent,claude,comfy-pr-bot,GitHub Action,github-actions,github-actions[bot],Glary Bot,Glary-Bot,*[bot]
+        run: |
+          others=$(gh api "repos/${{ github.repository }}/pulls/${PR_NUMBER}/commits" --paginate \
+            --jq '.[] | (.author.login // empty), (.committer.login // empty)' \
+            | sort -u | grep -vix "${PR_AUTHOR}" | paste -sd, -)
+          if [ -n "$others" ]; then
+            echo "allowlist=${BASE_ALLOWLIST},${others}" >> "$GITHUB_OUTPUT"
+          else
+            echo "allowlist=${BASE_ALLOWLIST}" >> "$GITHUB_OUTPUT"
+          fi
+
+      - name: CLA Assistant
+        # Run on PR events, on "recheck" comment, or when someone posts the exact signing phrase.
+        # IMPORTANT: this phrase must match `custom-pr-sign-comment` below.
+        if: >
+          github.event_name == 'pull_request_target' ||
+          (github.event_name == 'issue_comment' && github.event.issue.pull_request && (
+            github.event.comment.body == 'recheck' ||
+            github.event.comment.body == 'I have read and agree to the Contributor License Agreement'
+          ))
+        uses: contributor-assistant/github-action@ca4a40a7d1004f18d9960b404b97e5f30a505a08 # v2.6.1
+        env:
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+          # PAT required to write to the centralized signatures repo.
+          PERSONAL_ACCESS_TOKEN: ${{ secrets.PERSONAL_ACCESS_TOKEN }}
+        with:
+          # Where the CLA document lives (shown to contributors)
+          path-to-document: https://github.com/Comfy-Org/comfy-cla/blob/main/comfyui_icla.md
+
+          # Centralized signature storage
+          remote-organization-name: comfy-org
+          remote-repository-name: comfy-cla
+          path-to-signatures: signatures/cla.json
+          branch: main
+
+          # Only the PR author must sign: bots plus every non-author committer
+          # are allowlisted via the "Build author-only allowlist" step above.
+          # *[bot] is a catch-all for any GitHub App bot account.
+          allowlist: ${{ steps.allowlist.outputs.allowlist }}
+
+          # Custom PR comment messages
+          custom-notsigned-prcomment: |
+            🎉 Thank you for your contribution, we really appreciate it! 🎉
+
+            Like many open source projects, we require contributors to sign our [Contributor License Agreement (CLA)](https://github.com/Comfy-Org/comfy-cla/blob/main/comfyui_icla.md). A CLA makes the ownership of contributions explicit, so contributors and the project share a clear understanding of how the code can be used. By signing, you:
+
+            - Confirm that you own your contribution.
+            - Keep the right to reuse your own code.
+            - Grant us a copyright license to include and share it within our projects.
+
+            CLAs are standard practice across major open source projects including those under the Apache Software Foundation and the Linux Foundation. Ours is based on the Apache Software Foundation's CLA. Most importantly, it would enable us to relicense the project under a more permissive license in the future, giving the project and its community greater flexibility.
+
+            ✍ **To sign, please post a new comment on this PR with exactly the following text:** ✍
+
+          custom-pr-sign-comment: I have read and agree to the Contributor License Agreement
+
+          custom-allsigned-prcomment: |
+            ✅ All contributors have signed the CLA. Thank you! This PR is ready to be merged.

CLAUDE.md

@@ -1 +0,0 @@
-AGENTS.md

comfy/sd.py

@@ -468,6 +468,9 @@ class CLIP:
     def decode(self, token_ids, skip_special_tokens=True):
         return self.tokenizer.decode(token_ids, skip_special_tokens=skip_special_tokens)
 
+    def is_dynamic(self):
+        return self.patcher.is_dynamic()
+
 class VAE:
     def __init__(self, sd=None, device=None, config=None, dtype=None, metadata=None):
         if 'decoder.up_blocks.0.resnets.0.norm1.weight' in sd.keys(): #diffusers format
@@ -1251,6 +1254,8 @@ class VAE:
         except:
             return None
 
+    def is_dynamic(self):
+        return self.patcher.is_dynamic()
 
 class StyleModel:
     def __init__(self, model, device="cpu"):

comfy_api/latest/_input_impl/video_types.py

@@ -281,11 +281,18 @@ class VideoFromFile(VideoInput):
         video_done = False
         audio_done = True
 
-        if len(container.streams.audio):
-            audio_stream = container.streams.audio[-1]
+        # Use the last decodable audio stream. Streams FFmpeg has no decoder for have no codec context,
+        # and decoding their packets crashes the process. (e.g. APAC spatial-audio track in iPhone)
+        audio_stream = next(
+            (s for s in reversed(container.streams.audio) if s.codec_context is not None),
+            None,
+        )
+        if audio_stream is not None:
             streams += [audio_stream]
             resampler = av.audio.resampler.AudioResampler(format='fltp')
             audio_done = False
+        elif len(container.streams.audio):
+            logging.warning("No decodable audio stream found in video; ignoring audio.")
 
         for packet in container.demux(*streams):
             if video_done and audio_done:
@@ -457,10 +464,13 @@ class VideoFromFile(VideoInput):
                         else:
                             output_container.metadata[key] = json.dumps(value)
 
-                # Add streams to the new container
+                # Add streams to the new container. Streams with no codec context cannot be used as an output template.
                 stream_map = {}
                 for stream in streams:
                     if isinstance(stream, (av.VideoStream, av.AudioStream, SubtitleStream)):
+                        if stream.codec_context is None:
+                            logging.warning("Skipping %s stream %d with unsupported codec", stream.type, stream.index)
+                            continue
                         out_stream = output_container.add_stream_from_template(template=stream, opaque=True)
                         stream_map[stream] = out_stream
 

comfy_api/latest/_io.py

@@ -1651,6 +1651,15 @@ class Schema:
     Use this for nodes with interactive/operable UI regions that produce intermediate outputs
     (e.g., Image Crop, Painter) rather than final outputs (e.g., Save Image).
     """
+    lazy_outputs: bool=False
+    """When True, cache will invalidate when output connections change, and expected_outputs will be available.
+
+    Use this for nodes that can skip computing outputs that aren't connected downstream.
+    Check `comfy_execution.utils.is_output_needed(i)` inside execute() - False means output i is definitely unused
+    and safe to skip. Only nodes with this flag receive expected_outputs; all others see None.
+
+    Limitation: consumers must exist before this node runs - a subgraph expansion that
+    hand-builds a link to a pre-existing node's already-skipped output reads a stale value."""
 
     def validate(self):
         '''Validate the schema:
@@ -2108,6 +2117,14 @@ class _ComfyNodeBaseInternal(_ComfyNodeInternal):
             cls.GET_SCHEMA()
         return cls._ACCEPT_ALL_INPUTS
 
+    _LAZY_OUTPUTS = None
+    @final
+    @classproperty
+    def LAZY_OUTPUTS(cls):  # noqa
+        if cls._LAZY_OUTPUTS is None:
+            cls.GET_SCHEMA()
+        return cls._LAZY_OUTPUTS
+
     @final
     @classmethod
     def INPUT_TYPES(cls) -> dict[str, dict]:
@@ -2152,6 +2169,8 @@ class _ComfyNodeBaseInternal(_ComfyNodeInternal):
             cls._NOT_IDEMPOTENT = schema.not_idempotent
         if cls._ACCEPT_ALL_INPUTS is None:
             cls._ACCEPT_ALL_INPUTS = schema.accept_all_inputs
+        if cls._LAZY_OUTPUTS is None:
+            cls._LAZY_OUTPUTS = schema.lazy_outputs
 
         if cls._RETURN_TYPES is None:
             output = []

comfy_api_nodes/util/upload_helpers.py

@@ -158,7 +158,14 @@ async def upload_video_to_comfyapi(
 
     # Convert VideoInput to BytesIO using specified container/codec
     video_bytes_io = BytesIO()
-    video.save_to(video_bytes_io, format=container, codec=codec)
+    try:
+        video.save_to(video_bytes_io, format=container, codec=codec)
+    except Exception as e:
+        raise ValueError(
+            f"Could not convert the input video to {container.value.upper()} for upload; "
+            f"the file may be corrupted or use an unsupported codec. "
+            f"Try re-exporting it as MP4 (H.264). Original error: {e}"
+        ) from e
     video_bytes_io.seek(0)
 
     return await upload_file_to_comfyapi(cls, video_bytes_io, filename, upload_mime_type, wait_label)

comfy_execution/caching.py

@@ -6,7 +6,7 @@ import time
 import torch
 from typing import Sequence, Mapping, Dict
 from comfy.model_patcher import ModelPatcher
-from comfy_execution.graph import DynamicPrompt
+from comfy_execution.graph import DynamicPrompt, get_expected_outputs_for_node
 from abc import ABC, abstractmethod
 
 import nodes
@@ -116,6 +116,10 @@ class CacheKeySetInputSignature(CacheKeySet):
         signature = [class_type, await self.is_changed_cache.get(node_id)]
         if self.include_node_id_in_input() or (hasattr(class_def, "NOT_IDEMPOTENT") and class_def.NOT_IDEMPOTENT) or include_unique_id_in_input(class_type):
             signature.append(node_id)
+        # Include expected_outputs in cache key for nodes that opt in via LAZY_OUTPUTS
+        if hasattr(class_def, 'LAZY_OUTPUTS') and class_def.LAZY_OUTPUTS:
+            expected = get_expected_outputs_for_node(dynprompt, node_id)
+            signature.append(("expected_outputs", tuple(sorted(expected))))
         inputs = node["inputs"]
         for key in sorted(inputs.keys()):
             if is_link(inputs[key]):
@@ -503,6 +507,21 @@ RAM_CACHE_DEFAULT_RAM_USAGE = 0.05
 
 RAM_CACHE_OLD_WORKFLOW_OOM_MULTIPLIER = 1.3
 
+
+def all_outputs_dynamic(outputs):
+    if outputs is None:
+        return False
+
+    for output in outputs:
+        if isinstance(output, (list, tuple)):
+            if not all_outputs_dynamic(output):
+                return False
+        elif not hasattr(output, "is_dynamic") or not output.is_dynamic():
+            return False
+
+    return True
+
+
 class RAMPressureCache(LRUCache):
 
     def __init__(self, key_class, enable_providers=False):
@@ -533,7 +552,11 @@ class RAMPressureCache(LRUCache):
         for key, cache_entry in self.cache.items():
             if not free_active and self.used_generation[key] == self.generation:
                 continue
-            oom_score =  RAM_CACHE_OLD_WORKFLOW_OOM_MULTIPLIER ** (self.generation - self.used_generation[key])
+
+            if all_outputs_dynamic(cache_entry.outputs) and self.used_generation[key] == self.generation:
+                continue
+
+            oom_score = RAM_CACHE_OLD_WORKFLOW_OOM_MULTIPLIER ** (self.generation - self.used_generation[key])
 
             ram_usage = RAM_CACHE_DEFAULT_RAM_USAGE
             def scan_list_for_ram_usage(outputs):

comfy_execution/graph.py

@@ -18,6 +18,18 @@ class NodeInputError(Exception):
 class NodeNotFoundError(Exception):
     pass
 
+
+def get_expected_outputs_for_node(dynprompt, node_id: str) -> frozenset:
+    """Get the set of output indices that are connected downstream.
+    Returns outputs that MIGHT be used.
+    Outputs NOT in this set are DEFINITELY not used and safe to skip
+    (see Schema.lazy_outputs for the one expansion-related limitation).
+
+    Includes input links and consumers registered via add_output_consumer.
+    """
+    return dynprompt.get_expected_outputs_map().get(node_id, frozenset())
+
+
 class DynamicPrompt:
     def __init__(self, original_prompt):
         # The original prompt provided by the user
@@ -26,6 +38,9 @@ class DynamicPrompt:
         self.ephemeral_prompt = {}
         self.ephemeral_parents = {}
         self.ephemeral_display = {}
+        # Output sockets consumed outside of input links (subgraph expansions)
+        self._external_output_consumers = {}
+        self._expected_outputs_map = None
 
     def get_node(self, node_id):
         if node_id in self.ephemeral_prompt:
@@ -41,6 +56,7 @@ class DynamicPrompt:
         self.ephemeral_prompt[node_id] = node_info
         self.ephemeral_parents[node_id] = parent_id
         self.ephemeral_display[node_id] = display_id
+        self._expected_outputs_map = None
 
     def get_real_node_id(self, node_id):
         while node_id in self.ephemeral_parents:
@@ -58,6 +74,29 @@ class DynamicPrompt:
     def all_node_ids(self):
         return set(self.original_prompt.keys()).union(set(self.ephemeral_prompt.keys()))
 
+    def add_output_consumer(self, node_id, socket):
+        """Record an output socket consumed outside of input links, e.g. a subgraph
+        expansion mapping its parent's output to this node's output."""
+        self._external_output_consumers.setdefault(node_id, set()).add(socket)
+        self._expected_outputs_map = None
+
+    def _build_expected_outputs_map(self):
+        result = {}
+        for node_id in self.all_node_ids():
+            node_data = self.get_node(node_id)
+            for value in node_data.get("inputs", {}).values():
+                if is_link(value):
+                    from_node_id, from_socket = value
+                    result.setdefault(from_node_id, set()).add(from_socket)
+        for node_id, sockets in self._external_output_consumers.items():
+            result.setdefault(node_id, set()).update(sockets)
+        self._expected_outputs_map = {k: frozenset(v) for k, v in result.items()}
+
+    def get_expected_outputs_map(self):
+        if self._expected_outputs_map is None:
+            self._build_expected_outputs_map()
+        return self._expected_outputs_map
+
     def get_original_prompt(self):
         return self.original_prompt
 

comfy_execution/utils.py

@@ -1,23 +1,45 @@
 import contextvars
-from typing import Optional, NamedTuple
+from typing import NamedTuple, FrozenSet
 
 class ExecutionContext(NamedTuple):
     """
     Context information about the currently executing node.
 
     Attributes:
+        prompt_id: The ID of the current prompt execution
         node_id: The ID of the currently executing node
         list_index: The index in a list being processed (for operations on batches/lists)
+        expected_outputs: Set of output indices that might be used downstream.
+                         Outputs NOT in this set are definitely unused (safe to skip).
+                         None means the information is not available.
     """
     prompt_id: str
     node_id: str
-    list_index: Optional[int]
+    list_index: int | None
+    expected_outputs: FrozenSet[int] | None = None
 
-current_executing_context: contextvars.ContextVar[Optional[ExecutionContext]] = contextvars.ContextVar("current_executing_context", default=None)
+current_executing_context: contextvars.ContextVar[ExecutionContext | None] = contextvars.ContextVar("current_executing_context", default=None)
 
-def get_executing_context() -> Optional[ExecutionContext]:
+def get_executing_context() -> ExecutionContext | None:
     return current_executing_context.get(None)
 
+
+def is_output_needed(output_index: int) -> bool:
+    """Check if an output at the given index is connected downstream.
+
+    Returns True if the output might be used (should be computed).
+    Returns False if the output is definitely not connected (safe to skip).
+
+    Only meaningful for LAZY_OUTPUTS nodes; for all others expected_outputs is
+    None and this always returns True (skipping without the flag would not be
+    reflected in the cache key).
+    """
+    ctx = get_executing_context()
+    if ctx is None or ctx.expected_outputs is None:
+        return True
+    return output_index in ctx.expected_outputs
+
+
 class CurrentNodeContext:
     """
     Context manager for setting the current executing node context.
@@ -25,15 +47,22 @@ class CurrentNodeContext:
     Sets the current_executing_context on enter and resets it on exit.
 
     Example:
-        with CurrentNodeContext(node_id="123", list_index=0):
+        with CurrentNodeContext(prompt_id="abc", node_id="123", list_index=0):
             # Code that should run with the current node context set
             process_image()
     """
-    def __init__(self, prompt_id: str, node_id: str, list_index: Optional[int] = None):
+    def __init__(
+        self,
+        prompt_id: str,
+        node_id: str,
+        list_index: int | None = None,
+        expected_outputs: FrozenSet[int] | None = None,
+    ):
         self.context = ExecutionContext(
-            prompt_id= prompt_id,
-            node_id= node_id,
-            list_index= list_index
+            prompt_id=prompt_id,
+            node_id=node_id,
+            list_index=list_index,
+            expected_outputs=expected_outputs,
         )
         self.token = None
 

comfy_extras/nodes_text_overlay.py

@@ -0,0 +1,150 @@
+import numpy as np
+import torch
+from PIL import Image as PILImage, ImageColor, ImageDraw, ImageFont
+from typing_extensions import override
+
+from comfy_api.latest import ComfyExtension, IO
+
+
+class TextOverlay(IO.ComfyNode):
+    @classmethod
+    def define_schema(cls):
+        return IO.Schema(
+            node_id="TextOverlay",
+            display_name="Draw Text Overlay",
+            category="text",
+            description="Draw text overlay on an image or batch of images.",
+            search_aliases=["text", "label", "caption", "subtitle", "watermark", "title", "addlabel", "overlay"],
+            inputs=[
+                IO.Image.Input("images"),
+                IO.String.Input("text", multiline=True, default=""),
+                IO.Float.Input("font_size", default=5.0, min=0.5, max=50.0, step=0.5, tooltip="Font size as a percentage of the image height."),
+                IO.Color.Input("color", default="#ffffff", tooltip="Color of the text."),
+                IO.Combo.Input("position", options=["top", "bottom"], default="top"),
+                IO.Combo.Input("align", options=["left", "center", "right"], default="left"),
+                IO.Boolean.Input("outline", default=True, tooltip="Draw a black outline around the text."),
+            ],
+            outputs=[IO.Image.Output(display_name="images")],
+        )
+
+    @classmethod
+    def execute(cls, images, text, font_size, color, position, align, outline) -> IO.NodeOutput:
+        if text.strip() == "":
+            return IO.NodeOutput(images)
+
+        text = text.replace("\\n", "\n").replace("\\t", "\t")
+
+        text_rgba = cls.parse_color_to_rgba(color)
+        outline_rgba = (0, 0, 0, 255) if outline else (0, 0, 0, 0)
+
+        # Render the overlay once and composite it across all frames in the batch
+        height = images.shape[1]
+        width = images.shape[2]
+        overlay_rgb, overlay_alpha = cls.render_overlay_text(width, height, text, position, align, font_size, text_rgba, outline_rgba)
+        overlay_rgb = overlay_rgb.to(device=images.device, dtype=images.dtype)
+        overlay_alpha = overlay_alpha.to(device=images.device, dtype=images.dtype)
+
+        result = images * (1.0 - overlay_alpha) + overlay_rgb * overlay_alpha
+        return IO.NodeOutput(result)
+
+    @staticmethod
+    def parse_color_to_rgba(color_string):
+        parsed = ImageColor.getrgb(color_string)
+
+        if len(parsed) == 3:
+            return (*parsed, 255)
+
+        return parsed
+
+    @classmethod
+    def render_overlay_text(cls, width, height, text, position, align, font_size, text_rgba, outline_rgba):
+        line_spacing = 1.2
+        margin_percent = 1.0
+        min_font_percent = 2.0
+        min_font_pixels = 10
+        outline_thickness_factor = 0.04
+
+        # Draw onto a transparent layer so the result can be alpha-composited over any frame.
+        layer = PILImage.new("RGBA", (width, height), (0, 0, 0, 0))
+        draw = ImageDraw.Draw(layer)
+
+        margin = int(round(margin_percent / 100.0 * min(width, height)))
+        max_width = max(1, width - 2 * margin)
+        max_height = max(1, height - 2 * margin)
+
+        # Font scales with resolution, then shrinks to fit the height.
+        size = max(1, int(round(font_size / 100.0 * height)))
+        floor = min(size, max(min_font_pixels, int(round(min_font_percent / 100.0 * height))))
+
+        while True:
+            font = ImageFont.load_default(size=size)
+            stroke = max(1, int(round(size * outline_thickness_factor))) if outline_rgba[3] > 0 else 0
+            block = "\n".join(cls.wrap_text(text, font, max_width))
+            # convert line spacing to pixel spacing
+            single = draw.textbbox((0, 0), "Ay", font=font, stroke_width=stroke)
+            double = draw.multiline_textbbox((0, 0), "Ay\nAy", font=font, spacing=0, stroke_width=stroke)
+            natural_advance = (double[3] - double[1]) - (single[3] - single[1])
+            pixel_spacing = int(round(size * line_spacing - natural_advance))
+            box = draw.multiline_textbbox((0, 0), block, font=font, spacing=pixel_spacing, stroke_width=stroke)
+            block_height = box[3] - box[1]
+
+            if block_height <= max_height or size <= floor:
+                break
+
+            size = max(floor, int(size * 0.9))
+
+        anchor_h, x = {"left": ("l", margin), "center": ("m", width / 2), "right": ("r", width - margin)}[align]
+
+        # Offset y so the rendered text sits flush against the margin
+        if position == "bottom":
+            y = height - margin - box[3]
+        else:
+            y = margin - box[1]
+
+        draw.multiline_text((x, y), block, font=font, fill=text_rgba, anchor=anchor_h + "a",
+                            align=align, spacing=pixel_spacing, stroke_width=stroke, stroke_fill=outline_rgba)
+
+        overlay = np.array(layer).astype(np.float32) / 255.0
+        overlay_rgb = torch.from_numpy(overlay[:, :, :3])
+        overlay_alpha = torch.from_numpy(overlay[:, :, 3:4])
+        return overlay_rgb, overlay_alpha
+
+    @staticmethod
+    def wrap_text(text, font, max_width):
+        lines = []
+        for raw_line in text.split("\n"):
+            words = raw_line.split()
+            if not words:
+                lines.append("")
+                continue
+            current = ""
+            # Break the line into words and split words that are too long
+            for word in words:
+                while font.getlength(word) > max_width and len(word) > 1:
+                    cut = 1
+                    while cut < len(word) and font.getlength(word[:cut + 1]) <= max_width:
+                        cut += 1
+                    if current:
+                        lines.append(current)
+                        current = ""
+                    lines.append(word[:cut])
+                    word = word[cut:]
+                candidate = word if not current else current + " " + word
+                if not current or font.getlength(candidate) <= max_width:
+                    current = candidate
+                else:
+                    lines.append(current)
+                    current = word
+            if current:
+                lines.append(current)
+        return lines
+
+
+class TextOverlayExtension(ComfyExtension):
+    @override
+    async def get_node_list(self) -> list[type[IO.ComfyNode]]:
+        return [TextOverlay]
+
+
+async def comfy_entrypoint() -> TextOverlayExtension:
+    return TextOverlayExtension()

execution.py

@@ -35,6 +35,7 @@ from comfy_execution.graph import (
     ExecutionBlocker,
     ExecutionList,
     get_input_info,
+    get_expected_outputs_for_node,
 )
 from comfy_execution.graph_utils import GraphBuilder, is_link
 from comfy_execution.validation import validate_node_input
@@ -237,7 +238,18 @@ async def resolve_map_node_over_list_results(results):
                 raise exc
         return [x.result() if isinstance(x, asyncio.Task) else x for x in results]
 
-async def _async_map_node_over_list(prompt_id, unique_id, obj, input_data_all, func, allow_interrupt=False, execution_block_cb=None, pre_execute_cb=None, v3_data=None):
+async def _async_map_node_over_list(
+    prompt_id,
+    unique_id,
+    obj,
+    input_data_all,
+    func,
+    allow_interrupt=False,
+    execution_block_cb=None,
+    pre_execute_cb=None,
+    v3_data=None,
+    expected_outputs=None,
+):
     # check if node wants the lists
     input_is_list = getattr(obj, "INPUT_IS_LIST", False)
 
@@ -287,10 +299,12 @@ async def _async_map_node_over_list(prompt_id, unique_id, obj, input_data_all, f
             else:
                 f = getattr(obj, func)
             if inspect.iscoroutinefunction(f):
-                async def async_wrapper(f, prompt_id, unique_id, list_index, args):
-                    with CurrentNodeContext(prompt_id, unique_id, list_index):
+                async def async_wrapper(f, prompt_id, unique_id, list_index, args, expected_outputs):
+                    with CurrentNodeContext(prompt_id, unique_id, list_index, expected_outputs):
                         return await f(**args)
-                task = asyncio.create_task(async_wrapper(f, prompt_id, unique_id, index, args=inputs))
+                task = asyncio.create_task(
+                    async_wrapper(f, prompt_id, unique_id, index, args=inputs, expected_outputs=expected_outputs)
+                )
                 # Give the task a chance to execute without yielding
                 await asyncio.sleep(0)
                 if task.done():
@@ -299,7 +313,7 @@ async def _async_map_node_over_list(prompt_id, unique_id, obj, input_data_all, f
                 else:
                     results.append(task)
             else:
-                with CurrentNodeContext(prompt_id, unique_id, index):
+                with CurrentNodeContext(prompt_id, unique_id, index, expected_outputs):
                     result = f(**inputs)
                 results.append(result)
         else:
@@ -337,8 +351,17 @@ def merge_result_data(results, obj):
             output.append([o[i] for o in results])
     return output
 
-async def get_output_data(prompt_id, unique_id, obj, input_data_all, execution_block_cb=None, pre_execute_cb=None, v3_data=None):
-    return_values = await _async_map_node_over_list(prompt_id, unique_id, obj, input_data_all, obj.FUNCTION, allow_interrupt=True, execution_block_cb=execution_block_cb, pre_execute_cb=pre_execute_cb, v3_data=v3_data)
+async def get_output_data(
+    prompt_id,
+    unique_id,
+    obj,
+    input_data_all,
+    execution_block_cb=None,
+    pre_execute_cb=None,
+    v3_data=None,
+    expected_outputs=None,
+):
+    return_values = await _async_map_node_over_list(prompt_id, unique_id, obj, input_data_all, obj.FUNCTION, allow_interrupt=True, execution_block_cb=execution_block_cb, pre_execute_cb=pre_execute_cb, v3_data=v3_data, expected_outputs=expected_outputs)
     has_pending_task = any(isinstance(r, asyncio.Task) and not r.done() for r in return_values)
     if has_pending_task:
         return return_values, {}, False, has_pending_task
@@ -538,8 +561,12 @@ async def execute(server, dynprompt, caches, current_item, extra_data, executed,
                 # TODO - How to handle this with async functions without contextvars (which requires Python 3.12)?
                 GraphBuilder.set_default_prefix(unique_id, call_index, 0)
 
+            if getattr(class_def, "LAZY_OUTPUTS", False):
+                expected_outputs = get_expected_outputs_for_node(dynprompt, unique_id)
+            else:
+                expected_outputs = None
             try:
-                output_data, output_ui, has_subgraph, has_pending_tasks = await get_output_data(prompt_id, unique_id, obj, input_data_all, execution_block_cb=execution_block_cb, pre_execute_cb=pre_execute_cb, v3_data=v3_data)
+                output_data, output_ui, has_subgraph, has_pending_tasks = await get_output_data(prompt_id, unique_id, obj, input_data_all, execution_block_cb=execution_block_cb, pre_execute_cb=pre_execute_cb, v3_data=v3_data, expected_outputs=expected_outputs)
             finally:
                 if comfy.memory_management.aimdo_enabled:
                     if args.verbose == "DEBUG":
@@ -596,6 +623,7 @@ async def execute(server, dynprompt, caches, current_item, extra_data, executed,
                         if is_link(node_outputs[i]):
                             from_node_id, from_socket = node_outputs[i][0], node_outputs[i][1]
                             new_output_links.append((from_node_id, from_socket))
+                            dynprompt.add_output_consumer(from_node_id, from_socket)
                     cached_outputs.append((True, node_outputs))
             new_node_ids = set(new_node_ids)
             for cache in caches.all:

nodes.py

@@ -2478,6 +2478,7 @@ async def init_builtin_extra_nodes():
         "nodes_glsl.py",
         "nodes_lora_debug.py",
         "nodes_textgen.py",
+        "nodes_text_overlay.py",
         "nodes_color.py",
         "nodes_toolkit.py",
         "nodes_replacements.py",

tests-unit/execution_test/expected_outputs_test.py

@@ -0,0 +1,361 @@
+"""Unit tests for the expected_outputs feature.
+
+This feature allows nodes to know at runtime which outputs are connected downstream,
+enabling them to skip computing outputs that aren't needed.
+"""
+
+from comfy_api.latest import IO
+from comfy_execution.graph import DynamicPrompt, get_expected_outputs_for_node
+from comfy_execution.utils import (
+    CurrentNodeContext,
+    ExecutionContext,
+    get_executing_context,
+    is_output_needed,
+)
+
+
+class TestGetExpectedOutputsForNode:
+    """Tests for get_expected_outputs_for_node() function."""
+
+    def test_single_output_connected(self):
+        """Test node with single output connected to one downstream node."""
+        prompt = {
+            "1": {"class_type": "SourceNode", "inputs": {}},
+            "2": {"class_type": "ConsumerNode", "inputs": {"image": ["1", 0]}},
+        }
+        dynprompt = DynamicPrompt(prompt)
+        expected = get_expected_outputs_for_node(dynprompt, "1")
+        assert expected == frozenset({0})
+
+    def test_multiple_outputs_partial_connected(self):
+        """Test node with multiple outputs, only some connected."""
+        prompt = {
+            "1": {"class_type": "MultiOutputNode", "inputs": {}},
+            "2": {"class_type": "ConsumerA", "inputs": {"input": ["1", 0]}},
+            # Output 1 is not connected
+            "3": {"class_type": "ConsumerC", "inputs": {"input": ["1", 2]}},
+        }
+        dynprompt = DynamicPrompt(prompt)
+        expected = get_expected_outputs_for_node(dynprompt, "1")
+        assert expected == frozenset({0, 2})
+        assert 1 not in expected  # Output 1 is definitely unused
+
+    def test_no_outputs_connected(self):
+        """Test node with no outputs connected."""
+        prompt = {
+            "1": {"class_type": "SourceNode", "inputs": {}},
+            "2": {"class_type": "OtherNode", "inputs": {}},
+        }
+        dynprompt = DynamicPrompt(prompt)
+        expected = get_expected_outputs_for_node(dynprompt, "1")
+        assert expected == frozenset()
+
+    def test_same_output_connected_multiple_times(self):
+        """Test same output connected to multiple downstream nodes."""
+        prompt = {
+            "1": {"class_type": "SourceNode", "inputs": {}},
+            "2": {"class_type": "ConsumerA", "inputs": {"input": ["1", 0]}},
+            "3": {"class_type": "ConsumerB", "inputs": {"input": ["1", 0]}},
+            "4": {"class_type": "ConsumerC", "inputs": {"input": ["1", 0]}},
+        }
+        dynprompt = DynamicPrompt(prompt)
+        expected = get_expected_outputs_for_node(dynprompt, "1")
+        assert expected == frozenset({0})
+
+    def test_node_not_in_prompt(self):
+        """Test getting expected outputs for a node not in the prompt."""
+        prompt = {
+            "1": {"class_type": "SourceNode", "inputs": {}},
+        }
+        dynprompt = DynamicPrompt(prompt)
+        expected = get_expected_outputs_for_node(dynprompt, "999")
+        assert expected == frozenset()
+
+    def test_chained_nodes(self):
+        """Test expected outputs in a chain of nodes."""
+        prompt = {
+            "1": {"class_type": "SourceNode", "inputs": {}},
+            "2": {"class_type": "MiddleNode", "inputs": {"input": ["1", 0]}},
+            "3": {"class_type": "EndNode", "inputs": {"input": ["2", 0]}},
+        }
+        dynprompt = DynamicPrompt(prompt)
+
+        # Node 1's output 0 is connected to node 2
+        expected_1 = get_expected_outputs_for_node(dynprompt, "1")
+        assert expected_1 == frozenset({0})
+
+        # Node 2's output 0 is connected to node 3
+        expected_2 = get_expected_outputs_for_node(dynprompt, "2")
+        assert expected_2 == frozenset({0})
+
+        # Node 3 has no downstream connections
+        expected_3 = get_expected_outputs_for_node(dynprompt, "3")
+        assert expected_3 == frozenset()
+
+    def test_complex_graph(self):
+        """Test expected outputs in a complex graph with multiple connections."""
+        prompt = {
+            "1": {"class_type": "MultiOutputNode", "inputs": {}},
+            "2": {"class_type": "ProcessorA", "inputs": {"image": ["1", 0], "mask": ["1", 1]}},
+            "3": {"class_type": "ProcessorB", "inputs": {"data": ["1", 2]}},
+            "4": {"class_type": "Combiner", "inputs": {"a": ["2", 0], "b": ["3", 0]}},
+        }
+        dynprompt = DynamicPrompt(prompt)
+
+        # Node 1 has outputs 0, 1, 2 all connected
+        expected = get_expected_outputs_for_node(dynprompt, "1")
+        assert expected == frozenset({0, 1, 2})
+
+    def test_constant_inputs_ignored(self):
+        """Test that constant (non-link) inputs don't affect expected outputs."""
+        prompt = {
+            "1": {"class_type": "SourceNode", "inputs": {}},
+            "2": {
+                "class_type": "ConsumerNode",
+                "inputs": {
+                    "image": ["1", 0],
+                    "value": 42,
+                    "name": "test",
+                },
+            },
+        }
+        dynprompt = DynamicPrompt(prompt)
+        expected = get_expected_outputs_for_node(dynprompt, "1")
+        assert expected == frozenset({0})
+
+    def test_ephemeral_node_invalidates_cache(self):
+        """Test that adding ephemeral nodes updates expected outputs."""
+        prompt = {
+            "1": {"class_type": "SourceNode", "inputs": {}},
+            "2": {"class_type": "ConsumerNode", "inputs": {"image": ["1", 0]}},
+        }
+        dynprompt = DynamicPrompt(prompt)
+
+        # Initially only output 0 is connected
+        expected = get_expected_outputs_for_node(dynprompt, "1")
+        assert expected == frozenset({0})
+
+        # Add an ephemeral node that connects to output 1
+        dynprompt.add_ephemeral_node(
+            "eph_1",
+            {"class_type": "EphemeralNode", "inputs": {"data": ["1", 1]}},
+            parent_id="2",
+            display_id="2",
+        )
+
+        # Now both outputs 0 and 1 should be expected
+        expected = get_expected_outputs_for_node(dynprompt, "1")
+        assert expected == frozenset({0, 1})
+
+
+class TestExternalOutputConsumers:
+    """Tests for DynamicPrompt.add_output_consumer() — out-of-band consumers
+    (subgraph expansion output mappings) that have no input link in the prompt."""
+
+    def test_external_consumer_only(self):
+        """A socket consumed only externally must appear in expected outputs."""
+        prompt = {
+            "1": {"class_type": "SourceNode", "inputs": {}},
+        }
+        dynprompt = DynamicPrompt(prompt)
+        assert get_expected_outputs_for_node(dynprompt, "1") == frozenset()
+
+        dynprompt.add_output_consumer("1", 1)
+        assert get_expected_outputs_for_node(dynprompt, "1") == frozenset({1})
+
+    def test_external_consumer_merges_with_links(self):
+        """External consumers merge with input-link consumers."""
+        prompt = {
+            "1": {"class_type": "SourceNode", "inputs": {}},
+            "2": {"class_type": "ConsumerNode", "inputs": {"image": ["1", 0]}},
+        }
+        dynprompt = DynamicPrompt(prompt)
+        dynprompt.add_output_consumer("1", 2)
+        assert get_expected_outputs_for_node(dynprompt, "1") == frozenset({0, 2})
+
+    def test_external_consumer_invalidates_cached_map(self):
+        """Registering after the map was built must invalidate the cache."""
+        prompt = {
+            "1": {"class_type": "SourceNode", "inputs": {}},
+            "2": {"class_type": "ConsumerNode", "inputs": {"image": ["1", 0]}},
+        }
+        dynprompt = DynamicPrompt(prompt)
+        # Build (and cache) the map first
+        assert get_expected_outputs_for_node(dynprompt, "1") == frozenset({0})
+
+        dynprompt.add_output_consumer("1", 1)
+        assert get_expected_outputs_for_node(dynprompt, "1") == frozenset({0, 1})
+
+
+class TestExecutionContext:
+    """Tests for ExecutionContext with expected_outputs field."""
+
+    def test_context_with_expected_outputs(self):
+        """Test creating ExecutionContext with expected_outputs."""
+        ctx = ExecutionContext(
+            prompt_id="prompt-123", node_id="node-456", list_index=0, expected_outputs=frozenset({0, 2})
+        )
+        assert ctx.prompt_id == "prompt-123"
+        assert ctx.node_id == "node-456"
+        assert ctx.list_index == 0
+        assert ctx.expected_outputs == frozenset({0, 2})
+
+    def test_context_without_expected_outputs(self):
+        """Test ExecutionContext defaults to None for expected_outputs."""
+        ctx = ExecutionContext(prompt_id="prompt-123", node_id="node-456", list_index=0)
+        assert ctx.expected_outputs is None
+
+    def test_context_empty_expected_outputs(self):
+        """Test ExecutionContext with empty expected_outputs set."""
+        ctx = ExecutionContext(
+            prompt_id="prompt-123", node_id="node-456", list_index=None, expected_outputs=frozenset()
+        )
+        assert ctx.expected_outputs == frozenset()
+        assert len(ctx.expected_outputs) == 0
+
+
+class TestCurrentNodeContext:
+    """Tests for CurrentNodeContext context manager with expected_outputs."""
+
+    def test_context_manager_with_expected_outputs(self):
+        """Test CurrentNodeContext sets and resets context correctly."""
+        assert get_executing_context() is None
+
+        with CurrentNodeContext("prompt-1", "node-1", 0, frozenset({0, 1})):
+            ctx = get_executing_context()
+            assert ctx is not None
+            assert ctx.prompt_id == "prompt-1"
+            assert ctx.node_id == "node-1"
+            assert ctx.list_index == 0
+            assert ctx.expected_outputs == frozenset({0, 1})
+
+        assert get_executing_context() is None
+
+    def test_context_manager_without_expected_outputs(self):
+        """Test CurrentNodeContext works without expected_outputs (backwards compatible)."""
+        with CurrentNodeContext("prompt-1", "node-1"):
+            ctx = get_executing_context()
+            assert ctx is not None
+            assert ctx.expected_outputs is None
+
+    def test_nested_context_managers(self):
+        """Test nested CurrentNodeContext managers."""
+        with CurrentNodeContext("prompt-1", "node-1", 0, frozenset({0})):
+            ctx1 = get_executing_context()
+            assert ctx1.expected_outputs == frozenset({0})
+
+            with CurrentNodeContext("prompt-1", "node-2", 0, frozenset({1, 2})):
+                ctx2 = get_executing_context()
+                assert ctx2.expected_outputs == frozenset({1, 2})
+                assert ctx2.node_id == "node-2"
+
+            # After inner context exits, should be back to outer context
+            ctx1_again = get_executing_context()
+            assert ctx1_again.expected_outputs == frozenset({0})
+            assert ctx1_again.node_id == "node-1"
+
+    def test_output_check_pattern(self):
+        """Test the typical pattern nodes will use to check expected outputs."""
+        with CurrentNodeContext("prompt-1", "node-1", 0, frozenset({0, 2})):
+            ctx = get_executing_context()
+
+            # Typical usage pattern
+            if ctx and ctx.expected_outputs is not None:
+                should_compute_0 = 0 in ctx.expected_outputs
+                should_compute_1 = 1 in ctx.expected_outputs
+                should_compute_2 = 2 in ctx.expected_outputs
+            else:
+                # Fallback when info not available
+                should_compute_0 = should_compute_1 = should_compute_2 = True
+
+            assert should_compute_0 is True
+            assert should_compute_1 is False  # Not in expected_outputs
+            assert should_compute_2 is True
+
+
+class TestSchemaLazyOutputs:
+    """Tests for lazy_outputs in V3 Schema."""
+
+    def test_schema_lazy_outputs_default(self):
+        """Test that lazy_outputs defaults to False."""
+        schema = IO.Schema(
+            node_id="TestNode",
+            inputs=[],
+            outputs=[IO.Float.Output()],
+        )
+        assert schema.lazy_outputs is False
+
+    def test_schema_lazy_outputs_true(self):
+        """Test setting lazy_outputs to True."""
+        schema = IO.Schema(
+            node_id="TestNode",
+            lazy_outputs=True,
+            inputs=[],
+            outputs=[IO.Float.Output()],
+        )
+        assert schema.lazy_outputs is True
+
+    def test_v3_node_lazy_outputs_property(self):
+        """Test that LAZY_OUTPUTS property works on V3 nodes."""
+
+        class TestNodeWithLazyOutputs(IO.ComfyNode):
+            @classmethod
+            def define_schema(cls):
+                return IO.Schema(
+                    node_id="TestNodeWithLazyOutputs",
+                    lazy_outputs=True,
+                    inputs=[],
+                    outputs=[IO.Float.Output()],
+                )
+
+            @classmethod
+            def execute(cls):
+                return IO.NodeOutput(1.0)
+
+        assert TestNodeWithLazyOutputs.LAZY_OUTPUTS is True
+
+    def test_v3_node_lazy_outputs_default(self):
+        """Test that LAZY_OUTPUTS defaults to False on V3 nodes."""
+
+        class TestNodeWithoutLazyOutputs(IO.ComfyNode):
+            @classmethod
+            def define_schema(cls):
+                return IO.Schema(
+                    node_id="TestNodeWithoutLazyOutputs",
+                    inputs=[],
+                    outputs=[IO.Float.Output()],
+                )
+
+            @classmethod
+            def execute(cls):
+                return IO.NodeOutput(1.0)
+
+        assert TestNodeWithoutLazyOutputs.LAZY_OUTPUTS is False
+
+
+class TestIsOutputNeeded:
+    """Tests for is_output_needed() helper function."""
+
+    def test_output_needed_when_in_expected(self):
+        """Test that output is needed when in expected_outputs."""
+        with CurrentNodeContext("prompt-1", "node-1", 0, frozenset({0, 2})):
+            assert is_output_needed(0) is True
+            assert is_output_needed(2) is True
+
+    def test_output_not_needed_when_not_in_expected(self):
+        """Test that output is not needed when not in expected_outputs."""
+        with CurrentNodeContext("prompt-1", "node-1", 0, frozenset({0, 2})):
+            assert is_output_needed(1) is False
+            assert is_output_needed(3) is False
+
+    def test_output_needed_when_no_context(self):
+        """Test that output is needed when no context."""
+        assert get_executing_context() is None
+        assert is_output_needed(0) is True
+        assert is_output_needed(1) is True
+
+    def test_output_needed_when_expected_outputs_is_none(self):
+        """Test that output is needed when expected_outputs is None."""
+        with CurrentNodeContext("prompt-1", "node-1", 0, None):
+            assert is_output_needed(0) is True
+            assert is_output_needed(1) is True

tests/execution/test_execution.py

@@ -573,6 +573,144 @@ class TestExecution:
         else:
             assert result.did_run(test_node), "The execution should have been re-run"
 
+    def test_expected_outputs_all_connected(self, client: ComfyClient, builder: GraphBuilder):
+        """Test that expected_outputs contains all connected outputs."""
+        g = builder
+        # Create a node with 3 outputs, all connected
+        expected_outputs_node = g.node("TestExpectedOutputs", height=64, width=64)
+
+        # Connect all 3 outputs to preview nodes
+        output0 = g.node("PreviewImage", images=expected_outputs_node.out(0))
+        output1 = g.node("PreviewImage", images=expected_outputs_node.out(1))
+        output2 = g.node("PreviewImage", images=expected_outputs_node.out(2))
+
+        result = client.run(g)
+
+        # All outputs should be white (255) since all are connected
+        images0 = result.get_images(output0)
+        images1 = result.get_images(output1)
+        images2 = result.get_images(output2)
+
+        assert len(images0) == 1, "Should have 1 image for output0"
+        assert len(images1) == 1, "Should have 1 image for output1"
+        assert len(images2) == 1, "Should have 1 image for output2"
+
+        # White pixels = 255, meaning output was in expected_outputs
+        assert numpy.array(images0[0]).min() == 255, "Output 0 should be white (was expected)"
+        assert numpy.array(images1[0]).min() == 255, "Output 1 should be white (was expected)"
+        assert numpy.array(images2[0]).min() == 255, "Output 2 should be white (was expected)"
+
+    def test_expected_outputs_partial_connected(self, client: ComfyClient, builder: GraphBuilder):
+        """Test that expected_outputs only contains connected outputs."""
+        g = builder
+        # Create a node with 3 outputs, only some connected
+        expected_outputs_node = g.node("TestExpectedOutputs", height=64, width=64)
+
+        # Only connect outputs 0 and 2, leave output 1 disconnected
+        output0 = g.node("PreviewImage", images=expected_outputs_node.out(0))
+        # output1 is intentionally not connected
+        output2 = g.node("PreviewImage", images=expected_outputs_node.out(2))
+
+        result = client.run(g)
+
+        # Connected outputs should be white (255)
+        images0 = result.get_images(output0)
+        images2 = result.get_images(output2)
+
+        assert len(images0) == 1, "Should have 1 image for output0"
+        assert len(images2) == 1, "Should have 1 image for output2"
+
+        # White = expected, output 1 is not connected so we can't verify it directly but outputs 0 and 2 should be white
+        assert numpy.array(images0[0]).min() == 255, "Output 0 should be white (was expected)"
+        assert numpy.array(images2[0]).min() == 255, "Output 2 should be white (was expected)"
+
+    def test_expected_outputs_single_connected(self, client: ComfyClient, builder: GraphBuilder):
+        """Test that expected_outputs works with single connected output."""
+        g = builder
+        # Create a node with 3 outputs, only one connected
+        expected_outputs_node = g.node("TestExpectedOutputs", height=64, width=64)
+
+        # Only connect output 1
+        output1 = g.node("PreviewImage", images=expected_outputs_node.out(1))
+
+        result = client.run(g)
+
+        images1 = result.get_images(output1)
+        assert len(images1) == 1, "Should have 1 image for output1"
+
+        # Output 1 should be white (connected), others are not visible in this test
+        assert numpy.array(images1[0]).min() == 255, "Output 1 should be white (was expected)"
+
+    def test_expected_outputs_cache_invalidation(self, client: ComfyClient, builder: GraphBuilder, server):
+        """Test that cache invalidates when output connections change."""
+        g = builder
+        # Use unique dimensions to avoid cache collision with other expected_outputs tests
+        expected_outputs_node = g.node("TestExpectedOutputs", height=32, width=32)
+
+        # First run: only connect output 0
+        output0 = g.node("PreviewImage", images=expected_outputs_node.out(0))
+
+        result1 = client.run(g)
+        assert result1.did_run(expected_outputs_node), "First run should execute the node"
+
+        # Second run: same connections, should be cached
+        result2 = client.run(g)
+        if server["should_cache_results"]:
+            assert not result2.did_run(expected_outputs_node), "Second run should be cached"
+
+        # Third run: add connection to output 2
+        output2 = g.node("PreviewImage", images=expected_outputs_node.out(2))
+
+        result3 = client.run(g)
+        # Because LAZY_OUTPUTS=True, changing connections should invalidate cache
+        if server["should_cache_results"]:
+            assert result3.did_run(expected_outputs_node), "Adding output connection should invalidate cache"
+
+        # Verify both outputs are now white
+        images0 = result3.get_images(output0)
+        images2 = result3.get_images(output2)
+        assert numpy.array(images0[0]).min() == 255, "Output 0 should be white"
+        assert numpy.array(images2[0]).min() == 255, "Output 2 should be white"
+
+    def test_expected_outputs_expansion_output_mapping(self, client: ComfyClient, builder: GraphBuilder):
+        """A socket consumed only via an expansion's parent-output mapping must still
+        be in the inner LAZY_OUTPUTS node's expected_outputs (white, not black)."""
+        g = builder
+        expander = g.node("TestExpectedOutputsExpansion", height=80, width=80)
+        output = g.node("PreviewImage", images=expander.out(0))
+
+        result = client.run(g)
+
+        images = result.get_images(output)
+        assert len(images) == 1, "Should have 1 image"
+        assert numpy.array(images[0]).min() == 255, (
+            "Inner node skipped an output that is consumed via the expansion's "
+            "parent-output mapping (expected white, got black)"
+        )
+
+    def test_expected_outputs_requires_opt_in(self, client: ComfyClient, builder: GraphBuilder, server):
+        """Nodes without LAZY_OUTPUTS must see expected_outputs=None: their cache key
+        ignores topology, so a skipped output would be served stale after rewiring."""
+        g = builder
+        node = g.node("TestExpectedOutputsNotOptedIn", height=96, width=96)
+        output0 = g.node("PreviewImage", images=node.out(0))
+
+        # Only output 0 connected: correct gating -> node sees None, computes all
+        result1 = client.run(g)
+        assert numpy.array(result1.get_images(output0)[0]).min() == 255
+
+        # Connect output 1: key unchanged -> cache hit must still serve correct data
+        output1 = g.node("PreviewImage", images=node.out(1))
+        result2 = client.run(g)
+
+        if server["should_cache_results"]:
+            assert not result2.did_run(node), "Node should be a cache hit (key ignores topology)"
+        images1 = result2.get_images(output1)
+        assert len(images1) == 1, "Should have 1 image for output1"
+        assert numpy.array(images1[0]).min() == 255, (
+            "Non-opted-in node observed expected_outputs and skipped output 1; "
+            "the stale skipped value was then served from cache"
+        )
 
     def test_parallel_sleep_nodes(self, client: ComfyClient, builder: GraphBuilder, skip_timing_checks):
         # Warmup execution to ensure server is fully initialized

tests/execution/testing_nodes/testing-pack/specific_tests.py

@@ -6,6 +6,7 @@ from .tools import VariantSupport
 from comfy_execution.graph_utils import GraphBuilder
 from comfy.comfy_types.node_typing import ComfyNodeABC
 from comfy.comfy_types import IO
+from comfy_execution.utils import get_executing_context, is_output_needed
 
 class TestLazyMixImages:
     @classmethod
@@ -482,6 +483,106 @@ class TestOutputNodeWithSocketOutput:
         result = image * value
         return (result,)
 
+
+class TestExpectedOutputs:
+    """Test node for the expected_outputs feature.
+
+    This node has 3 IMAGE outputs that encode which outputs were expected:
+    - White image (255) if the output was in expected_outputs
+    - Black image (0) if the output was NOT in expected_outputs
+
+    This allows integration tests to verify which outputs were expected by checking pixel values.
+    """
+    LAZY_OUTPUTS = True  # Opt into cache invalidation on output connection changes
+
+    @classmethod
+    def INPUT_TYPES(cls):
+        return {
+            "required": {
+                "height": ("INT", {"default": 64, "min": 1, "max": 1024}),
+                "width": ("INT", {"default": 64, "min": 1, "max": 1024}),
+            },
+        }
+
+    RETURN_TYPES = ("IMAGE", "IMAGE", "IMAGE")
+    RETURN_NAMES = ("output0", "output1", "output2")
+    FUNCTION = "execute"
+    CATEGORY = "_for_testing"
+
+    def execute(self, height, width):
+        # Return white image if expected, black if not
+        # This allows tests to verify which outputs were expected via pixel values
+        white = torch.ones(1, height, width, 3)
+        black = torch.zeros(1, height, width, 3)
+
+        return (
+            white if is_output_needed(0) else black,
+            white if is_output_needed(1) else black,
+            white if is_output_needed(2) else black,
+        )
+
+
+class TestExpectedOutputsExpansion:
+    """Expands into an inner LAZY_OUTPUTS node whose output 1 is consumed ONLY via
+    the parent-output mapping (no input link anywhere). If that mapping is not part
+    of the expected-outputs map, the inner node wrongly skips it -> black not white.
+    """
+
+    @classmethod
+    def INPUT_TYPES(cls):
+        return {
+            "required": {
+                "height": ("INT", {"default": 64, "min": 1, "max": 1024}),
+                "width": ("INT", {"default": 64, "min": 1, "max": 1024}),
+            },
+        }
+
+    RETURN_TYPES = ("IMAGE",)
+    FUNCTION = "execute"
+    CATEGORY = "_for_testing"
+
+    def execute(self, height, width):
+        g = GraphBuilder()
+        inner = g.node("TestExpectedOutputs", height=height, width=width)
+        return {"result": (inner.out(1),), "expand": g.finalize()}
+
+
+class TestExpectedOutputsNotOptedIn:
+    """Reads expected_outputs WITHOUT declaring LAZY_OUTPUTS; the executor must pass
+    None (such nodes have no cache-key protection against output rewiring). Outputs
+    are white when the node correctly sees None, otherwise they encode membership.
+    """
+
+    @classmethod
+    def INPUT_TYPES(cls):
+        return {
+            "required": {
+                "height": ("INT", {"default": 64, "min": 1, "max": 1024}),
+                "width": ("INT", {"default": 64, "min": 1, "max": 1024}),
+            },
+        }
+
+    RETURN_TYPES = ("IMAGE", "IMAGE")
+    RETURN_NAMES = ("output0", "output1")
+    FUNCTION = "execute"
+    CATEGORY = "_for_testing"
+
+    def execute(self, height, width):
+        # Raw context access (not is_output_needed): must distinguish None from a set
+        ctx = get_executing_context()
+        expected = ctx.expected_outputs if ctx is not None else None
+
+        white = torch.ones(1, height, width, 3)
+        black = torch.zeros(1, height, width, 3)
+
+        if expected is None:
+            return (white, white.clone())
+        return (
+            white if 0 in expected else black,
+            white if 1 in expected else black,
+        )
+
+
 TEST_NODE_CLASS_MAPPINGS = {
     "TestLazyMixImages": TestLazyMixImages,
     "TestVariadicAverage": TestVariadicAverage,
@@ -498,6 +599,9 @@ TEST_NODE_CLASS_MAPPINGS = {
     "TestSleep": TestSleep,
     "TestParallelSleep": TestParallelSleep,
     "TestOutputNodeWithSocketOutput": TestOutputNodeWithSocketOutput,
+    "TestExpectedOutputs": TestExpectedOutputs,
+    "TestExpectedOutputsExpansion": TestExpectedOutputsExpansion,
+    "TestExpectedOutputsNotOptedIn": TestExpectedOutputsNotOptedIn,
 }
 
 TEST_NODE_DISPLAY_NAME_MAPPINGS = {
@@ -516,4 +620,7 @@ TEST_NODE_DISPLAY_NAME_MAPPINGS = {
     "TestSleep": "Test Sleep",
     "TestParallelSleep": "Test Parallel Sleep",
     "TestOutputNodeWithSocketOutput": "Test Output Node With Socket Output",
+    "TestExpectedOutputs": "Test Expected Outputs",
+    "TestExpectedOutputsExpansion": "Test Expected Outputs Expansion",
+    "TestExpectedOutputsNotOptedIn": "Test Expected Outputs Not Opted In",
 }