fix(image-list): reject upstream data bypasses around Start

image_list/runtime.py

@@ -395,6 +395,41 @@ def _downstream_until(prompt: dict[str, Any], start_id: str, end_id: str) -> set
     return visited
 
 
+def _start_ancestors(prompt: dict[str, Any], start_id: str) -> set[str]:
+    """All dependencies materialized on the master, without crossing Start again."""
+    ancestors = set()
+    pending = [source for _, source, _ in _prompt_links(prompt[start_id], prompt)]
+    while pending:
+        node_id = pending.pop()
+        if node_id == start_id or node_id in ancestors:
+            continue
+        ancestors.add(node_id)
+        pending.extend(source for _, source, _ in _prompt_links(prompt[node_id], prompt))
+    return ancestors
+
+
+def _safe_shared_dependency(node: dict[str, Any], prompt: dict[str, Any], registry) -> bool:
+    """Only typed model dependencies or scalar constants may straddle Start.
+
+    A leaf alone is not sufficient: LoadImage/EmptyImage are leaves too. Unknown
+    classes and list-producing nodes fail closed. All shared ancestors are checked,
+    so a model-producing node cannot hide a shared image/latent dependency.
+    """
+    node_class = registry.get(node.get("class_type"))
+    if node_class is None:
+        return False
+    info = node_class.GET_NODE_INFO_V1() if hasattr(node_class, "GET_NODE_INFO_V1") else {}
+    outputs = info.get("output", getattr(node_class, "RETURN_TYPES", ()))
+    if not outputs or any(info.get("output_is_list", getattr(node_class, "OUTPUT_IS_LIST", ()))):
+        return False
+    types = {str(output) for output in outputs}
+    model_types = {"MODEL", "CLIP", "VAE", "CLIP_VISION", "CONTROL_NET"}
+    scalar_types = {"INT", "FLOAT", "STRING", "BOOLEAN"}
+    if types <= model_types:
+        return True
+    return types <= scalar_types and not any(_prompt_links(node, prompt))
+
+
 def validate_image_list_regions(prompt: dict[str, Any]) -> list[dict[str, Any]]:
     """Validate pairing, non-nesting, and closed per-image graph boundaries."""
     if not isinstance(prompt, dict):
@@ -431,6 +466,18 @@ def validate_image_list_regions(prompt: dict[str, Any]) -> list[dict[str, Any]]:
         # evaluate the processing branch on the master as well as the worker.
         import sys
         registry = getattr(sys.modules.get("nodes"), "NODE_CLASS_MAPPINGS", {})
+        shared = region_nodes & _start_ancestors(prompt, start_id)
+        unsafe = {node_id for node_id in shared
+                  if not _safe_shared_dependency(prompt[node_id], prompt, registry)}
+        if unsafe:
+            links = [f"{node_id}.{name} <- {source}:{slot}"
+                     for node_id in sorted(region_nodes - {start_id})
+                     for name, source, slot in _prompt_links(prompt[node_id], prompt)
+                     if source in unsafe]
+            raise ValueError(
+                f"Image-list region {start_id}->{end_id} bypasses Start through upstream nodes "
+                f"{sorted(unsafe)} ({'; '.join(links)}); route per-item data through Start"
+            )
         for node_id in region_nodes - {start_id, end_id}:
             node_class = registry.get(prompt[node_id].get("class_type"))
             if node_class is None:

tests/test_image_list.py

@@ -151,6 +151,49 @@ class ImageListGraphBoundaryTests(unittest.TestCase):
         self.assertEqual(worker["4"]["inputs"]["model"], ["1", 0])
         self.assertNotIn("6", worker)
 
+    def test_rejects_upstream_processing_and_leaf_image_bypasses(self):
+        # Both a leaf source and a generated batch must cross Start exactly once.
+        for generated in (False, True):
+            with self.subTest(generated=generated):
+                prompt = graph_one_region()
+                if generated:
+                    prompt["2"] = {"class_type": "VAEDecode", "inputs": {"samples": ["7", 0]}}
+                    prompt["7"] = {"class_type": "KSampler", "inputs": {"model": ["1", 0]}}
+                prompt["4"]["inputs"]["reference"] = ["2", 0]
+                with self.assertRaisesRegex(ValueError, "bypasses Start.*2"):
+                    validate_image_list_regions(prompt)
+
+    def test_shared_typed_loader_and_leaf_constant_remain_valid(self):
+        import sys
+        import types
+        from unittest.mock import patch
+        prompt = graph_one_region()
+        prompt["7"] = {"class_type": "SeedConstant", "inputs": {"value": 9}}
+        prompt["2"]["inputs"].update({"model": ["1", 0], "seed": ["7", 0]})
+        prompt["4"]["inputs"]["shared_seed"] = ["7", 0]
+        registry = {
+            "CheckpointLoaderSimple": type("Loader", (), {"RETURN_TYPES": ("MODEL", "CLIP", "VAE")}),
+            "SeedConstant": type("Constant", (), {"RETURN_TYPES": ("INT",)}),
+        }
+        with patch.dict(sys.modules, {"nodes": types.SimpleNamespace(NODE_CLASS_MAPPINGS=registry)}):
+            region = validate_image_list_regions(prompt)[0]
+        worker = build_worker_prompt(prompt, region, "token", "ctx", 0, "http://master")
+        self.assertNotIn("2", worker)
+        self.assertTrue({"1", "7"}.issubset(worker))
+
+    def test_unknown_or_list_producing_shared_dependency_is_not_exempt(self):
+        import sys
+        import types
+        from unittest.mock import patch
+        for node_class in (None, type("ListLoader", (), {
+                "RETURN_TYPES": ("MODEL",), "OUTPUT_IS_LIST": (True,)})):
+            prompt = graph_one_region()
+            prompt["2"]["inputs"]["model"] = ["1", 0]
+            registry = {} if node_class is None else {"CheckpointLoaderSimple": node_class}
+            with patch.dict(sys.modules, {"nodes": types.SimpleNamespace(NODE_CLASS_MAPPINGS=registry)}):
+                with self.assertRaisesRegex(ValueError, "bypasses Start.*1"):
+                    validate_image_list_regions(prompt)
+
     def test_two_sequential_regions_are_valid_and_isolated(self):
         regions = validate_image_list_regions(graph_two_sequential_regions())
         self.assertEqual([(r["start_id"], r["end_id"]) for r in regions], [("2", "4"), ("5", "7")])

tests/test_image_list_integration_gaps.py

@@ -172,7 +172,11 @@ class SchedulingBoundaryTests(unittest.IsolatedAsyncioTestCase):
         })
         for node_id in ['3', '6', '9']:
             prompt[node_id]['inputs']['model'] = ['11', 0]
-        regions = runtime.validate_image_list_regions(prompt)
+        # Shared dependencies are verified against the registered output schema.
+        loader = type('GenericModelLoader', (), {'RETURN_TYPES': ('MODEL',)})
+        registry = types.SimpleNamespace(NODE_CLASS_MAPPINGS={'GenericModelLoader': loader})
+        with patch.dict(sys.modules, {'nodes': registry}):
+            regions = runtime.validate_image_list_regions(prompt)
         self.assertEqual(len(regions), 3)
         for region in regions:
             self.assertIn('11', region['node_ids'])