Merge a75bf6f1d8 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/model_management.py

@@ -190,6 +190,14 @@ def is_wsl():
         return True
     return False
 
+_WSL_SOFT_EMPTY_CACHE_SKIP_LOGGED = False
+
+def wsl_skip_nonforced_soft_empty_cache():
+    return is_wsl() and os.getenv("COMFYUI_WSL_SOFT_EMPTY_CACHE", "0") != "1"
+
+def wsl_skip_model_load_synchronize():
+    return is_wsl() and os.getenv("COMFYUI_WSL_MODEL_LOAD_SYNCHRONIZE", "0") != "1"
+
 def get_torch_device():
     global directml_enabled
     global cpu_state
@@ -941,7 +949,14 @@ def load_models_gpu(models, memory_required=0, force_patch_weights=False, minimu
         if vram_set_state == VRAMState.NO_VRAM:
             lowvram_model_memory = 0.1
 
+        model_name = model.model.__class__.__name__ if hasattr(model, "model") else model.__class__.__name__
+        logging.info(
+            f"Loading model {model_name} start: device={torch_dev} "
+            f"vram_state={vram_set_state.name} lowvram_model_memory={lowvram_model_memory} "
+            f"force_full_load={force_full_load} force_patch_weights={force_patch_weights}"
+        )
         loaded_model.model_load(lowvram_model_memory, force_patch_weights=force_patch_weights)
+        logging.info(f"Loading model {model_name} complete")
         current_loaded_models.insert(0, loaded_model)
     return
 
@@ -1961,6 +1976,12 @@ def soft_empty_cache(force=False):
     elif is_mlu():
         torch.mlu.empty_cache()
     elif torch.cuda.is_available():
+        if wsl_skip_nonforced_soft_empty_cache() and not force:
+            global _WSL_SOFT_EMPTY_CACHE_SKIP_LOGGED
+            if not _WSL_SOFT_EMPTY_CACHE_SKIP_LOGGED:
+                logging.info("Skipping non-forced CUDA soft_empty_cache on WSL; set COMFYUI_WSL_SOFT_EMPTY_CACHE=1 to re-enable.")
+                _WSL_SOFT_EMPTY_CACHE_SKIP_LOGGED = True
+            return
         torch.cuda.synchronize()
         torch.cuda.empty_cache()
         torch.cuda.ipc_collect()

comfy/model_patcher.py

@@ -42,6 +42,8 @@ from comfy.patcher_extension import CallbacksMP, PatcherInjection, WrappersMP
 
 import comfy_aimdo.model_vbar
 
+_WSL_MODEL_LOAD_SYNC_SKIP_LOGGED = False
+
 def set_model_options_patch_replace(model_options, patch, name, block_name, number, transformer_index=None):
     to = model_options["transformer_options"].copy()
 
@@ -1006,6 +1008,11 @@ class ModelPatcher:
                 mem_counter += move_weight_functions(m, device_to)
 
             load_completely.sort(reverse=True)
+            skip_wsl_load_sync = comfy.model_management.is_device_cuda(device_to) and comfy.model_management.wsl_skip_model_load_synchronize()
+            global _WSL_MODEL_LOAD_SYNC_SKIP_LOGGED
+            if skip_wsl_load_sync and len(load_completely) > 0 and not _WSL_MODEL_LOAD_SYNC_SKIP_LOGGED:
+                logging.info("Skipping per-module CUDA synchronize during model load on WSL; set COMFYUI_WSL_MODEL_LOAD_SYNCHRONIZE=1 to re-enable.")
+                _WSL_MODEL_LOAD_SYNC_SKIP_LOGGED = True
             for x in load_completely:
                 n = x[1]
                 m = x[2]
@@ -1020,7 +1027,7 @@ class ModelPatcher:
                     key = key_param_name_to_key(n, param)
                     self.unpin_weight(key)
                     self.patch_weight_to_device(key, device_to=device_to)
-                if comfy.model_management.is_device_cuda(device_to):
+                if comfy.model_management.is_device_cuda(device_to) and not skip_wsl_load_sync:
                     torch.cuda.synchronize()
 
                 logging.debug("lowvram: loaded module regularly {} {}".format(n, m))

comfy/sd.py

@@ -1198,7 +1198,17 @@ class VAE:
             else:
                 pixel_samples = pixel_samples.unsqueeze(2)
 
-        memory_used = self.memory_used_encode(pixel_samples.shape, self.vae_dtype)  # TODO: calculate mem required for tile
+        if dims == 2:
+            default_tile_x = 512 if tile_x is None else tile_x
+            default_tile_y = 512 if tile_y is None else tile_y
+            tile_shapes = [
+                (1, pixel_samples.shape[1], min(pixel_samples.shape[2], max(1, default_tile_y)), min(pixel_samples.shape[3], max(1, default_tile_x))),
+                (1, pixel_samples.shape[1], min(pixel_samples.shape[2], max(1, default_tile_y // 2)), min(pixel_samples.shape[3], max(1, default_tile_x * 2))),
+                (1, pixel_samples.shape[1], min(pixel_samples.shape[2], max(1, default_tile_y * 2)), min(pixel_samples.shape[3], max(1, default_tile_x // 2))),
+            ]
+            memory_used = max(self.memory_used_encode(shape, self.vae_dtype) for shape in tile_shapes)
+        else:
+            memory_used = self.memory_used_encode(pixel_samples.shape, self.vae_dtype)
         model_management.load_models_gpu([self.patcher], memory_required=memory_used, force_full_load=self.disable_offload)
 
         args = {}

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_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

@@ -1,5 +1,6 @@
 import asyncio
 import bisect
+import gc
 import itertools
 import psutil
 import time
@@ -17,6 +18,7 @@ NODE_CLASS_CONTAINS_UNIQUE_ID: Dict[str, bool] = {}
 
 
 def include_unique_id_in_input(class_type: str) -> bool:
+    """Return whether a node class includes UNIQUE_ID among its hidden inputs."""
     if class_type in NODE_CLASS_CONTAINS_UNIQUE_ID:
         return NODE_CLASS_CONTAINS_UNIQUE_ID[class_type]
     class_def = nodes.NODE_CLASS_MAPPINGS[class_type]
@@ -24,52 +26,412 @@ def include_unique_id_in_input(class_type: str) -> bool:
     return NODE_CLASS_CONTAINS_UNIQUE_ID[class_type]
 
 class CacheKeySet(ABC):
+    """Base helper for building and storing cache keys for prompt nodes."""
     def __init__(self, dynprompt, node_ids, is_changed_cache):
+        """Initialize cache-key storage for a dynamic prompt execution pass."""
         self.keys = {}
         self.subcache_keys = {}
 
     @abstractmethod
     async def add_keys(self, node_ids):
+        """Populate cache keys for the provided node ids."""
         raise NotImplementedError()
 
     def all_node_ids(self):
+        """Return the set of node ids currently tracked by this key set."""
         return set(self.keys.keys())
 
     def get_used_keys(self):
+        """Return the computed cache keys currently in use."""
         return self.keys.values()
 
     def get_used_subcache_keys(self):
+        """Return the computed subcache keys currently in use."""
         return self.subcache_keys.values()
 
     def get_data_key(self, node_id):
+        """Return the cache key for a node, if present."""
         return self.keys.get(node_id, None)
 
     def get_subcache_key(self, node_id):
+        """Return the subcache key for a node, if present."""
         return self.subcache_keys.get(node_id, None)
 
 class Unhashable:
-    def __init__(self):
-        self.value = float("NaN")
+    """Hashable identity sentinel for values that cannot be represented safely in cache keys."""
+    pass
 
-def to_hashable(obj):
-    # So that we don't infinitely recurse since frozenset and tuples
-    # are Sequences.
-    if isinstance(obj, (int, float, str, bool, bytes, type(None))):
-        return obj
-    elif isinstance(obj, Mapping):
-        return frozenset([(to_hashable(k), to_hashable(v)) for k, v in sorted(obj.items())])
-    elif isinstance(obj, Sequence):
-        return frozenset(zip(itertools.count(), [to_hashable(i) for i in obj]))
-    else:
-        # TODO - Support other objects like tensors?
+
+_PRIMITIVE_SIGNATURE_TYPES = (int, float, str, bool, bytes, type(None))
+_CONTAINER_SIGNATURE_TYPES = (dict, list, tuple, set, frozenset)
+_MAX_SIGNATURE_DEPTH = 32
+_MAX_SIGNATURE_CONTAINER_VISITS = 10_000
+_FAILED_SIGNATURE = object()
+
+
+def _shallow_is_changed_signature(value):
+    """Reduce execution-time `is_changed` values through a fail-closed builtin canonicalizer."""
+    value_type = type(value)
+    if value_type in _PRIMITIVE_SIGNATURE_TYPES:
+        return value
+
+    if value_type not in _CONTAINER_SIGNATURE_TYPES:
         return Unhashable()
 
+    canonical = _signature_to_hashable(value, max_nodes=64)
+    if type(canonical) is Unhashable:
+        return canonical
+    if value_type is list or value_type is tuple:
+        container_tag = "is_changed_list" if value_type is list else "is_changed_tuple"
+        return (container_tag, canonical[1])
+
+    return canonical
+
+
+def _primitive_signature_sort_key(obj):
+    """Return a deterministic ordering key for primitive signature values."""
+    obj_type = type(obj)
+    return ("primitive", obj_type.__module__, obj_type.__qualname__, repr(obj))
+
+
+def _sanitized_sort_key(obj, depth=0, max_depth=_MAX_SIGNATURE_DEPTH, active=None, memo=None):
+    """Return a deterministic ordering key for sanitized built-in container content."""
+    if depth >= max_depth:
+        return ("MAX_DEPTH",)
+
+    if active is None:
+        active = set()
+    if memo is None:
+        memo = {}
+
+    obj_type = type(obj)
+    if obj_type is Unhashable:
+        return ("UNHASHABLE",)
+    elif obj_type in _PRIMITIVE_SIGNATURE_TYPES:
+        return (obj_type.__module__, obj_type.__qualname__, repr(obj))
+    elif obj_type not in _CONTAINER_SIGNATURE_TYPES:
+        return (obj_type.__module__, obj_type.__qualname__, "OPAQUE")
+
+    obj_id = id(obj)
+    if obj_id in memo:
+        return memo[obj_id]
+    if obj_id in active:
+        return ("CYCLE",)
+
+    active.add(obj_id)
+    try:
+        if obj_type is dict:
+            items = [
+                (
+                    _sanitized_sort_key(k, depth + 1, max_depth, active, memo),
+                    _sanitized_sort_key(v, depth + 1, max_depth, active, memo),
+                )
+                for k, v in obj.items()
+            ]
+            items.sort()
+            result = ("dict", tuple(items))
+        elif obj_type is list:
+            result = ("list", tuple(_sanitized_sort_key(i, depth + 1, max_depth, active, memo) for i in obj))
+        elif obj_type is tuple:
+            result = ("tuple", tuple(_sanitized_sort_key(i, depth + 1, max_depth, active, memo) for i in obj))
+        elif obj_type is set:
+            result = ("set", tuple(sorted(_sanitized_sort_key(i, depth + 1, max_depth, active, memo) for i in obj)))
+        else:
+            result = ("frozenset", tuple(sorted(_sanitized_sort_key(i, depth + 1, max_depth, active, memo) for i in obj)))
+    finally:
+        active.discard(obj_id)
+
+    memo[obj_id] = result
+    return result
+
+
+def _signature_to_hashable_impl(obj, depth=0, max_depth=_MAX_SIGNATURE_DEPTH, active=None, memo=None, budget=None):
+    """Canonicalize signature inputs directly into their final hashable form."""
+    if depth >= max_depth:
+        return _FAILED_SIGNATURE
+
+    if active is None:
+        active = set()
+    if memo is None:
+        memo = {}
+    if budget is None:
+        budget = {"remaining": _MAX_SIGNATURE_CONTAINER_VISITS}
+
+    obj_type = type(obj)
+    if obj_type in _PRIMITIVE_SIGNATURE_TYPES:
+        return obj, _primitive_signature_sort_key(obj)
+    if obj_type is Unhashable or obj_type not in _CONTAINER_SIGNATURE_TYPES:
+        return _FAILED_SIGNATURE
+
+    obj_id = id(obj)
+    if obj_id in memo:
+        return memo[obj_id]
+    if obj_id in active:
+        return _FAILED_SIGNATURE
+
+    budget["remaining"] -= 1
+    if budget["remaining"] < 0:
+        return _FAILED_SIGNATURE
+
+    active.add(obj_id)
+    try:
+        if obj_type is dict:
+            try:
+                items = list(obj.items())
+            except RuntimeError:
+                return _FAILED_SIGNATURE
+
+            ordered_items = []
+            for key, value in items:
+                if type(key) not in _PRIMITIVE_SIGNATURE_TYPES:
+                    return _FAILED_SIGNATURE
+                key_result = (key, _primitive_signature_sort_key(key))
+                value_result = _signature_to_hashable_impl(value, depth + 1, max_depth, active, memo, budget)
+                if value_result is _FAILED_SIGNATURE:
+                    return _FAILED_SIGNATURE
+                key_value, key_sort = key_result
+                value_value, value_sort = value_result
+                ordered_items.append((key_sort, value_sort, key_value, value_value))
+
+            ordered_items.sort(key=lambda item: (item[0], item[1]))
+            for index in range(1, len(ordered_items)):
+                previous_key_sort = ordered_items[index - 1][0]
+                current_key_sort = ordered_items[index][0]
+                if previous_key_sort == current_key_sort:
+                    return _FAILED_SIGNATURE
+
+            value = ("dict", tuple((key_value, value_value) for _, _, key_value, value_value in ordered_items))
+            sort_key = ("dict", tuple((key_sort, value_sort) for key_sort, value_sort, _, _ in ordered_items))
+        elif obj_type is list or obj_type is tuple:
+            try:
+                items = list(obj)
+            except RuntimeError:
+                return _FAILED_SIGNATURE
+
+            child_results = []
+            for item in items:
+                child_result = _signature_to_hashable_impl(item, depth + 1, max_depth, active, memo, budget)
+                if child_result is _FAILED_SIGNATURE:
+                    return _FAILED_SIGNATURE
+                child_results.append(child_result)
+
+            container_tag = "list" if obj_type is list else "tuple"
+            value = (container_tag, tuple(child for child, _ in child_results))
+            sort_key = (container_tag, tuple(child_sort for _, child_sort in child_results))
+        else:
+            try:
+                items = list(obj)
+            except RuntimeError:
+                return _FAILED_SIGNATURE
+
+            ordered_items = []
+            for item in items:
+                child_result = _signature_to_hashable_impl(item, depth + 1, max_depth, active, memo, budget)
+                if child_result is _FAILED_SIGNATURE:
+                    return _FAILED_SIGNATURE
+                child_value, child_sort = child_result
+                ordered_items.append((child_sort, child_value))
+
+            ordered_items.sort(key=lambda item: item[0])
+            for index in range(1, len(ordered_items)):
+                previous_sort_key, previous_value = ordered_items[index - 1]
+                current_sort_key, current_value = ordered_items[index]
+                if previous_sort_key == current_sort_key and previous_value != current_value:
+                    return _FAILED_SIGNATURE
+
+            container_tag = "set" if obj_type is set else "frozenset"
+            value = (container_tag, tuple(child_value for _, child_value in ordered_items))
+            sort_key = (container_tag, tuple(child_sort for child_sort, _ in ordered_items))
+    finally:
+        active.discard(obj_id)
+
+    memo[obj_id] = (value, sort_key)
+    return memo[obj_id]
+
+
+def _signature_to_hashable(obj, max_nodes=_MAX_SIGNATURE_CONTAINER_VISITS):
+    """Build the final cache-signature representation in one fail-closed pass."""
+    try:
+        result = _signature_to_hashable_impl(obj, budget={"remaining": max_nodes})
+    except RuntimeError:
+        return Unhashable()
+    if result is _FAILED_SIGNATURE:
+        return Unhashable()
+    return result[0]
+
+
+def to_hashable(obj, max_nodes=_MAX_SIGNATURE_CONTAINER_VISITS):
+    """Convert sanitized prompt inputs into a stable hashable representation.
+
+    The input is expected to already be sanitized to plain built-in containers,
+    but this function still fails safe for anything unexpected. Traversal is
+    iterative and memoized so shared built-in substructures do not trigger
+    exponential re-walks during cache-key construction.
+    """
+    obj_type = type(obj)
+    if obj_type in _PRIMITIVE_SIGNATURE_TYPES or obj_type is Unhashable:
+        return obj
+    if obj_type not in _CONTAINER_SIGNATURE_TYPES:
+        return Unhashable()
+
+    memo = {}
+    active = set()
+    snapshots = {}
+    sort_memo = {}
+    processed = 0
+    # Keep traversal state separate from container snapshots/results.
+    work_stack = [(obj, False)]
+
+    def resolve_value(value):
+        """Resolve a child value from the completed memo table when available."""
+        value_type = type(value)
+        if value_type in _PRIMITIVE_SIGNATURE_TYPES or value_type is Unhashable:
+            return value
+        return memo.get(id(value), Unhashable())
+
+    def is_failed(value):
+        """Return whether a resolved child value represents failed canonicalization."""
+        return type(value) is Unhashable
+
+    def resolve_unordered_values(current_items, container_tag):
+        """Resolve a set-like container or fail closed if ordering is ambiguous."""
+        try:
+            ordered_items = [
+                (_sanitized_sort_key(item, memo=sort_memo), resolve_value(item))
+                for item in current_items
+            ]
+            if any(is_failed(value) for _, value in ordered_items):
+                return Unhashable()
+            ordered_items.sort(key=lambda item: item[0])
+        except RuntimeError:
+            return Unhashable()
+
+        for index in range(1, len(ordered_items)):
+            previous_key, previous_value = ordered_items[index - 1]
+            current_key, current_value = ordered_items[index]
+            if previous_key == current_key and previous_value != current_value:
+                return Unhashable()
+
+        return (container_tag, tuple(value for _, value in ordered_items))
+
+    while work_stack:
+        entry = work_stack.pop()
+        if len(entry) == 3:
+            _, current_id, current_type = entry
+            current = None
+            expanded = True
+        else:
+            current, expanded = entry
+            current_type = type(current)
+            current_id = id(current)
+
+        if not expanded and (current_type in _PRIMITIVE_SIGNATURE_TYPES or current_type is Unhashable):
+            continue
+        if not expanded and current_type not in _CONTAINER_SIGNATURE_TYPES:
+            memo[current_id] = Unhashable()
+            continue
+
+        if current_id in memo:
+            continue
+
+        if expanded:
+            active.discard(current_id)
+            try:
+                items = snapshots.pop(current_id, None)
+                if items is None:
+                    memo[current_id] = Unhashable()
+                    continue
+
+                if current_type is dict:
+                    ordered_items = [
+                        (_sanitized_sort_key(k, memo=sort_memo), k, resolve_value(v))
+                        for k, v in items
+                    ]
+                    if any(type(key) not in _PRIMITIVE_SIGNATURE_TYPES or is_failed(value) for _, key, value in ordered_items):
+                        memo[current_id] = Unhashable()
+                        continue
+                    ordered_items.sort(key=lambda item: item[0])
+                    for index in range(1, len(ordered_items)):
+                        if ordered_items[index - 1][0] == ordered_items[index][0]:
+                            memo[current_id] = Unhashable()
+                            break
+                    else:
+                        memo[current_id] = (
+                            "dict",
+                            tuple((key, value) for _, key, value in ordered_items),
+                        )
+                elif current_type is list:
+                    resolved_items = tuple(resolve_value(item) for item in items)
+                    if any(is_failed(item) for item in resolved_items):
+                        memo[current_id] = Unhashable()
+                    else:
+                        memo[current_id] = ("list", resolved_items)
+                elif current_type is tuple:
+                    resolved_items = tuple(resolve_value(item) for item in items)
+                    if any(is_failed(item) for item in resolved_items):
+                        memo[current_id] = Unhashable()
+                    else:
+                        memo[current_id] = ("tuple", resolved_items)
+                elif current_type is set:
+                    memo[current_id] = resolve_unordered_values(items, "set")
+                else:
+                    memo[current_id] = resolve_unordered_values(items, "frozenset")
+            except RuntimeError:
+                memo[current_id] = Unhashable()
+            continue
+
+        if current_id in active:
+            memo[current_id] = Unhashable()
+            continue
+
+        processed += 1
+        if processed > max_nodes:
+            return Unhashable()
+
+        active.add(current_id)
+        if current_type is dict:
+            try:
+                items = list(current.items())
+                snapshots[current_id] = items
+            except RuntimeError:
+                memo[current_id] = Unhashable()
+                active.discard(current_id)
+                continue
+            for key, value in items:
+                if type(key) not in _PRIMITIVE_SIGNATURE_TYPES:
+                    snapshots.pop(current_id, None)
+                    memo[current_id] = Unhashable()
+                    active.discard(current_id)
+                    break
+            else:
+                work_stack.append(("EXPANDED", current_id, current_type))
+                for _, value in reversed(items):
+                    work_stack.append((value, False))
+                continue
+            continue
+        else:
+            try:
+                items = list(current)
+                snapshots[current_id] = items
+            except RuntimeError:
+                memo[current_id] = Unhashable()
+                active.discard(current_id)
+                continue
+            work_stack.append(("EXPANDED", current_id, current_type))
+            for item in reversed(items):
+                work_stack.append((item, False))
+
+    return memo.get(id(obj), Unhashable())
+
 class CacheKeySetID(CacheKeySet):
+    """Cache-key strategy that keys nodes by node id and class type."""
     def __init__(self, dynprompt, node_ids, is_changed_cache):
+        """Initialize identity-based cache keys for the supplied dynamic prompt."""
         super().__init__(dynprompt, node_ids, is_changed_cache)
         self.dynprompt = dynprompt
 
     async def add_keys(self, node_ids):
+        """Populate identity-based keys for nodes that exist in the dynamic prompt."""
         for node_id in node_ids:
             if node_id in self.keys:
                 continue
@@ -80,15 +442,19 @@ class CacheKeySetID(CacheKeySet):
             self.subcache_keys[node_id] = (node_id, node["class_type"])
 
 class CacheKeySetInputSignature(CacheKeySet):
+    """Cache-key strategy that hashes a node's immediate inputs plus ancestor references."""
     def __init__(self, dynprompt, node_ids, is_changed_cache):
+        """Initialize input-signature-based cache keys for the supplied dynamic prompt."""
         super().__init__(dynprompt, node_ids, is_changed_cache)
         self.dynprompt = dynprompt
         self.is_changed_cache = is_changed_cache
 
     def include_node_id_in_input(self) -> bool:
+        """Return whether node ids should be included in computed input signatures."""
         return False
 
     async def add_keys(self, node_ids):
+        """Populate input-signature-based keys for nodes in the dynamic prompt."""
         for node_id in node_ids:
             if node_id in self.keys:
                 continue
@@ -99,21 +465,37 @@ class CacheKeySetInputSignature(CacheKeySet):
             self.subcache_keys[node_id] = (node_id, node["class_type"])
 
     async def get_node_signature(self, dynprompt, node_id):
+        """Build the full cache signature for a node and its ordered ancestors."""
         signature = []
         ancestors, order_mapping = self.get_ordered_ancestry(dynprompt, node_id)
-        signature.append(await self.get_immediate_node_signature(dynprompt, node_id, order_mapping))
+        immediate = await self.get_immediate_node_signature(dynprompt, node_id, order_mapping)
+        if type(immediate) is Unhashable:
+            return immediate
+        signature.append(immediate)
         for ancestor_id in ancestors:
-            signature.append(await self.get_immediate_node_signature(dynprompt, ancestor_id, order_mapping))
-        return to_hashable(signature)
+            immediate = await self.get_immediate_node_signature(dynprompt, ancestor_id, order_mapping)
+            if type(immediate) is Unhashable:
+                return immediate
+            signature.append(immediate)
+        return tuple(signature)
 
     async def get_immediate_node_signature(self, dynprompt, node_id, ancestor_order_mapping):
+        """Build the immediate cache-signature fragment for a node.
+
+        Link inputs are reduced to ancestor references here. Non-link values
+        are canonicalized or failed closed before being appended so the final
+        node signature is assembled from already-hashable fragments.
+        """
         if not dynprompt.has_node(node_id):
             # This node doesn't exist -- we can't cache it.
-            return [float("NaN")]
+            return Unhashable()
         node = dynprompt.get_node(node_id)
         class_type = node["class_type"]
         class_def = nodes.NODE_CLASS_MAPPINGS[class_type]
-        signature = [class_type, await self.is_changed_cache.get(node_id)]
+        is_changed_signature = _shallow_is_changed_signature(await self.is_changed_cache.get(node_id))
+        if type(is_changed_signature) is Unhashable:
+            return is_changed_signature
+        signature = [class_type, is_changed_signature]
         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)
         inputs = node["inputs"]
@@ -123,18 +505,23 @@ class CacheKeySetInputSignature(CacheKeySet):
                 ancestor_index = ancestor_order_mapping[ancestor_id]
                 signature.append((key,("ANCESTOR", ancestor_index, ancestor_socket)))
             else:
-                signature.append((key, inputs[key]))
-        return signature
+                value_signature = to_hashable(inputs[key])
+                if type(value_signature) is Unhashable:
+                    return value_signature
+                signature.append((key, value_signature))
+        return tuple(signature)
 
     # This function returns a list of all ancestors of the given node. The order of the list is
     # deterministic based on which specific inputs the ancestor is connected by.
     def get_ordered_ancestry(self, dynprompt, node_id):
+        """Return ancestors in deterministic traversal order and their index mapping."""
         ancestors = []
         order_mapping = {}
         self.get_ordered_ancestry_internal(dynprompt, node_id, ancestors, order_mapping)
         return ancestors, order_mapping
 
     def get_ordered_ancestry_internal(self, dynprompt, node_id, ancestors, order_mapping):
+        """Recursively collect ancestors in input order without revisiting prior nodes."""
         if not dynprompt.has_node(node_id):
             return
         inputs = dynprompt.get_node(node_id)["inputs"]

comfy_execution/graph_utils.py

@@ -1,11 +1,17 @@
 def is_link(obj):
-    if not isinstance(obj, list):
+    """Return whether obj is a plain prompt link of the form [node_id, output_index]."""
+    # Prompt links produced by the frontend / GraphBuilder are plain Python
+    # lists in the form [node_id, output_index]. Some custom-node paths can
+    # inject foreign runtime objects into prompt inputs during on-prompt graph
+    # rewriting or subgraph construction. Be strict here so cache signature
+    # building never tries to treat list-like proxy objects as links.
+    if type(obj) is not list:
         return False
     if len(obj) != 2:
         return False
-    if not isinstance(obj[0], str):
+    if type(obj[0]) is not str:
         return False
-    if not isinstance(obj[1], int) and not isinstance(obj[1], float):
+    if type(obj[1]) is not int:
         return False
     return True
 

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()

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/caching_test.py

@@ -0,0 +1,473 @@
+"""Unit tests for cache-signature canonicalization hardening."""
+
+import asyncio
+import importlib
+import sys
+import types
+
+import pytest
+
+
+class _DummyNode:
+    """Minimal node stub used to satisfy cache-signature class lookups."""
+
+    @staticmethod
+    def INPUT_TYPES():
+        """Return a minimal empty input schema for unit tests."""
+        return {"required": {}}
+
+
+class _FakeDynPrompt:
+    """Small DynamicPrompt stand-in with only the methods these tests need."""
+
+    def __init__(self, nodes_by_id):
+        """Store test nodes by id."""
+        self._nodes_by_id = nodes_by_id
+
+    def has_node(self, node_id):
+        """Return whether the fake prompt contains the requested node."""
+        return node_id in self._nodes_by_id
+
+    def get_node(self, node_id):
+        """Return the stored node payload for the requested id."""
+        return self._nodes_by_id[node_id]
+
+
+class _FakeIsChangedCache:
+    """Async stub for `is_changed` lookups used by cache-key generation."""
+
+    def __init__(self, values):
+        """Store canned `is_changed` responses keyed by node id."""
+        self._values = values
+
+    async def get(self, node_id):
+        """Return the canned `is_changed` value for a node."""
+        return self._values[node_id]
+
+
+class _OpaqueValue:
+    """Hashable opaque object used to exercise fail-closed unordered hashing paths."""
+
+
+@pytest.fixture
+def caching_module(monkeypatch):
+    """Import `comfy_execution.caching` with lightweight stub dependencies."""
+    torch_module = types.ModuleType("torch")
+    psutil_module = types.ModuleType("psutil")
+    nodes_module = types.ModuleType("nodes")
+    nodes_module.NODE_CLASS_MAPPINGS = {}
+    graph_module = types.ModuleType("comfy_execution.graph")
+
+    class DynamicPrompt:
+        """Placeholder graph type so the caching module can import cleanly."""
+
+        pass
+
+    graph_module.DynamicPrompt = DynamicPrompt
+
+    monkeypatch.setitem(sys.modules, "torch", torch_module)
+    monkeypatch.setitem(sys.modules, "psutil", psutil_module)
+    monkeypatch.setitem(sys.modules, "nodes", nodes_module)
+    monkeypatch.setitem(sys.modules, "comfy_execution.graph", graph_module)
+    monkeypatch.delitem(sys.modules, "comfy_execution.caching", raising=False)
+
+    module = importlib.import_module("comfy_execution.caching")
+    module = importlib.reload(module)
+    return module, nodes_module
+
+
+def test_signature_to_hashable_handles_shared_builtin_substructures(caching_module):
+    """Shared built-in substructures should canonicalize without collapsing to Unhashable."""
+    caching, _ = caching_module
+    shared = [{"value": 1}, {"value": 2}]
+
+    signature = caching._signature_to_hashable([shared, shared])
+
+    assert signature[0] == "list"
+    assert signature[1][0] == signature[1][1]
+    assert signature[1][0][0] == "list"
+    assert signature[1][0][1][0] == ("dict", (("value", 1),))
+    assert signature[1][0][1][1] == ("dict", (("value", 2),))
+
+
+def test_signature_to_hashable_fails_closed_on_opaque_values(caching_module):
+    """Opaque values should collapse the full signature to Unhashable immediately."""
+    caching, _ = caching_module
+
+    signature = caching._signature_to_hashable(["safe", object()])
+
+    assert isinstance(signature, caching.Unhashable)
+
+
+def test_signature_to_hashable_stops_descending_after_failure(caching_module, monkeypatch):
+    """Once canonicalization fails, later recursive descent should stop immediately."""
+    caching, _ = caching_module
+    original = caching._signature_to_hashable_impl
+    marker = object()
+    marker_seen = False
+
+    def tracking_canonicalize(obj, *args, **kwargs):
+        """Track whether recursion reaches the nested marker after failure."""
+        nonlocal marker_seen
+        if obj is marker:
+            marker_seen = True
+        return original(obj, *args, **kwargs)
+
+    monkeypatch.setattr(caching, "_signature_to_hashable_impl", tracking_canonicalize)
+
+    signature = caching._signature_to_hashable([object(), [marker]])
+
+    assert isinstance(signature, caching.Unhashable)
+    assert marker_seen is False
+
+
+def test_signature_to_hashable_snapshots_list_before_recursing(caching_module, monkeypatch):
+    """List canonicalization should read a point-in-time snapshot before recursive descent."""
+    caching, _ = caching_module
+    original = caching._signature_to_hashable_impl
+    marker = ("marker",)
+    values = [marker, 2]
+
+    def mutating_canonicalize(obj, *args, **kwargs):
+        """Mutate the live list during recursion to verify snapshot-based traversal."""
+        if obj is marker:
+            values[1] = 3
+        return original(obj, *args, **kwargs)
+
+    monkeypatch.setattr(caching, "_signature_to_hashable_impl", mutating_canonicalize)
+
+    signature = caching._signature_to_hashable(values)
+
+    assert signature == ("list", (("tuple", ("marker",)), 2))
+    assert values[1] == 3
+
+
+def test_signature_to_hashable_snapshots_dict_before_recursing(caching_module, monkeypatch):
+    """Dict canonicalization should read a point-in-time snapshot before recursive descent."""
+    caching, _ = caching_module
+    original = caching._signature_to_hashable_impl
+    marker = ("marker",)
+    values = {"first": marker, "second": 2}
+
+    def mutating_canonicalize(obj, *args, **kwargs):
+        """Mutate the live dict during recursion to verify snapshot-based traversal."""
+        if obj is marker:
+            values["second"] = 3
+        return original(obj, *args, **kwargs)
+
+    monkeypatch.setattr(caching, "_signature_to_hashable_impl", mutating_canonicalize)
+
+    signature = caching._signature_to_hashable(values)
+
+    assert signature == ("dict", (("first", ("tuple", ("marker",))), ("second", 2)))
+    assert values["second"] == 3
+
+
+@pytest.mark.parametrize(
+    "container_factory",
+    [
+        lambda marker: [marker],
+        lambda marker: (marker,),
+        lambda marker: {marker},
+        lambda marker: frozenset({marker}),
+        lambda marker: {"key": marker},
+    ],
+)
+def test_signature_to_hashable_fails_closed_on_runtimeerror(caching_module, monkeypatch, container_factory):
+    """Traversal RuntimeError should degrade canonicalization to Unhashable."""
+    caching, _ = caching_module
+    original = caching._signature_to_hashable_impl
+    marker = object()
+
+    def raising_canonicalize(obj, *args, **kwargs):
+        """Raise a traversal RuntimeError for the marker value and delegate otherwise."""
+        if obj is marker:
+            raise RuntimeError("container changed during iteration")
+        return original(obj, *args, **kwargs)
+
+    monkeypatch.setattr(caching, "_signature_to_hashable_impl", raising_canonicalize)
+
+    signature = caching._signature_to_hashable(container_factory(marker))
+
+    assert isinstance(signature, caching.Unhashable)
+
+
+def test_to_hashable_handles_shared_builtin_substructures(caching_module):
+    """The legacy helper should still hash sanitized built-ins stably when used directly."""
+    caching, _ = caching_module
+    shared = [{"value": 1}, {"value": 2}]
+
+    sanitized = [shared, shared]
+    hashable = caching.to_hashable(sanitized)
+
+    assert hashable[0] == "list"
+    assert hashable[1][0] == hashable[1][1]
+    assert hashable[1][0][0] == "list"
+
+
+def test_to_hashable_uses_parent_snapshot_during_expanded_phase(caching_module, monkeypatch):
+    """Expanded-phase assembly should not reread a live parent container after snapshotting."""
+    caching, _ = caching_module
+    original_sort_key = caching._sanitized_sort_key
+    outer = [{"marker"}, 2]
+
+    def mutating_sort_key(obj, *args, **kwargs):
+        """Mutate the live parent while a child container is being canonicalized."""
+        if obj == "marker":
+            outer[1] = 3
+        return original_sort_key(obj, *args, **kwargs)
+
+    monkeypatch.setattr(caching, "_sanitized_sort_key", mutating_sort_key)
+
+    hashable = caching.to_hashable(outer)
+
+    assert hashable == ("list", (("set", ("marker",)), 2))
+    assert outer[1] == 3
+
+
+def test_to_hashable_fails_closed_for_ordered_container_with_opaque_child(caching_module):
+    """Ordered containers should fail closed when a child cannot be canonicalized."""
+    caching, _ = caching_module
+
+    result = caching.to_hashable([object()])
+
+    assert isinstance(result, caching.Unhashable)
+
+
+def test_to_hashable_canonicalizes_dict_insertion_order(caching_module):
+    """Dicts with the same content should hash identically regardless of insertion order."""
+    caching, _ = caching_module
+
+    first = {"b": 2, "a": 1}
+    second = {"a": 1, "b": 2}
+
+    assert caching.to_hashable(first) == ("dict", (("a", 1), ("b", 2)))
+    assert caching.to_hashable(first) == caching.to_hashable(second)
+
+
+def test_to_hashable_fails_closed_for_opaque_dict_key(caching_module):
+    """Opaque dict keys should fail closed instead of being traversed during hashing."""
+    caching, _ = caching_module
+
+    hashable = caching.to_hashable({_OpaqueValue(): 1})
+
+    assert isinstance(hashable, caching.Unhashable)
+
+
+@pytest.mark.parametrize(
+    "container_factory",
+    [
+        set,
+        frozenset,
+    ],
+)
+def test_to_hashable_fails_closed_on_runtimeerror(caching_module, monkeypatch, container_factory):
+    """Traversal RuntimeError should degrade unordered hash conversion to Unhashable."""
+    caching, _ = caching_module
+
+    def raising_sort_key(obj, *args, **kwargs):
+        """Raise a traversal RuntimeError while unordered values are canonicalized."""
+        raise RuntimeError("container changed during iteration")
+
+    monkeypatch.setattr(caching, "_sanitized_sort_key", raising_sort_key)
+
+    hashable = caching.to_hashable(container_factory({"value"}))
+
+    assert isinstance(hashable, caching.Unhashable)
+
+
+def test_to_hashable_fails_closed_for_ambiguous_dict_ordering(caching_module, monkeypatch):
+    """Ambiguous dict key ordering should fail closed instead of using insertion order."""
+    caching, _ = caching_module
+    original_sort_key = caching._sanitized_sort_key
+    ambiguous = {"a": 1, "b": 1}
+
+    def colliding_sort_key(obj, *args, **kwargs):
+        """Force two distinct primitive keys to share the same ordering key."""
+        if obj == "a" or obj == "b":
+            return ("COLLIDE",)
+        return original_sort_key(obj, *args, **kwargs)
+
+    monkeypatch.setattr(caching, "_sanitized_sort_key", colliding_sort_key)
+
+    hashable = caching.to_hashable(ambiguous)
+
+    assert isinstance(hashable, caching.Unhashable)
+
+
+def test_signature_to_hashable_fails_closed_for_ambiguous_dict_ordering(caching_module, monkeypatch):
+    """Ambiguous dict sort ties should fail closed instead of depending on input order."""
+    caching, _ = caching_module
+    original_sort_key = caching._primitive_signature_sort_key
+    ambiguous = {"a": 1, "b": 1}
+
+    def colliding_sort_key(obj):
+        """Force two distinct primitive keys to share the same ordering key."""
+        if obj == "a" or obj == "b":
+            return ("COLLIDE",)
+        return original_sort_key(obj)
+
+    monkeypatch.setattr(caching, "_primitive_signature_sort_key", colliding_sort_key)
+
+    sanitized = caching._signature_to_hashable(ambiguous)
+
+    assert isinstance(sanitized, caching.Unhashable)
+
+
+def test_signature_to_hashable_fails_closed_for_opaque_dict_key(caching_module):
+    """Opaque dict keys should fail closed instead of being recursively canonicalized."""
+    caching, _ = caching_module
+
+    sanitized = caching._signature_to_hashable({_OpaqueValue(): 1})
+
+    assert isinstance(sanitized, caching.Unhashable)
+
+
+def test_signature_to_hashable_fails_closed_on_dict_key_sort_collisions_even_with_distinct_values(caching_module, monkeypatch):
+    """Different values must not mask dict key-sort collisions during canonicalization."""
+    caching, _ = caching_module
+    original_sort_key = caching._primitive_signature_sort_key
+
+    def colliding_sort_key(obj):
+        """Force two distinct primitive keys to share the same ordering key."""
+        if obj == "a" or obj == "b":
+            return ("COLLIDE",)
+        return original_sort_key(obj)
+
+    monkeypatch.setattr(caching, "_primitive_signature_sort_key", colliding_sort_key)
+
+    sanitized = caching._signature_to_hashable({"a": 1, "b": 2})
+
+    assert isinstance(sanitized, caching.Unhashable)
+
+
+@pytest.mark.parametrize(
+    "container_factory",
+    [
+        set,
+        frozenset,
+    ],
+)
+def test_to_hashable_fails_closed_for_ambiguous_unordered_values(caching_module, monkeypatch, container_factory):
+    """Ambiguous unordered values should fail closed instead of depending on iteration order."""
+    caching, _ = caching_module
+    original_sort_key = caching._sanitized_sort_key
+    container = container_factory({"a", "b"})
+
+    def colliding_sort_key(obj, *args, **kwargs):
+        """Force two distinct primitive values to share the same ordering key."""
+        if obj == "a" or obj == "b":
+            return ("COLLIDE",)
+        return original_sort_key(obj, *args, **kwargs)
+
+    monkeypatch.setattr(caching, "_sanitized_sort_key", colliding_sort_key)
+
+    hashable = caching.to_hashable(container)
+
+    assert isinstance(hashable, caching.Unhashable)
+
+
+def test_get_node_signature_returns_top_level_unhashable_for_tainted_signature(caching_module, monkeypatch):
+    """Tainted full signatures should fail closed before `to_hashable()` runs."""
+    caching, nodes_module = caching_module
+    monkeypatch.setitem(nodes_module.NODE_CLASS_MAPPINGS, "UnitTestNode", _DummyNode)
+    monkeypatch.setattr(
+        caching,
+        "to_hashable",
+        lambda *_args, **_kwargs: pytest.fail("to_hashable should not run for tainted signatures"),
+    )
+
+    is_changed_value = []
+    is_changed_value.append(is_changed_value)
+
+    dynprompt = _FakeDynPrompt(
+        {
+            "node": {
+                "class_type": "UnitTestNode",
+                "inputs": {"value": 5},
+            }
+        }
+    )
+    key_set = caching.CacheKeySetInputSignature(
+        dynprompt,
+        ["node"],
+        _FakeIsChangedCache({"node": is_changed_value}),
+    )
+
+    signature = asyncio.run(key_set.get_node_signature(dynprompt, "node"))
+
+    assert isinstance(signature, caching.Unhashable)
+
+
+def test_shallow_is_changed_signature_accepts_primitive_lists(caching_module):
+    """Primitive-only `is_changed` lists should stay hashable without deep descent."""
+    caching, _ = caching_module
+
+    sanitized = caching._shallow_is_changed_signature([1, "two", None, True])
+
+    assert sanitized == ("is_changed_list", (1, "two", None, True))
+
+
+def test_shallow_is_changed_signature_accepts_structured_builtin_fingerprint_lists(caching_module):
+    """Structured built-in `is_changed` fingerprints should remain representable."""
+    caching, _ = caching_module
+
+    sanitized = caching._shallow_is_changed_signature([("seed", 42), {"cfg": 8}])
+
+    assert sanitized == (
+        "is_changed_list",
+        (
+            ("tuple", ("seed", 42)),
+            ("dict", (("cfg", 8),)),
+        ),
+    )
+
+
+def test_shallow_is_changed_signature_fails_closed_for_opaque_payload(caching_module):
+    """Opaque `is_changed` payloads should still fail closed."""
+    caching, _ = caching_module
+
+    sanitized = caching._shallow_is_changed_signature([_OpaqueValue()])
+
+    assert isinstance(sanitized, caching.Unhashable)
+
+
+def test_get_immediate_node_signature_fails_closed_for_unhashable_is_changed(caching_module, monkeypatch):
+    """Recursive `is_changed` payloads should fail the full fragment closed."""
+    caching, nodes_module = caching_module
+    monkeypatch.setitem(nodes_module.NODE_CLASS_MAPPINGS, "UnitTestNode", _DummyNode)
+
+    is_changed_value = []
+    is_changed_value.append(is_changed_value)
+    dynprompt = _FakeDynPrompt(
+        {
+            "node": {
+                "class_type": "UnitTestNode",
+                "inputs": {"value": 5},
+            }
+        }
+    )
+    key_set = caching.CacheKeySetInputSignature(
+        dynprompt,
+        ["node"],
+        _FakeIsChangedCache({"node": is_changed_value}),
+    )
+
+    signature = asyncio.run(key_set.get_immediate_node_signature(dynprompt, "node", {}))
+
+    assert isinstance(signature, caching.Unhashable)
+
+
+def test_get_immediate_node_signature_fails_closed_for_missing_node(caching_module):
+    """Missing nodes should return the fail-closed sentinel instead of a NaN tuple."""
+    caching, _ = caching_module
+    dynprompt = _FakeDynPrompt({})
+    key_set = caching.CacheKeySetInputSignature(
+        dynprompt,
+        [],
+        _FakeIsChangedCache({}),
+    )
+
+    signature = asyncio.run(key_set.get_immediate_node_signature(dynprompt, "missing", {}))
+
+    assert isinstance(signature, caching.Unhashable)

tests/execution/test_caching.py

@@ -0,0 +1,242 @@
+import asyncio
+
+from comfy_execution import caching
+
+
+class _StubDynPrompt:
+    def __init__(self, nodes):
+        self._nodes = nodes
+
+    def has_node(self, node_id):
+        return node_id in self._nodes
+
+    def get_node(self, node_id):
+        return self._nodes[node_id]
+
+
+class _StubIsChangedCache:
+    async def get(self, node_id):
+        return None
+
+
+class _StubNode:
+    @classmethod
+    def INPUT_TYPES(cls):
+        return {"required": {}}
+
+
+def test_shallow_is_changed_signature_keeps_primitive_only_list_shallow():
+    assert caching._shallow_is_changed_signature([1, "two", None, True]) == (
+        "is_changed_list",
+        (1, "two", None, True),
+    )
+
+
+def test_shallow_is_changed_signature_keeps_primitive_only_tuple_shallow():
+    assert caching._shallow_is_changed_signature((1, "two", None, True)) == (
+        "is_changed_tuple",
+        (1, "two", None, True),
+    )
+
+
+def test_shallow_is_changed_signature_keeps_structured_builtin_fingerprint_list():
+    assert caching._shallow_is_changed_signature([("seed", 42), {"cfg": 8}]) == (
+        "is_changed_list",
+        (
+            ("tuple", ("seed", 42)),
+            ("dict", (("cfg", 8),)),
+        ),
+    )
+
+
+def test_shallow_is_changed_signature_does_not_use_to_hashable(monkeypatch):
+    monkeypatch.setattr(
+        caching,
+        "to_hashable",
+        lambda *_args, **_kwargs: (_ for _ in ()).throw(
+            AssertionError("is_changed signature must not deep-canonicalize")
+        ),
+    )
+
+    signature = caching._shallow_is_changed_signature([("seed", 42), {"cfg": 8}])
+
+    assert signature == (
+        "is_changed_list",
+        (
+            ("tuple", ("seed", 42)),
+            ("dict", (("cfg", 8),)),
+        ),
+    )
+
+
+def test_get_immediate_node_signature_canonicalizes_non_link_inputs(monkeypatch):
+    live_value = [1, {"nested": [2, 3]}]
+    dynprompt = _StubDynPrompt(
+        {
+            "1": {
+                "class_type": "TestCacheNode",
+                "inputs": {"value": live_value},
+            }
+        }
+    )
+
+    monkeypatch.setitem(caching.nodes.NODE_CLASS_MAPPINGS, "TestCacheNode", _StubNode)
+    monkeypatch.setattr(caching, "NODE_CLASS_CONTAINS_UNIQUE_ID", {})
+
+    keyset = caching.CacheKeySetInputSignature(dynprompt, [], _StubIsChangedCache())
+    signature = asyncio.run(keyset.get_immediate_node_signature(dynprompt, "1", {}))
+
+    assert signature == (
+        "TestCacheNode",
+        None,
+        ("value", ("list", (1, ("dict", (("nested", ("list", (2, 3))),))))),
+    )
+
+
+def test_to_hashable_walks_dicts_without_rebinding_traversal_stack():
+    live_value = {
+        "outer": {"nested": [2, 3]},
+        "items": [{"leaf": 4}],
+    }
+
+    assert caching.to_hashable(live_value) == (
+        "dict",
+        (
+            ("items", ("list", (("dict", (("leaf", 4),)),))),
+            ("outer", ("dict", (("nested", ("list", (2, 3))),))),
+        ),
+    )
+
+
+def test_get_immediate_node_signature_fails_closed_for_opaque_non_link_input(monkeypatch):
+    class OpaqueRuntimeValue:
+        pass
+
+    live_value = OpaqueRuntimeValue()
+    dynprompt = _StubDynPrompt(
+        {
+            "1": {
+                "class_type": "TestCacheNode",
+                "inputs": {"value": live_value},
+            }
+        }
+    )
+
+    monkeypatch.setitem(caching.nodes.NODE_CLASS_MAPPINGS, "TestCacheNode", _StubNode)
+    monkeypatch.setattr(caching, "NODE_CLASS_CONTAINS_UNIQUE_ID", {})
+
+    keyset = caching.CacheKeySetInputSignature(dynprompt, [], _StubIsChangedCache())
+    signature = asyncio.run(keyset.get_immediate_node_signature(dynprompt, "1", {}))
+
+    assert isinstance(signature, caching.Unhashable)
+
+
+def test_get_node_signature_propagates_unhashable_immediate_fragment(monkeypatch):
+    class OpaqueRuntimeValue:
+        pass
+
+    dynprompt = _StubDynPrompt(
+        {
+            "1": {
+                "class_type": "TestCacheNode",
+                "inputs": {"value": OpaqueRuntimeValue()},
+            }
+        }
+    )
+
+    monkeypatch.setitem(caching.nodes.NODE_CLASS_MAPPINGS, "TestCacheNode", _StubNode)
+    monkeypatch.setattr(caching, "NODE_CLASS_CONTAINS_UNIQUE_ID", {})
+
+    keyset = caching.CacheKeySetInputSignature(dynprompt, [], _StubIsChangedCache())
+    signature = asyncio.run(keyset.get_node_signature(dynprompt, "1"))
+
+    assert isinstance(signature, caching.Unhashable)
+
+
+def test_get_node_signature_never_visits_raw_non_link_input(monkeypatch):
+    live_value = [1, 2, 3]
+    dynprompt = _StubDynPrompt(
+        {
+            "1": {
+                "class_type": "TestCacheNode",
+                "inputs": {"value": live_value},
+            }
+        }
+    )
+
+    monkeypatch.setitem(caching.nodes.NODE_CLASS_MAPPINGS, "TestCacheNode", _StubNode)
+    monkeypatch.setattr(caching, "NODE_CLASS_CONTAINS_UNIQUE_ID", {})
+    monkeypatch.setattr(
+        caching,
+        "_signature_to_hashable",
+        lambda *_args, **_kwargs: (_ for _ in ()).throw(
+            AssertionError("outer signature canonicalizer should not run")
+        ),
+    )
+
+    keyset = caching.CacheKeySetInputSignature(dynprompt, [], _StubIsChangedCache())
+    signature = asyncio.run(keyset.get_node_signature(dynprompt, "1"))
+
+    assert isinstance(signature, tuple)
+
+
+def test_get_node_signature_keeps_deep_canonicalized_input_fragment(monkeypatch):
+    live_value = 1
+    for _ in range(8):
+        live_value = [live_value]
+    expected = caching.to_hashable(live_value)
+
+    dynprompt = _StubDynPrompt(
+        {
+            "1": {
+                "class_type": "TestCacheNode",
+                "inputs": {"value": live_value},
+            }
+        }
+    )
+
+    monkeypatch.setitem(caching.nodes.NODE_CLASS_MAPPINGS, "TestCacheNode", _StubNode)
+    monkeypatch.setattr(caching, "NODE_CLASS_CONTAINS_UNIQUE_ID", {})
+
+    keyset = caching.CacheKeySetInputSignature(dynprompt, [], _StubIsChangedCache())
+    signature = asyncio.run(keyset.get_node_signature(dynprompt, "1"))
+
+    assert isinstance(signature, tuple)
+    assert signature[0][2][0] == "value"
+    assert signature[0][2][1] == expected
+
+
+def test_get_node_signature_keeps_large_precanonicalized_fragment(monkeypatch):
+    live_value = object()
+    canonical_fragment = ("tuple", tuple(("list", (index, index + 1)) for index in range(256)))
+    dynprompt = _StubDynPrompt(
+        {
+            "1": {
+                "class_type": "TestCacheNode",
+                "inputs": {"value": live_value},
+            }
+        }
+    )
+
+    monkeypatch.setitem(caching.nodes.NODE_CLASS_MAPPINGS, "TestCacheNode", _StubNode)
+    monkeypatch.setattr(caching, "NODE_CLASS_CONTAINS_UNIQUE_ID", {})
+    monkeypatch.setattr(
+        caching,
+        "to_hashable",
+        lambda value, max_nodes=caching._MAX_SIGNATURE_CONTAINER_VISITS: (
+            canonical_fragment if value is live_value else caching.Unhashable()
+        ),
+    )
+    monkeypatch.setattr(
+        caching,
+        "_signature_to_hashable",
+        lambda *_args, **_kwargs: (_ for _ in ()).throw(
+            AssertionError("outer signature canonicalizer should not run")
+        ),
+    )
+
+    keyset = caching.CacheKeySetInputSignature(dynprompt, [], _StubIsChangedCache())
+    signature = asyncio.run(keyset.get_node_signature(dynprompt, "1"))
+
+    assert isinstance(signature, tuple)
+    assert signature[0][2] == ("value", canonical_fragment)