From 7504bcd20c0376100ec8e82abccd92dc68c6ff50 Mon Sep 17 00:00:00 2001 From: Vito Sansevero Date: Mon, 19 Oct 2026 11:19:35 -0700 Subject: [PATCH] fix(inspect): queue each node once in breadth-first ancestry walks The latent and detailer image walks queued every link before checking whether its node was already seen, and popped from the list head. A dense shared fan-in grew the queue far past the visit budget before any visit was charged. Both walks now use a deque and enqueue each node at most once, charging every examined edge as scanned work; outcomes are unchanged. --- tests/test_prompt_pair_resolver.py | 86 ++++++++++++++++++++++++++++++ utils/prompt_pair_resolver.py | 52 ++++++++++-------- 2 files changed, 116 insertions(+), 22 deletions(-) diff --git a/tests/test_prompt_pair_resolver.py b/tests/test_prompt_pair_resolver.py index 78009a9..13b1dcc 100644 --- a/tests/test_prompt_pair_resolver.py +++ b/tests/test_prompt_pair_resolver.py @@ -1,5 +1,6 @@ """Conservative prompt-pair resolution contracts, independent of ComfyUI.""" +import collections import random import subprocess # nosec B404 - fixed interpreter command, no shell import sys @@ -2634,6 +2635,91 @@ class TestLatentAncestry(unittest.TestCase): self.assertEqual(_inspect(source)["state"], "resolved") +class _CountingDeque(collections.deque): + """Counts every item queued by the resolver's breadth-first walks.""" + + queued = 0 + + def append(self, item): + _CountingDeque.queued += 1 + super().append(item) + + def extend(self, items): + for item in items: + self.append(item) + + +def _dense_layers( + prefix: str, bottom: list, layers: int = 4, width: int = 40, fan: int = None +): + """A hub over ``layers`` x ``width`` nodes; each links to ``fan`` (default: + all) nodes of the next layer, so every node is shared by many consumers.""" + nodes = {} + fan = fan or width + for layer in range(layers): + for index in range(width): + below = ( + [ + [f"{prefix}{layer + 1}_{(index + step) % width}", 0] + for step in range(fan) + ] + if layer + 1 < layers + else [bottom] + ) + nodes[f"{prefix}{layer}_{index}"] = { + "class_type": "DenseBlend", + "inputs": {f"in_{k}": link for k, link in enumerate(below)}, + } + # One hub reads the whole first layer. + nodes[f"{prefix}top"] = { + "class_type": "DenseBlend", + "inputs": {f"in_{k}": [f"{prefix}0_{k}", 0] for k in range(width)}, + } + return nodes, [f"{prefix}top", 0], layers * width + 1 + + +class TestBreadthFirstWalkWork(unittest.TestCase): + """Dense shared fan-in: each node is queued at most once per walk.""" + + def counted(self, source: dict, **options) -> tuple: + _CountingDeque.queued = 0 + with mock.patch.object(resolver, "deque", _CountingDeque): + result = _inspect(source, **options) + return result, _CountingDeque.queued + + def test_dense_latent_ancestry_queues_each_node_once(self): + source = _latent_from({"class_type": "EmptyLatentImage", "inputs": {}}) + dense, top, count = _dense_layers("L", ["5", 0]) + source.update(dense) + source["3"]["inputs"]["latent_image"] = top + result, queued = self.counted(source) + self.assertEqual(result["state"], "resolved") + self.assertEqual(_sides(result), (parity.POS, parity.NEG)) + # The dense layers plus the empty latent, each queued once. + self.assertLessEqual(queued, count + 1) + + def test_dense_detailer_image_ancestry_queues_each_node_once(self): + source = _detailed(same_prompt=True) + expected = _inspect(source, output_node_id="9") + dense, top, count = _dense_layers("D", source["50"]["inputs"]["image"]) + source.update(dense) + source["50"]["inputs"]["image"] = top + result, queued = self.counted(source, output_node_id="9") + self.assertEqual(_outcomes(result), _outcomes(expected)) + # Dense layers + the decoder + its sampler, plus the latent walk of + # that sampler (one EmptyLatentImage). + self.assertLessEqual(queued, count + 3) + + def test_dense_fan_in_beyond_the_budget_is_work_limit(self): + source = _latent_from({"class_type": "EmptyLatentImage", "inputs": {}}) + dense, top, count = _dense_layers("W", ["5", 0], layers=3, width=800, fan=40) + source.update(dense) + source["3"]["inputs"]["latent_image"] = top + result, queued = self.counted(source) + self.assertEqual(result["reason_code"], "work-limit") + self.assertLessEqual(queued, count + 1) + + REPO = Path(__file__).resolve().parents[1] SEMANTICS_UNAVAILABLE = "Inspection node semantics are unavailable" diff --git a/utils/prompt_pair_resolver.py b/utils/prompt_pair_resolver.py index dbffea9..34fef10 100644 --- a/utils/prompt_pair_resolver.py +++ b/utils/prompt_pair_resolver.py @@ -30,6 +30,7 @@ When the semantics table cannot be loaded, inspection of a prompt graph is import logging import re +from collections import deque from dataclasses import dataclass, replace from typing import Any, Dict, Mapping, Optional, Tuple @@ -987,6 +988,27 @@ class _GraphResolver: reason_code="multistage-differs", ) + @staticmethod + def walk_from(value: list) -> Tuple[deque, set]: + """A breadth-first queue holding one start link, and its queued IDs.""" + start = str(value[0]) + return deque([(start, 0)]), {start} + + def enqueue(self, queue: deque, queued: set, inputs: dict, depth: int) -> None: + """Queue each linked input once per walk; every edge is charged as scanned. + + Deduplicating at enqueue keeps a dense shared fan-in to one entry per + node, so the queue never outgrows the nodes the visit budget pays for. + Breadth-first, a node's first enqueue is at its least depth, as before. + """ + self.scan(len(inputs)) + for item in inputs.values(): + if _link(item): + key = str(item[0]) + if key not in queued: + queued.add(key) + queue.append((key, depth)) + def upstream_stages(self, origin: str, value: Any) -> list: """Samplers reached from a detailer's image input, in breadth-first order. @@ -996,26 +1018,19 @@ class _GraphResolver: if not _link(value): raise _Unresolved("Expected an image link", "malformed") found: list = [] - seen: set = set() - queue = [(str(value[0]), 0)] + queue, queued = self.walk_from(value) while queue: - node_id, depth = queue.pop(0) - if node_id in seen: - continue + node_id, depth = queue.popleft() if node_id == origin: raise _Unresolved("Cyclic sampler stage ancestry", "malformed") if depth >= self.max_depth: raise _Unresolved("Resolution depth limit reached", "depth-limit") - seen.add(node_id) self.budget.spend() name, inputs = self.node(node_id) if name in self.stage_classes: found.append(node_id) continue - self.scan(len(inputs)) - queue.extend( - (str(item[0]), depth + 1) for item in inputs.values() if _link(item) - ) + self.enqueue(queue, queued, inputs, depth + 1) return found def final_stages(self, sampler_ids: list) -> list: @@ -1065,9 +1080,9 @@ class _GraphResolver: def reaches_output(self, node_id: str, consumers: dict) -> bool: """Whether a sampler's output reaches a terminal node besides later stages.""" - seen, queue = {node_id}, [node_id] + seen, queue = {node_id}, deque([node_id]) while queue: - current = queue.pop(0) + current = queue.popleft() for consumer in consumers.get(current, ()): if consumer in seen: continue @@ -1210,17 +1225,13 @@ class _GraphResolver: """ if not _link(value): raise _Unresolved("Unknown latent ancestry", "malformed") - seen: set = set() - queue = [(str(value[0]), 0)] + queue, queued = self.walk_from(value) while queue: - node_id, depth = queue.pop(0) - if node_id in seen: - continue + node_id, depth = queue.popleft() if node_id == origin: raise _Unresolved("Cyclic latent ancestry", "malformed") if depth >= self.max_depth: raise _Unresolved("Resolution depth limit reached", "depth-limit") - seen.add(node_id) self.budget.spend() name, inputs = self.node(node_id) if name in self.stage_classes: @@ -1231,10 +1242,7 @@ class _GraphResolver: _link(inputs.get(key)) for key in _SAMPLER_LIKE_INPUTS ): raise self.unsupported("Unsupported latent ancestry", node_id) - self.scan(len(inputs)) - queue.extend( - (str(item[0]), depth + 1) for item in inputs.values() if _link(item) - ) + self.enqueue(queue, queued, inputs, depth + 1) def anchor(self, node_id: str) -> str: seen = set()