feat: preserve required output utilities in image-list execution

tests/api/test_worker_routes.py

@@ -271,6 +271,74 @@ def _load_worker_routes_module():
 worker_routes = _load_worker_routes_module()
 
 
+class WebsocketExecutionTargetTests(unittest.IsolatedAsyncioTestCase):
+    def setUp(self):
+        from test_async_helpers import _load_async_helpers_module
+        self.modules = patch.dict(sys.modules)
+        self.modules.start()
+        self.addCleanup(self.modules.stop)
+        self.helper, self.server = _load_async_helpers_module()
+        sys.modules['dist_api_worker_testpkg.utils.async_helpers'] = self.helper
+        self.prompt = {'9': {'class_type': 'Output'}, '10': {'class_type': 'Output'}}
+
+    async def receive(self, payload):
+        import json
+        sent = []
+        class Socket:
+            async def prepare(self, request):
+                pass
+            async def send_json(self, value):
+                sent.append(value)
+            def __aiter__(self):
+                async def messages():
+                    yield types.SimpleNamespace(type=worker_routes.aiohttp.WSMsgType.TEXT,
+                                                data=json.dumps(payload))
+                return messages()
+        with patch.object(worker_routes, 'web', types.SimpleNamespace(WebSocketResponse=lambda **kw: Socket())), \
+             patch.object(worker_routes, 'queue_prompt_payload', self.helper.queue_prompt_payload), \
+             patch.object(worker_routes, 'PromptValidationError', self.helper.PromptValidationError):
+            await worker_routes.worker_ws_endpoint(_FakeRequest())
+        return sent[0]
+
+    def payload(self, targets):
+        return {'type': 'dispatch_prompt', 'request_id': 'req', 'prompt': self.prompt,
+                'partial_execution_targets': targets}
+
+    async def test_explicit_targets_reach_queue_and_acknowledgement(self):
+        ack = await self.receive(self.payload(['9']))
+        self.assertTrue(ack['ok'], ack)
+        self.assertEqual(ack['request_id'], 'req')
+        self.assertEqual(self.server.prompt_queue.items[0][4], ['9'])
+
+    async def test_malformed_targets_and_messages_are_rejected(self):
+        for targets in (None, [], '9', [True], [9], ['absent'], ['9', '9']):
+            with self.subTest(targets=targets):
+                ack = await self.receive(self.payload(targets))
+                self.assertFalse(ack['ok'])
+                self.assertIn('partial_execution_targets', ack['error'])
+        self.assertFalse((await self.receive([]))['ok'])
+        self.assertEqual(self.server.prompt_queue.items, [])
+
+    async def test_validation_cannot_silently_queue_a_partial_subset(self):
+        ack = await self.receive(self.payload(['9', '10']))
+        self.assertFalse(ack['ok'])
+        self.assertEqual(ack['validation_error']['type'], 'partial_execution_targets_failed_validation')
+        self.assertEqual(self.server.prompt_queue.items, [])
+
+    async def test_missing_target_field_keeps_legacy_queue_call(self):
+        from unittest.mock import AsyncMock
+        queued = AsyncMock(return_value='legacy')
+        original = self.helper.queue_prompt_payload
+        self.helper.queue_prompt_payload = queued
+        try:
+            payload = self.payload(['9'])
+            payload.pop('partial_execution_targets')
+            self.assertTrue((await self.receive(payload))['ok'])
+            self.assertNotIn('partial_execution_targets', queued.call_args.kwargs)
+        finally:
+            self.helper.queue_prompt_payload = original
+
+
 class WorkerRoutesTests(unittest.IsolatedAsyncioTestCase):
     async def test_network_info_reports_actual_master_and_available_ports(self):
         with patch.object(worker_routes, "_get_cuda_info", return_value=(0, 1, 4)), \

tests/helpers/image_list_network_check.py

@@ -1,11 +1,12 @@
 """Model-free multi-process acceptance using real ComfyUI executors and loopback HTTP.
 
-Only the transport test facade (/prompt, /queue, /history, /interrupt) is synthetic.
+Isolated listeners use real core /prompt, /queue, /history and /interrupt handlers.
 Start/End, graph extraction, scheduling, payload codecs and callbacks are production code.
 No managed ComfyUI service, user workflow, model, or remote worker is touched.
 """
 import asyncio
 from collections import Counter
+import copy
 import importlib
 import json
 import os
@@ -36,6 +37,7 @@ import aiohttp
 
 torch.set_num_threads(1)
 SOURCE_CALLS, BRANCH_CALLS, CAPTURED, MODEL_CALLS = [], [], [], []
+UTILITY_CALLS, ITEM_EXECUTIONS = [], []
 ALLOW_MASTER = False
 COUNT = 7
 
@@ -85,6 +87,51 @@ class TestCachedModel:
         return (object(),)
 
 
+def assert_item_execution(prompt, index):
+    """Prove the operation belongs to an item, not merely a participating host."""
+    from comfy_execution.utils import get_executing_context
+    context = get_executing_context()
+    assert context is not None
+    assert prompt['2']['inputs'].get('worker_item_token'), 'per-item node ran in outer graph'
+    running, _ = server.PromptServer.instance.prompt_queue.get_current_queue_volatile()
+    assert len(running) == 1, running
+    entry = running[0]
+    if WORKER:
+        assert entry[1] == context.prompt_id and entry[2] == prompt
+        assert entry[4] == ['4'], entry[4]
+    else:
+        assert ALLOW_MASTER, 'per-item node ran on delegate-only master'
+        assert entry[1] != context.prompt_id, 'per-item node ran in outer executor'
+        assert set(entry[4]) == {'4', '5'}, entry[4]
+    return {'index': index, 'prompt_id': context.prompt_id,
+            'role': 'worker' if WORKER else 'master-item', 'queue_prompt_id': entry[1]}
+
+
+class TestRequiredOutputUtility:
+    @classmethod
+    def INPUT_TYPES(cls):
+        return {'required': {'image': ('IMAGE',), 'index': ('INT', {'forceInput': True})},
+                'hidden': {'prompt': 'PROMPT'}}
+    OUTPUT_NODE = True
+    RETURN_TYPES = ('IMAGE',)
+    FUNCTION = 'run'
+    def run(self, image, index, prompt):
+        assert_item_execution(prompt, index)
+        UTILITY_CALLS.append(index)
+        return (image,)
+
+
+class TestDiscardedPreview:
+    @classmethod
+    def INPUT_TYPES(cls):
+        return {'required': {'image': ('IMAGE',)}}
+    OUTPUT_NODE = True
+    RETURN_TYPES = ()
+    FUNCTION = 'run'
+    def run(self, image):
+        raise AssertionError('discarded per-item output preview executed')
+
+
 class TestImageOperation:
     @classmethod
     def INPUT_TYPES(cls):
@@ -92,11 +139,13 @@ class TestImageOperation:
                              'text': ('STRING', {'forceInput': True}),
                              'strength': ('FLOAT', {'forceInput': True}),
                              'seed': ('INT', {'forceInput': True}), 'index': ('INT', {'forceInput': True}),
-                             'flag': ('BOOLEAN', {'forceInput': True}), 'model': ('MODEL',)}}
+                             'flag': ('BOOLEAN', {'forceInput': True}), 'model': ('MODEL',)},
+                'hidden': {'prompt': 'PROMPT'}}
     RETURN_TYPES = ('IMAGE', 'MASK', 'STRING')
     FUNCTION = 'run'
-    def run(self, image, reference_image, mask, text, strength, seed, index, flag, model):
+    def run(self, image, reference_image, mask, text, strength, seed, index, flag, model, prompt):
         assert WORKER or ALLOW_MASTER, 'per-image branch executed in orchestrator-only mode'
+        ITEM_EXECUTIONS.append(assert_item_execution(prompt, index))
         BRANCH_CALLS.append(index)
         # Sleep in short chunks so the real Comfy interrupt signal can be tested.
         until = time.monotonic() + DELAY
@@ -120,6 +169,7 @@ class TestCapture:
     RETURN_TYPES = ()
     FUNCTION = 'run'
     def run(self, images, masks, texts):
+        assert not WORKER, 'post-End capture leaked into a worker prompt'
         CAPTURED.append((images, masks, texts))
         return ()
 
@@ -131,7 +181,9 @@ async def setup():
     module = sys.modules[str(ROOT).replace('.', '_x_')]
     nodes.NODE_CLASS_MAPPINGS.update(TestImageSource=TestImageSource,
                                    TestImageOperation=TestImageOperation, TestCapture=TestCapture,
-                                   TestCachedModel=TestCachedModel, TestSharedFlag=TestSharedFlag)
+                                   TestCachedModel=TestCachedModel, TestSharedFlag=TestSharedFlag,
+                                   TestRequiredOutputUtility=TestRequiredOutputUtility,
+                                   TestDiscardedPreview=TestDiscardedPreview)
     # Match the CORS-enabled direct-worker launch contract for canvas probes.
     app = web.Application(middlewares=[server.create_cors_middleware('*')])
     # Register the actual feature routes, not test replacements.
@@ -147,61 +199,56 @@ async def listen(app):
     return runner, site._server.sockets[0].getsockname()[1]
 
 
+def finish_queue_item(ps, item_id, executor):
+    ps.prompt_queue.task_done(item_id, executor.history_result,
+        status=execution.PromptQueue.ExecutionStatus(
+            status_str='success' if executor.success else 'error',
+            completed=executor.success, messages=executor.status_messages),
+        process_item=lambda item: item[:5] + item[6:])
+
+
 async def worker_main():
     ps, assets, module, app = await setup()
-    pending, running, history = [], {}, {}
     wake = asyncio.Event()
-    interrupted_ids = []
+    interrupted_ids, submissions = [], []
+    core_handlers = {(route.method, route.path): route.handler for route in ps.routes}
     async def prompt(request):
         data = await request.json()
-        prompt_id = str(uuid.uuid4())
-        graph = data['prompt']
-        valid = await execution.validate_prompt(prompt_id, graph, None)
-        if not valid[0]:
-            return web.json_response({'error': valid[1], 'node_errors': valid[3]}, status=400)
-        pending.append((prompt_id, graph, valid[2]))
-        wake.set()
-        return web.json_response({'prompt_id': prompt_id})
-    async def queue(request):
-        if request.method == 'POST':
-            deleted = (await request.json()).get('delete', [])
-            pending[:] = [entry for entry in pending if entry[0] not in deleted]
-            return web.json_response({})
-        return web.json_response({'queue_running': [[0, key] for key in running],
-                                  'queue_pending': [[0, entry[0]] for entry in pending]})
-    async def get_history(request):
-        pid = request.match_info['prompt_id']
-        return web.json_response({pid: history[pid]} if pid in history else {})
+        # Use core's actual validation/enqueue/response schema, including
+        # partial_execution_targets; only observe it and wake our CPU consumer.
+        response = await core_handlers['POST', '/prompt'](request)
+        if response.status == 200:
+            accepted = json.loads(response.body)
+            _, pending = ps.prompt_queue.get_current_queue_volatile()
+            entry = next(item for item in pending if item[1] == accepted['prompt_id'])
+            submissions.append({'prompt_id': entry[1], 'requested': data.get('partial_execution_targets'),
+                                'queued': entry[4], 'node_ids': sorted(entry[2]),
+                                'end_ids': [key for key, node in entry[2].items()
+                                            if node['class_type'] == 'DistributedImageListEnd']})
+            wake.set()
+        return response
     async def interrupt(request):
-        pid = (await request.json())['prompt_id']
-        interrupted_ids.append(pid)
-        if pid in running:
-            comfy.model_management.interrupt_current_processing(True)
-        return web.json_response({})
+        interrupted_ids.append((await request.json())['prompt_id'])
+        return await core_handlers['POST', '/interrupt'](request)
     async def stats(request):
-        return web.json_response({'calls': BRANCH_CALLS, 'running': list(running),
-                                  'pending': [e[0] for e in pending], 'interrupted': interrupted_ids,
-                                  'history': history})
+        running, pending = ps.prompt_queue.get_current_queue_volatile()
+        return web.json_response({'calls': BRANCH_CALLS, 'utility_calls': UTILITY_CALLS,
+                                  'item_executions': ITEM_EXECUTIONS, 'submissions': submissions,
+                                  'running': [e[1] for e in running], 'pending': [e[1] for e in pending],
+                                  'interrupted': interrupted_ids, 'history': ps.prompt_queue.get_history()})
     async def consume():
         executor = execution.PromptExecutor(ps, cache_args={'ram': 0, 'ram_inactive': 0}, asset_manager=assets)
         while True:
             await wake.wait()
-            while pending:
-                pid, graph, outputs = pending.pop(0)
-                running[pid] = True
-                await asyncio.to_thread(executor.execute, graph, pid, {}, outputs)
-                history[pid] = {'status': {'completed': True,
-                    'status_str': 'success' if executor.success else 'error',
-                    'messages': str(executor.status_messages)}}
-                running.pop(pid, None)
+            while (queued := ps.prompt_queue.get(timeout=0)) is not None:
+                entry, item_id = queued
+                await asyncio.to_thread(executor.execute, entry[2], entry[1], entry[3], entry[4])
+                finish_queue_item(ps, item_id, executor)
             wake.clear()
-    async def probe(_request):
-        return web.json_response({'exec_info': {'queue_remaining': len(pending) + len(running)}})
-    app.router.add_get('/prompt', probe)
+    app.add_routes([route for route in ps.routes if
+                    (route.path in ('/queue', '/history/{prompt_id}') or
+                     (route.path == '/prompt' and route.method == 'GET'))])
     app.router.add_post('/prompt', prompt)
-    app.router.add_get('/queue', queue)
-    app.router.add_post('/queue', queue)
-    app.router.add_get('/history/{prompt_id}', get_history)
     app.router.add_post('/interrupt', interrupt)
     app.router.add_get('/test-stats', stats)
     runner, port = await listen(app)
@@ -228,12 +275,14 @@ def graph(callback, worker_ids):
         '3': {'class_type': 'TestImageOperation', 'inputs': {
             key: ['2', index] for key, index in [('image', 0), ('mask', 1), ('index', 2),
                 ('reference_image', 4), ('text', 5), ('strength', 6), ('seed', 7), ('flag', 8)]}
-              | {'model': ['6', 0]}},
+              | {'image': ['8', 0], 'model': ['6', 0]}},
         '4': {'class_type': 'DistributedImageListEnd', 'inputs': {
             'images': ['3', 0], 'masks': ['3', 1], 'texts': ['3', 2], 'context': ['2', 3]}},
         '5': {'class_type': 'TestCapture', 'inputs': {'images': ['4', 0], 'masks': ['4', 1], 'texts': ['4', 2]}},
         '6': {'class_type': 'TestCachedModel', 'inputs': {}},
         '7': {'class_type': 'TestSharedFlag', 'inputs': {}},
+        '8': {'class_type': 'TestRequiredOutputUtility', 'inputs': {'image': ['2', 0], 'index': ['2', 2]}},
+        '9': {'class_type': 'TestDiscardedPreview', 'inputs': {'image': ['8', 0]}},
     }
 
 
@@ -241,6 +290,8 @@ async def coordinator():
     global ALLOW_MASTER
     ps, assets, module, app = await setup()
     runtime = importlib.import_module(module.__name__ + '.image_list.runtime')
+    execution_plan = importlib.import_module(module.__name__ + '.image_list.execution_plan')
+    lifecycle = importlib.import_module(module.__name__ + '.image_list.lifecycle')
     config_mod = importlib.import_module(module.__name__ + '.utils.config')
     orchestration = importlib.import_module(module.__name__ + '.api.queue_orchestration')
     runner, port = await listen(app)
@@ -294,16 +345,49 @@ async def coordinator():
             prompt = graph(f'http://127.0.0.1:{port}', [w['id'] for w in workers])
             valid = await execution.validate_prompt('network-acceptance', prompt, None)
             assert valid[0], valid
+            assert {'4', '5', '8', '9'} == set(valid[2]), valid
             executor = execution.PromptExecutor(ps, cache_args={'ram': 0, 'ram_inactive': 0}, asset_manager=assets)
             # Exercise production queue preparation; only master enqueue is redirected
             # to this isolated executor, not the managed server's live queue.
             prompt['2']['inputs'].pop('image_list_enabled_worker_ids')
             prompt['2']['inputs'].pop('image_list_master_callback')
+            saved_prompt = copy.deepcopy(prompt)
+            outer_executions = []
             async def enqueue(prepared, workflow, client_id, **kwargs):
-                checked = await execution.validate_prompt('network-acceptance', prepared, None)
+                targets = kwargs['partial_execution_targets']
+                planned, planned_targets = execution_plan.prepare_image_list_execution(prepared)
+                assert planned == prepared and targets == planned_targets
+                if '9' in saved_prompt and 'TestImageOperation' in {
+                        node['class_type'] for node in prepared.values()}:
+                    assert set(prepared) == set(saved_prompt) - {'9'}, prepared
+                    assert prepared['8'] == saved_prompt['8'], 'required utility changed'
+                    assert prepared['5'] == saved_prompt['5'], 'post-End output changed'
+                    assert set(targets) == {'4', '5'}, targets
+                    assert nodes.NODE_CLASS_MAPPINGS['TestRequiredOutputUtility'].OUTPUT_NODE is True
+                    assert nodes.NODE_CLASS_MAPPINGS['TestDiscardedPreview'].OUTPUT_NODE is True
+                pid = 'network-acceptance' if not outer_executions else str(uuid.uuid4())
+                checked = await execution.validate_prompt(pid, prepared, targets)
                 assert checked[0], checked
-                await asyncio.to_thread(executor.execute, prepared, 'network-acceptance', {}, checked[2])
-                return {'prompt_id': 'network-acceptance', 'number': 1, 'node_errors': {}}
+                assert not checked[3] and Counter(checked[2]) == Counter(targets), checked
+                assert not lifecycle.has_safe_execution_targets(ps, pid, prepared)
+                # Exactly the core queue tuple/put/get/task_done lifecycle. Start
+                # must prove its roots from this real running record, not a mock.
+                number = len(outer_executions)
+                ps.prompt_queue.put((number, pid, prepared, {'client_id': client_id}, checked[2], {}))
+                entry, item_id = ps.prompt_queue.get(timeout=0)
+                assert lifecycle.has_safe_execution_targets(ps, pid, prepared)
+                before_captures = len(CAPTURED)
+                await asyncio.to_thread(executor.execute, entry[2], entry[1], entry[3], entry[4])
+                finish_queue_item(ps, item_id, executor)
+                assert ps.prompt_queue.get_current_queue_volatile() == ([], [])
+                history = ps.prompt_queue.get_history(pid)[pid]
+                assert Counter(history['prompt'][4]) == Counter(targets)
+                assert not lifecycle.has_safe_execution_targets(ps, pid, prepared)
+                outer_executions.append({'prompt_id': pid, 'targets': targets,
+                                         'captures': len(CAPTURED) - before_captures})
+                if executor.success:
+                    assert len(CAPTURED) == before_captures + 1, 'post-End Capture must run exactly once'
+                return {'prompt_id': pid, 'number': number, 'node_errors': checked[3]}
             with patch.object(config_mod, 'load_config', return_value=config), \
                  patch.object(orchestration, 'load_config', return_value=config), \
                  patch.object(orchestration, 'queue_prompt_payload', enqueue), \
@@ -312,11 +396,12 @@ async def coordinator():
                     prompt, None, 'test', [w['id'] for w in workers]), 45)
                 assert queued[0] == 'network-acceptance' and queued[2] == 2, queued
                 assert not ps.distributed_pending_jobs, 'image-list created unused Collector queues'
+            assert prompt == saved_prompt, 'queue preparation changed the saved/source graph'
             if not executor.success:
                 for err in stderr_files:
                     err.flush(); err.seek(0); print(err.read(), file=sys.stderr)
             assert executor.success, executor.status_messages
-            assert len(SOURCE_CALLS) == 1 and not BRANCH_CALLS
+            assert len(SOURCE_CALLS) == 1 and not BRANCH_CALLS and not UTILITY_CALLS
             assert len(CAPTURED) == 1
             images, masks, texts = CAPTURED[0]
             assert len(images) == len(masks) == len(texts) == COUNT
@@ -332,6 +417,14 @@ async def coordinator():
                 counts = [len(row['calls']) for row in stats]
                 assert counts[0] > counts[1] > 0, counts
                 assert Counter(i for row in stats for i in row['calls']) == Counter(range(COUNT))
+                for row in stats:
+                    assert Counter(row['utility_calls']) == Counter(row['calls']), row
+                    assert len(row['submissions']) == len(row['calls'])
+                    assert len({event['prompt_id'] for event in row['item_executions']}) == len(row['calls'])
+                    for submission in row['submissions']:
+                        assert submission['requested'] == submission['queued'] == ['4'], submission
+                        assert '8' in submission['node_ids'] and '9' not in submission['node_ids']
+                        assert '5' not in submission['node_ids']
             assert not runtime._MASTER_JOBS
             # Three sequential regions on one worker, including optional outputs
             # left unconnected. Region N+1 must not race the prior worker End.
@@ -346,12 +439,14 @@ async def coordinator():
                 previous = end
             chained['8'] = {'class_type': 'TestCapture', 'inputs': {
                 'images': ['7', 0], 'masks': ['7', 1], 'texts': ['7', 2]}}
+            saved_chained = copy.deepcopy(chained)
             with patch.object(config_mod, 'load_config', return_value=config), \
                  patch.object(orchestration, 'load_config', return_value=config), \
                  patch.object(orchestration, 'queue_prompt_payload', enqueue), \
                  patch.object(config_mod, 'get_worker_timeout_seconds', return_value=15):
                 await asyncio.wait_for(orchestration.orchestrate_distributed_execution(
                     chained, None, 'test', ['slow']), 45)
+            assert chained == saved_chained
             assert executor.success, executor.status_messages
             assert len(CAPTURED) == 2 and len(CAPTURED[1][0]) == 2
             assert all(torch.equal(image, torch.zeros((1, 8, 8, 3))) for image in CAPTURED[1][0])
@@ -366,29 +461,43 @@ async def coordinator():
             for mode, selected in [('mixed', [w['id'] for w in workers]), ('master-only', [])]:
                 executor = execution.PromptExecutor(ps, cache_args={'ram': 0, 'ram_inactive': 0}, asset_manager=assets)
                 before_sources, before_local, before_models = len(SOURCE_CALLS), len(BRANCH_CALLS), len(MODEL_CALLS)
+                before_utilities, before_events = len(UTILITY_CALLS), len(ITEM_EXECUTIONS)
                 async with aiohttp.ClientSession() as session:
-                    before_worker_calls = []
+                    before_worker_stats = []
                     for worker in workers:
                         async with session.get(f"http://127.0.0.1:{worker['port']}/test-stats") as response:
-                            before_worker_calls.append(len((await response.json())['calls']))
+                            before_worker_stats.append(await response.json())
                 with patch.object(config_mod, 'load_config', return_value=config), \
                      patch.object(orchestration, 'load_config', return_value=config), \
                      patch.object(orchestration, 'queue_prompt_payload', enqueue), \
                      patch.object(config_mod, 'get_worker_timeout_seconds', return_value=15):
                     queued = await asyncio.wait_for(orchestration.orchestrate_distributed_execution(
                         prompt, None, 'test', selected), 45)
+                assert prompt == saved_prompt, (mode, 'saved/source graph changed')
                 assert executor.success, (mode, executor.status_messages)
                 assert queued[2] == len(selected)
                 assert len(SOURCE_CALLS) == before_sources + 1, 'source was reevaluated per local item'
                 local_calls = BRANCH_CALLS[before_local:]
+                assert Counter(UTILITY_CALLS[before_utilities:]) == Counter(local_calls), mode
+                assert len({event['prompt_id'] for event in ITEM_EXECUTIONS[before_events:]}) == len(local_calls)
                 assert local_calls, f'{mode}: master did not process any items'
                 assert len(MODEL_CALLS) == before_models + 1, 'local model dependency did not stay cached'
                 remote_calls = []
                 async with aiohttp.ClientSession() as session:
-                    for worker, offset in zip(workers, before_worker_calls):
+                    for worker, before_stats in zip(workers, before_worker_stats):
                         async with session.get(f"http://127.0.0.1:{worker['port']}/test-stats") as response:
                             current = await response.json()
-                        remote_calls.extend(current['calls'][offset:])
+                        calls = current['calls'][len(before_stats['calls']):]
+                        utilities = current['utility_calls'][len(before_stats['utility_calls']):]
+                        submissions = current['submissions'][len(before_stats['submissions']):]
+                        events = current['item_executions'][len(before_stats['item_executions']):]
+                        assert Counter(utilities) == Counter(calls), (mode, current)
+                        assert len(submissions) == len(calls) == len({event['prompt_id'] for event in events})
+                        for submission in submissions:
+                            assert submission['requested'] == submission['queued'] == ['4'], submission
+                            assert '8' in submission['node_ids'] and '9' not in submission['node_ids']
+                            assert '5' not in submission['node_ids']
+                        remote_calls.extend(calls)
                         assert not current['running'] and not current['pending'], current
                 if mode == 'mixed':
                     assert remote_calls, 'mixed execution failed to use direct workers'
@@ -404,8 +513,19 @@ async def coordinator():
                 assert not runtime._MASTER_JOBS and not runtime._WORKER_ITEMS
                 participation_counts[mode] = {'master': len(local_calls), 'workers': len(remote_calls)}
             print('IMAGE_LIST_MASTER_OK ' + json.dumps(participation_counts))
+            assert len(outer_executions) == 4 and all(row['captures'] == 1 for row in outer_executions)
+            print('IMAGE_LIST_TARGETS_OK ' + json.dumps({
+                'modes': ['delegate-only', 'mixed', 'master-only'],
+                'outer_targets': outer_executions[0]['targets'], 'worker_targets': ['4'],
+                'required_utility_once_per_item': True, 'preview_discarded': True,
+                'saved_prompt_unchanged': prompt == saved_prompt, 'capture_once_per_outer_prompt': True}))
             ALLOW_MASTER = False
             config['settings']['master_delegate_only'] = True
+            # Preserve the original direct-runtime cancellation/failure fixture:
+            # it has no outer queue and must not claim target-control evidence.
+            direct_prompt = copy.deepcopy(prompt)
+            del direct_prompt['8'], direct_prompt['9']
+            direct_prompt['3']['inputs']['image'] = ['2', 0]
             # Real Comfy user-interrupt flag while a real worker executor is active.
             slow = workers[1]
             async with aiohttp.ClientSession() as session:
@@ -420,7 +540,7 @@ async def coordinator():
                         await asyncio.sleep(0.02)
                     raise AssertionError('worker state transition timed out')
                 before = await until(lambda state: not state['running'] and not state['pending'])
-                context = runtime.create_master_job('2', '4', prompt, source_rows(), ['slow'],
+                context = runtime.create_master_job('2', '4', direct_prompt, source_rows(), ['slow'],
                                                   f'http://127.0.0.1:{port}')
                 with patch.object(config_mod, 'load_config', return_value=config), \
                      patch.object(config_mod, 'get_worker_timeout_seconds', return_value=15):
@@ -444,7 +564,7 @@ async def coordinator():
             # A worker exception before End must fail, not hang or return a partial list.
             bad_rows = source_rows()
             bad_rows[0]['seed'] = -1
-            context = runtime.create_master_job('2', '4', prompt, bad_rows, ['fast'],
+            context = runtime.create_master_job('2', '4', direct_prompt, bad_rows, ['fast'],
                                               f'http://127.0.0.1:{port}')
             with patch.object(config_mod, 'load_config', return_value=config), \
                  patch.object(config_mod, 'get_worker_timeout_seconds', return_value=15):

tests/test_async_helpers.py

@@ -3,6 +3,7 @@ import sys
 import types
 import unittest
 from pathlib import Path
+from unittest.mock import AsyncMock, patch
 
 
 class _PromptQueue:
@@ -87,5 +88,59 @@ class QueuePromptPayloadTests(unittest.IsolatedAsyncioTestCase):
         self.assertEqual(extra_data["extra_pnginfo"]["workflow"], {"id": "workflow-1"})
 
 
+class ExplicitTargetTests(unittest.IsolatedAsyncioTestCase):
+    def setUp(self):
+        self.module, self.server = _load_async_helpers_module()
+        self.prompt = {"9": {"class_type": "Output"}, "10": {"class_type": "Output"}}
+        self.validate = AsyncMock(return_value=(True, None, ["9"], {}))
+        self.module.execution.validate_prompt = self.validate
+
+    async def test_targets_survive_callback_omission_and_are_queued_exactly(self):
+        seen = []
+        def callback(payload):
+            seen.append(payload["partial_execution_targets"][:])
+            payload["partial_execution_targets"].append("10")
+            return {"prompt": payload["prompt"]}
+        self.server.trigger_on_prompt = callback
+        targets = ["9"]
+        await self.module.queue_prompt_payload(self.prompt, partial_execution_targets=targets)
+        self.assertEqual(seen, [["9"]])
+        self.assertEqual(targets, ["9"])
+        self.assertEqual(self.validate.call_args.args[2], ["9"])
+        self.assertEqual(self.server.prompt_queue.items[0][4], ["9"])
+
+    async def test_invalid_requests_never_reach_validation_or_queue(self):
+        for targets in ([], "9", [9], [True], ["9", "9"], ["missing"], {}):
+            with self.subTest(targets=targets):
+                with self.assertRaises(self.module.PromptValidationError):
+                    await self.module.queue_prompt_payload(self.prompt, partial_execution_targets=targets)
+        self.validate.assert_not_awaited()
+        self.assertEqual(self.server.prompt_queue.items, [])
+
+    async def test_removed_target_after_callback_is_rejected(self):
+        self.server.trigger_on_prompt = lambda payload: {"prompt": {"10": self.prompt["10"]}}
+        with self.assertRaises(self.module.PromptValidationError):
+            await self.module.queue_prompt_payload(self.prompt, partial_execution_targets=["9"])
+        self.validate.assert_not_awaited()
+        self.assertEqual(self.server.prompt_queue.items, [])
+
+    async def test_partial_subset_extra_roots_and_target_errors_fail_closed(self):
+        for roots, errors in ((["9"], {}), (["9", "10", "extra"], {}),
+                              (["9", "10"], {"9": {"errors": [{"message": "bad"}]}}),
+                              (["9", "10"], {"dep": {"dependent_outputs": ["10"], "errors": []}})):
+            with self.subTest(roots=roots, errors=errors):
+                self.validate.return_value = (True, None, roots, errors)
+                with self.assertRaises(self.module.PromptValidationError):
+                    await self.module.queue_prompt_payload(self.prompt, partial_execution_targets=["9", "10"])
+        self.assertEqual(self.server.prompt_queue.items, [])
+
+    async def test_none_preserves_legacy_payload_and_validation(self):
+        seen = []
+        self.server.trigger_on_prompt = lambda payload: seen.append(payload.copy()) or payload
+        await self.module.queue_prompt_payload(self.prompt)
+        self.assertNotIn("partial_execution_targets", seen[0])
+        self.assertIsNone(self.validate.call_args.args[2])
+
+
 if __name__ == "__main__":
     unittest.main()

tests/test_dispatch_selection.py

@@ -4,7 +4,7 @@ import sys
 import types
 import unittest
 from pathlib import Path
-from unittest.mock import patch
+from unittest.mock import AsyncMock, patch
 
 
 def _load_dispatch_module():
@@ -168,6 +168,7 @@ class _FakeSession:
 def _valid_capability():
     return {
         "supported": True,
+        "execution_targets": True,
         "protocol_version": 2,
         "node_ids": ["DistributedImageListStart", "DistributedImageListEnd"],
     }
@@ -177,6 +178,43 @@ def _valid_queue(running=None, pending=None):
     return {"queue_running": running or [], "queue_pending": pending or []}
 
 
+class ExplicitTargetTransportTests(unittest.IsolatedAsyncioTestCase):
+    async def test_http_preserves_targets_and_omits_default_field(self):
+        response = types.SimpleNamespace(raise_for_status=lambda: None)
+        class Response:
+            async def __aenter__(self): return response
+            async def __aexit__(self, *args): pass
+        calls = []
+        def post(url, **kwargs):
+            calls.append(kwargs['json'])
+            return Response()
+        session = types.SimpleNamespace(post=post)
+        with patch.object(dispatch, 'get_client_session', AsyncMock(return_value=session)):
+            await dispatch.dispatch_worker_prompt({'id': 'worker'}, {'9': {}}, None,
+                                                  partial_execution_targets=['9'])
+            await dispatch.dispatch_worker_prompt({'id': 'worker'}, {'9': {}}, None)
+        self.assertEqual(calls[0]['partial_execution_targets'], ['9'])
+        self.assertNotIn('partial_execution_targets', calls[1])
+
+    async def test_websocket_sends_targets_all_the_way_to_wire(self):
+        import json
+        sent = []
+        async def receive():
+            return types.SimpleNamespace(type=dispatch.aiohttp.WSMsgType.TEXT, data=json.dumps({
+                'type': 'dispatch_ack', 'request_id': sent[-1]['request_id'],
+                'ok': True, 'prompt_id': 'worker-prompt'}))
+        ws = types.SimpleNamespace(send_json=AsyncMock(side_effect=sent.append),
+                                   receive=receive, close=AsyncMock())
+        session = types.SimpleNamespace(ws_connect=AsyncMock(return_value=ws))
+        with patch.object(dispatch, 'get_client_session', AsyncMock(return_value=session)):
+            result = await dispatch.dispatch_worker_prompt({'id': 'worker'}, {'9': {}}, None,
+                use_websocket=True, return_prompt_id=True, partial_execution_targets=['9'])
+            await dispatch.dispatch_worker_prompt({'id': 'worker'}, {'9': {}}, None, use_websocket=True)
+        self.assertEqual(result, 'worker-prompt')
+        self.assertEqual(sent[0]['partial_execution_targets'], ['9'])
+        self.assertNotIn('partial_execution_targets', sent[1])
+
+
 class DispatchSelectionTests(unittest.IsolatedAsyncioTestCase):
     async def test_select_active_workers_filters_offline(self):
         workers = [
@@ -417,6 +455,23 @@ class DispatchSelectionTests(unittest.IsolatedAsyncioTestCase):
             with self.assertRaisesRegex(RuntimeError, "No selected feature-compatible direct worker is idle"):
                 await dispatch.preflight_image_list_workers(workers, require_idle=True)
 
+    async def test_image_list_preflight_requires_exact_execution_targets_support(self):
+        for value in (None, False, 1, "true", []):
+            with self.subTest(value=value):
+                capability = _valid_capability()
+                if value is None:
+                    capability.pop("execution_targets")
+                else:
+                    capability["execution_targets"] = value
+                session = _FakeSession({("worker", "capabilities"): (200, capability)})
+                async def fake_session():
+                    return session
+                with patch.object(dispatch, "get_client_session", side_effect=fake_session), patch.object(
+                    dispatch, "build_worker_url", side_effect=lambda worker, path: f"http://{worker['id']}{path}"
+                ):
+                    with self.assertRaisesRegex(RuntimeError, "execution_targets.*upgrade"):
+                        await dispatch.preflight_image_list_workers([{"id": "worker"}])
+
     async def test_image_list_preflight_fails_closed_for_malformed_capabilities(self):
         malformed = {"supported": True, "protocol_version": 2, "node_ids": None}
         session = _FakeSession({("worker", "capabilities"): (200, malformed)})

tests/test_image_list.py

@@ -202,8 +202,10 @@ class ImageListGraphBoundaryTests(unittest.TestCase):
                     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)
+                for explicit_targets in (False, True):
+                    with self.subTest(explicit_targets=explicit_targets):
+                        with self.assertRaisesRegex(ValueError, "bypasses Start.*2"):
+                            validate_image_list_regions(prompt, explicit_targets=explicit_targets)
 
     def test_shared_typed_loader_and_leaf_constant_remain_valid(self):
         import sys
@@ -247,19 +249,25 @@ class ImageListGraphBoundaryTests(unittest.TestCase):
     def test_rejects_missing_context_pair_nested_region_and_cross_boundary_branch(self):
         prompt = graph_one_region()
         prompt["5"]["inputs"]["context"] = ["2", 0]
-        with self.assertRaisesRegex(ValueError, "typed context output"):
-            validate_image_list_regions(prompt)
+        for explicit_targets in (False, True):
+            with self.subTest(explicit_targets=explicit_targets):
+                with self.assertRaisesRegex(ValueError, "typed context output"):
+                    validate_image_list_regions(prompt, explicit_targets=explicit_targets)
         nested = graph_one_region()
         nested["7"] = start_node(["4", 0])
         nested["8"] = {"class_type": "NestedOperation", "inputs": {"image": ["7", IMAGE_OUT]}}
         nested["5"]["inputs"]["images"] = ["8", 0]
         nested["9"] = {"class_type": END_CLASS, "inputs": {"images": ["8", 0], "context": ["7", CONTEXT_OUT]}}
-        with self.assertRaisesRegex(ValueError, "Nested/ambiguous"):
-            validate_image_list_regions(nested)
+        for explicit_targets in (False, True):
+            with self.subTest(explicit_targets=explicit_targets):
+                with self.assertRaisesRegex(ValueError, "Nested/ambiguous"):
+                    validate_image_list_regions(nested, explicit_targets=explicit_targets)
         outside = graph_one_region()
         outside["7"] = {"class_type": "PreviewImage", "inputs": {"images": ["3", IMAGE_OUT]}}
-        with self.assertRaisesRegex(ValueError, "cross-boundary"):
-            validate_image_list_regions(outside)
+        for explicit_targets in (False, True):
+            with self.subTest(explicit_targets=explicit_targets):
+                with self.assertRaisesRegex(ValueError, "cross-boundary"):
+                    validate_image_list_regions(outside, explicit_targets=explicit_targets)
 
     def test_rejects_duplicate_or_missing_results(self):
         prompt = graph_one_region()

tests/test_image_list_execution_plan.py

@@ -0,0 +1,245 @@
+"""CPU-only execution-copy planning; no ComfyUI imports or prompt execution."""
+import copy
+import sys
+import types
+
+import pytest
+
+from image_list.execution_plan import prepare_image_list_execution
+from image_list.runtime import (
+    CONTEXT_OUT,
+    END_CLASS,
+    IMAGE_OUT,
+    START_CLASS,
+    build_worker_prompt,
+    validate_image_list_regions,
+)
+
+
+class OutputNode:
+    OUTPUT_NODE = True
+
+
+class SchemaOutputNode:
+    OUTPUT_NODE = False  # V3's schema, not this fallback, is authoritative.
+
+    @classmethod
+    def GET_NODE_INFO_V1(cls):
+        return {"output_node": True, "output": ("IMAGE",)}
+
+
+class ModelLoader:
+    RETURN_TYPES = ("MODEL", "CLIP", "VAE")
+
+
+@pytest.fixture
+def registry(monkeypatch):
+    classes = {
+        END_CLASS: SchemaOutputNode,
+        "SaveImage": OutputNode,
+        "PreviewImage": OutputNode,
+        "OutputUtility": OutputNode,
+        "SchemaUtility": SchemaOutputNode,
+        "ModelLoader": ModelLoader,
+    }
+    monkeypatch.setitem(sys.modules, "nodes", types.SimpleNamespace(NODE_CLASS_MAPPINGS=classes))
+    return classes
+
+
+def node(class_type, **inputs):
+    return {"class_type": class_type, "inputs": inputs}
+
+
+def graph():
+    return {
+        "loader": node("ModelLoader"),
+        "source": node("ImageSource"),
+        "start": node(START_CLASS, images=["source", 0]),
+        "process": node("FrameOperation", image=["start", IMAGE_OUT], model=["loader", 0]),
+        "end": node(END_CLASS, images=["process", 0], context=["start", CONTEXT_OUT]),
+        "save": node("SaveImage", images=["end", 0]),
+    }
+
+
+def test_discards_multistep_previews_but_preserves_other_output_roots(registry):
+    prompt = graph()
+    prompt.update({
+        "debug-transform": node("DebugTransform", images=["start", IMAGE_OUT]),
+        "debug-preview": node("PreviewImage", images=["debug-transform", 0]),
+        "process-preview": node("PreviewImage", images=["process", 0]),
+        "unrelated-source": node("ImageSource"),
+        "unrelated-preview": node("PreviewImage", images=["unrelated-source", 0]),
+        "upstream-preview": node("PreviewImage", images=["source", 0]),
+        "post-transform": node("FrameOperation", images=["end", 0]),
+        "post-preview": node("PreviewImage", images=["post-transform", 0]),
+    })
+    before = copy.deepcopy(prompt)
+    with pytest.raises(ValueError, match="cross-boundary"):
+        validate_image_list_regions(prompt)
+    prepared, targets = prepare_image_list_execution(prompt)
+    assert set(prepared) == set(prompt) - {"debug-transform", "debug-preview", "process-preview"}
+    assert targets == ["end", "save", "unrelated-preview", "upstream-preview", "post-preview"]
+    validate_image_list_regions(prepared, explicit_targets=True)
+    assert prompt == before
+
+
+@pytest.mark.parametrize("class_type", ["OutputUtility", "SchemaUtility"])
+def test_required_output_utility_keeps_complete_dependency_chain(registry, class_type):
+    prompt = graph()
+    prompt.update({
+        "utility-config": node("Constant", value=7),
+        "utility-first": node(class_type, images=["process", 0], config=["utility-config", 0]),
+        "utility-second": node(class_type, images=["utility-first", 0]),
+        "result": node("FrameOperation", image=["utility-second", 0]),
+        "discarded-preview": node("PreviewImage", images=["utility-first", 0]),
+    })
+    prompt["end"]["inputs"]["images"] = ["result", 0]
+    before = copy.deepcopy(prompt)
+    registered = dict(registry)
+    prepared, targets = prepare_image_list_execution(prompt)
+    assert set(prepared) == set(prompt) - {"discarded-preview"}
+    assert targets == ["end", "save"]
+    with pytest.raises(ValueError, match="Output node"):
+        validate_image_list_regions(prepared)
+    region = validate_image_list_regions(prepared, explicit_targets=True)[0]
+    worker = build_worker_prompt(prepared, region, "token", "context", 0, "http://master.invalid")
+    assert set(worker) == set(prepared) - {"source", "save"}
+    assert worker["utility-second"]["inputs"] == {"images": ["utility-first", 0]}
+    assert prompt == before
+    assert registry == registered
+    assert OutputNode.OUTPUT_NODE is True
+    assert SchemaOutputNode.OUTPUT_NODE is False
+    assert SchemaOutputNode.GET_NODE_INFO_V1()["output_node"] is True
+
+
+@pytest.mark.parametrize("source", ["start", "process"])
+@pytest.mark.parametrize("direct", [False, True])
+def test_internal_bypass_into_post_end_computation_is_rejected(registry, source, direct):
+    prompt = graph()
+    prompt["post-process"] = node("Composite", images=["end", 0])
+    if not direct:
+        prompt["bypass"] = node("FrameOperation", image=[source, 0])
+        source = "bypass"
+    prompt["post-process"]["inputs"]["reference"] = [source, 0]
+    prompt["save"]["inputs"]["images"] = ["post-process", 0]
+    before = copy.deepcopy(prompt)
+    with pytest.raises(ValueError, match="cross-boundary"):
+        prepare_image_list_execution(prompt)
+    assert prompt == before
+
+
+def test_sequential_namespaced_regions_and_intermediate_output_are_preserved(registry):
+    prompt = graph()
+    prompt.update({
+        "next-start": node(START_CLASS, images=["end", 0]),
+        "next-process": node("SchemaUtility", images=["next-start", IMAGE_OUT], model=["loader", 0]),
+        "next-end": node(END_CLASS, images=["next-process", 0], context=["next-start", CONTEXT_OUT]),
+        "next-save": node("SaveImage", images=["next-end", 0]),
+        "first-debug": node("PreviewImage", images=["process", 0]),
+        "second-debug": node("PreviewImage", images=["next-process", 0]),
+    })
+    renamed = {key: f"group:7/{key}" for key in prompt}
+    for spec in prompt.values():
+        for name, value in spec["inputs"].items():
+            if isinstance(value, list):
+                spec["inputs"][name] = [renamed[value[0]], value[1]]
+    prompt = {renamed[key]: spec for key, spec in prompt.items()}
+    prepared, targets = prepare_image_list_execution(prompt)
+    assert set(prepared) == set(prompt) - {renamed["first-debug"], renamed["second-debug"]}
+    assert targets == [renamed[name] for name in ("end", "save", "next-end", "next-save")]
+    regions = validate_image_list_regions(prepared, explicit_targets=True)
+    assert {(r["start_id"], r["end_id"]) for r in regions} == {
+        (renamed["start"], renamed["end"]), (renamed["next-start"], renamed["next-end"]),
+    }
+    second = next(r for r in regions if r["start_id"] == renamed["next-start"])
+    worker = build_worker_prompt(prepared, second, "token", "context", 0, "http://master.invalid")
+    assert set(worker) == {renamed[name] for name in ("loader", "next-start", "next-process", "next-end")}
+
+
+def test_shared_loader_and_its_independent_output_root_are_preserved(registry):
+    registry["ModelLoader"] = type("OutputLoader", (ModelLoader, OutputNode), {})
+    prompt = graph()
+    prompt["source"]["inputs"]["model"] = ["loader", 0]
+    prompt["independent-save"] = node("SaveImage", model=["loader", 0])
+    prepared, targets = prepare_image_list_execution(prompt)
+    assert prepared == prompt
+    assert targets == ["loader", "end", "save", "independent-save"]
+    region = validate_image_list_regions(prepared, explicit_targets=True)[0]
+    assert "loader" in region["node_ids"]
+    assert "loader" not in region["branch_ids"]
+
+
+@pytest.mark.parametrize("unsafe_class", [None, type("ListLoader", (), {
+    "RETURN_TYPES": ("MODEL",), "OUTPUT_IS_LIST": (True,),
+})])
+def test_unknown_or_list_producing_shared_loader_still_fails_closed(registry, unsafe_class):
+    if unsafe_class is None:
+        registry.pop("ModelLoader")
+    else:
+        registry["ModelLoader"] = unsafe_class
+    prompt = graph()
+    prompt["source"]["inputs"]["model"] = ["loader", 0]
+    prompt["debug"] = node("PreviewImage", images=["process", 0])
+    with pytest.raises(ValueError, match="bypasses Start.*loader"):
+        prepare_image_list_execution(prompt)
+
+
+def test_upstream_per_item_bypass_is_not_hidden_by_preview_pruning(registry):
+    prompt = graph()
+    prompt["process"]["inputs"]["reference"] = ["source", 0]
+    prompt["debug"] = node("PreviewImage", images=["process", 0])
+    with pytest.raises(ValueError, match="bypasses Start.*source"):
+        prepare_image_list_execution(prompt)
+
+
+@pytest.mark.parametrize("nested", [False, True])
+def test_overlapping_or_nested_regions_cannot_be_discarded_as_side_branches(registry, nested):
+    prompt = graph()
+    prompt["other-start"] = node(START_CLASS, images=["process" if nested else "source", 0])
+    prompt["other-process"] = node("FrameOperation", image=["other-start", IMAGE_OUT], reference=["process", 0])
+    prompt["other-end"] = node(END_CLASS, images=["other-process", 0], context=["other-start", CONTEXT_OUT])
+    with pytest.raises(ValueError, match="Nested/ambiguous|cross-boundary|ambiguous processing"):
+        prepare_image_list_execution(prompt)
+
+
+@pytest.mark.parametrize("invalid", ["missing-end", "wrong-context", "missing-images", "unrelated-images"])
+def test_pairing_and_result_validation_cannot_be_pruned_away(registry, invalid):
+    prompt = graph()
+    if invalid == "missing-end":
+        del prompt["end"]
+    elif invalid == "wrong-context":
+        prompt["end"]["inputs"]["context"] = ["source", 0]
+    elif invalid == "missing-images":
+        del prompt["end"]["inputs"]["images"]
+    else:
+        prompt["end"]["inputs"]["images"] = ["source", 0]
+    before = copy.deepcopy(prompt)
+    with pytest.raises(ValueError):
+        prepare_image_list_execution(prompt)
+    assert prompt == before
+
+
+def test_execution_copy_is_deep_and_repeated_preparation_is_stable(registry):
+    prompt = graph()
+    prompt["process"]["_meta"] = {"title": "Original", "tags": ["saved"]}
+    before = copy.deepcopy(prompt)
+    prepared, targets = prepare_image_list_execution(prompt)
+    prepared_again, targets_again = prepare_image_list_execution(prepared)
+    assert prepared_again == prepared
+    assert targets_again == targets
+    prepared["start"]["inputs"]["images"][0] = "changed"
+    prepared["process"]["_meta"]["tags"].append("execution-only")
+    assert prompt == before
+    assert prepared_again == before
+
+
+def test_v3_false_output_flag_overrides_legacy_true(registry):
+    registry["NotOutput"] = type("NotOutput", (), {
+        "OUTPUT_NODE": True,
+        "GET_NODE_INFO_V1": classmethod(lambda cls: {"output_node": False}),
+    })
+    prompt = graph()
+    prompt["not-output"] = node("NotOutput", images=["end", 0])
+    prepared, targets = prepare_image_list_execution(prompt)
+    assert prepared == prompt
+    assert targets == ["end", "save"]

tests/test_image_list_integration_gaps.py

@@ -35,6 +35,15 @@ def load_routes():
     return module, routes
 
 
+class CapabilityTests(unittest.IsolatedAsyncioTestCase):
+    async def test_capability_advertises_target_control_without_tensor_protocol_bump(self):
+        module, _routes = load_routes()
+        response = await module.image_list_capabilities_endpoint(None)
+        payload = json.loads(response.text)
+        self.assertIs(payload['execution_targets'], True)
+        self.assertEqual(payload['protocol_version'], 2)
+
+
 class TransportTests(unittest.IsolatedAsyncioTestCase):
     async def asyncSetUp(self):
         self.module, routes = load_routes()

tests/test_image_list_lifecycle.py

@@ -11,6 +11,83 @@ from aiohttp import web
 from image_list import lifecycle, runtime
 
 
+class ExecutionTargetProofTests(unittest.TestCase):
+    def setUp(self):
+        from test_image_list_master_runtime import graph
+        self.prompt = graph()
+        self.registry = SimpleNamespace(NODE_CLASS_MAPPINGS={
+            'Operation': type('Flagged', (), {'OUTPUT_NODE': True})})
+        self.modules = patch.dict('sys.modules', {'nodes': self.registry})
+        self.modules.start()
+        self.addCleanup(self.modules.stop)
+
+    def server(self, roots, *, prompt=None, running=True):
+        item = (0, 'owned', self.prompt if prompt is None else prompt, {}, roots, {})
+        return SimpleNamespace(prompt_queue=SimpleNamespace(
+            get_current_queue_volatile=lambda: ([item], []) if running else ([], [item])))
+
+    def test_outer_roots_in_running_queue_are_authoritative(self):
+        self.assertTrue(lifecycle.has_safe_execution_targets(self.server(['end']), 'owned', self.prompt))
+        for roots in (['op', 'end'], [], ['end', 'end'], None, [True]):
+            with self.subTest(roots=roots):
+                self.assertFalse(lifecycle.has_safe_execution_targets(self.server(roots), 'owned', self.prompt))
+        self.assertFalse(lifecycle.has_safe_execution_targets(self.server(['end']), 'other', self.prompt))
+        self.assertFalse(lifecycle.has_safe_execution_targets(self.server(['end'], running=False), 'owned', self.prompt))
+        self.assertFalse(lifecycle.has_safe_execution_targets(self.server(['end'], prompt={}), 'owned', self.prompt))
+
+    def test_prompt_marker_is_not_an_execution_guarantee(self):
+        self.prompt['start']['inputs']['image_list_explicit_targets'] = True
+        self.assertFalse(lifecycle.has_safe_execution_targets(self.server(['op', 'end']), 'owned', self.prompt))
+        with self.assertRaisesRegex(ValueError, 'must be placed after'):
+            runtime.create_master_job('start', 'end', self.prompt, [], [], '', master_participates=True)
+        self.assertEqual(runtime._MASTER_JOBS, {})
+
+    def test_start_node_uses_running_queue_proof(self):
+        import importlib.util
+        import sys
+        from pathlib import Path
+        from types import ModuleType
+        import torch
+
+        root = Path(__file__).resolve().parents[1]
+        package = '_image_list_start_targets'
+        modules = {}
+        for suffix in ('', '.nodes', '.image_list', '.utils'):
+            module = ModuleType(package + suffix)
+            module.__path__ = [str(root / suffix.lstrip('.'))]
+            modules[package + suffix] = module
+        modules[package + '.image_list.runtime'] = runtime
+        modules[package + '.image_list.lifecycle'] = lifecycle
+        modules[package + '.utils.async_helpers'] = SimpleNamespace(run_async_in_server_loop=None)
+        io = SimpleNamespace(ComfyNode=type('ComfyNode', (), {}), NodeOutput=lambda *args: args,
+                             **{name: object() for name in ('Image', 'Mask', 'String', 'Int', 'Float', 'Boolean')})
+        modules['comfy_api.v0_0_2'] = SimpleNamespace(io=io)
+        modules['comfy_execution.utils'] = SimpleNamespace(
+            get_executing_context=lambda: SimpleNamespace(prompt_id='owned'))
+        server = SimpleNamespace(PromptServer=SimpleNamespace(instance=self.server(['end'])))
+        modules['server'] = server
+        spec = importlib.util.spec_from_file_location(package + '.nodes.image_list', root / 'nodes/image_list.py')
+        node_module = importlib.util.module_from_spec(spec)
+        with patch.dict(sys.modules, modules):
+            spec.loader.exec_module(node_module)
+            start = node_module.DistributedImageListStart
+            start.hidden = SimpleNamespace(prompt=self.prompt, unique_id='start')
+            result = start.execute(images=torch.zeros((1, 2, 2, 3)), image_list_master_participates=True)
+            context = result[3][0]
+            self.assertIsNotNone(runtime.get_master_job(context))
+            runtime.remove_master_job(context)
+            server.PromptServer.instance = self.server(['op', 'end'])
+            with self.assertRaisesRegex(ValueError, 'must be placed after'):
+                start.execute(images=torch.zeros((1, 2, 2, 3)), image_list_master_participates=True)
+        self.assertEqual(runtime._MASTER_JOBS, {})
+
+    def test_owned_target_controlled_start_can_create_job(self):
+        context = runtime.create_master_job('start', 'end', self.prompt, [], [], '',
+                                            master_participates=True, explicit_targets=True)
+        self.addCleanup(runtime.remove_master_job, context)
+        self.assertIsNotNone(runtime.get_master_job(context))
+
+
 class ImageListLifecycleTests(unittest.IsolatedAsyncioTestCase):
     def setUp(self):
         with runtime._LOCK:

tests/test_image_list_master_selection.py

@@ -34,6 +34,23 @@ class MasterSelectionTests(unittest.IsolatedAsyncioTestCase):
     def prepared_inputs(self):
         return self.q.queue_prompt_payload.call_args.args[0]["2"]["inputs"]
 
+    async def test_prepared_master_prunes_side_outputs_without_mutating_input(self):
+        import copy
+        self.prompt['op'] = {'class_type': 'Flagged', 'inputs': {'image': ['2', 0]}}
+        self.prompt['3']['inputs']['images'] = ['op', 0]
+        self.prompt['preview'] = {'class_type': 'Flagged', 'inputs': {'image': ['op', 0]}}
+        self.prompt['save'] = {'class_type': 'Flagged', 'inputs': {'image': ['3', 0]}}
+        original = copy.deepcopy(self.prompt)
+        registry = type('Registry', (), {'NODE_CLASS_MAPPINGS': {
+            'Flagged': type('Flagged', (), {'OUTPUT_NODE': True})}})
+        with patch.dict(sys.modules, {'nodes': registry}):
+            await self.queue()
+        queued = self.q.queue_prompt_payload.call_args
+        self.assertEqual(queued.kwargs['partial_execution_targets'], ['3', 'save'])
+        self.assertIn('op', queued.args[0])
+        self.assertNotIn('preview', queued.args[0])
+        self.assertEqual(self.prompt, original)
+
     async def test_default_includes_master_and_direct_workers(self):
         result = await self.queue()
         self.assertIs(self.prepared_inputs()["image_list_master_participates"], True)

tests/test_image_list_runtime_boundaries.py

@@ -84,7 +84,7 @@ class RuntimeBoundaryTests(unittest.IsolatedAsyncioTestCase):
         # Other legacy test modules replace aiohttp during collection. This
         # controlled provider fixture needs only its explicit timeout/error seam.
         sys.modules['aiohttp'] = types.SimpleNamespace(
-            ClientTimeout=lambda **kwargs: kwargs, ClientError=ConnectionError)
+            ClientTimeout=lambda **kwargs: kwargs, ClientError=ConnectionError, ClientConnectionError=ConnectionError)
         comfy = types.ModuleType('comfy')
         comfy.model_management = types.SimpleNamespace(processing_interrupted=lambda: self.interrupted)
         sys.modules['comfy'] = comfy
@@ -94,6 +94,7 @@ class RuntimeBoundaryTests(unittest.IsolatedAsyncioTestCase):
         prompt['7']['inputs']['image'] = 'shared/ref.png'
 
     async def dispatch_prompt(self, worker, prompt, meta, **kwargs):
+        self.assertEqual(kwargs.get('partial_execution_targets'), ['5'])
         if self.reject:
             raise RuntimeError('prompt rejected')
         self.prompts.append((json.loads(json.dumps(prompt)), meta))
@@ -136,6 +137,23 @@ class RuntimeBoundaryTests(unittest.IsolatedAsyncioTestCase):
             ['worker'], 'http://public-master', self.metadata)
         return self.context
 
+    async def test_runtime_preflight_rejects_old_target_capability_before_upload(self):
+        from test_dispatch_selection import _load_dispatch_module, _FakeSession, _valid_capability
+        real_dispatch = _load_dispatch_module()
+        capability = _valid_capability()
+        capability.pop('execution_targets')
+        session = _FakeSession({('worker', 'capabilities'): (200, capability)})
+        module = sys.modules['_image_list_boundaries.api.orchestration.dispatch']
+        module.preflight_image_list_workers = real_dispatch.preflight_image_list_workers
+        with patch.object(real_dispatch, 'get_client_session', AsyncMock(return_value=session)), patch.object(
+            real_dispatch, 'build_worker_url', side_effect=lambda worker, path: f"http://{worker['id']}{path}"
+        ):
+            with self.assertRaisesRegex(RuntimeError, 'execution_targets.*upgrade'):
+                await self.rt.run_image_list_job(self.create(1))
+        self.assertFalse(self.uploads)
+        self.dispatch.assert_not_awaited()
+        self.assertFalse(self.rt._MASTER_JOBS)
+
     async def test_each_assignment_reuses_media_paths_with_fresh_tokens_and_metadata(self):
         context = self.create()
         await self.rt.run_image_list_job(context)

tests/test_v3_migration.py

@@ -1,4 +1,5 @@
 """Real-framework acceptance; opt in with COMFYUI_SOURCE_ROOT."""
+import json
 import os
 from pathlib import Path
 import subprocess
@@ -35,3 +36,15 @@ def test_image_list_network_runtime():
     assert 'IMAGE_LIST_NETWORK_OK' in result.stdout
     assert 'IMAGE_LIST_CANCEL_OK' in result.stdout
     assert 'IMAGE_LIST_MASTER_OK' in result.stdout
+    assert 'IMAGE_LIST_PREFLIGHT_OK' in result.stdout
+    assert 'IMAGE_LIST_SEQUENTIAL_OK' in result.stdout
+    assert 'IMAGE_LIST_FAILURE_OK' in result.stdout
+    reports = [json.loads(line.removeprefix('IMAGE_LIST_TARGETS_OK '))
+               for line in result.stdout.splitlines() if line.startswith('IMAGE_LIST_TARGETS_OK ')]
+    assert len(reports) == 1, result.stdout
+    assert reports[0] == {
+        'modes': ['delegate-only', 'mixed', 'master-only'],
+        'outer_targets': ['4', '5'], 'worker_targets': ['4'],
+        'required_utility_once_per_item': True, 'preview_discarded': True,
+        'saved_prompt_unchanged': True, 'capture_once_per_outer_prompt': True,
+    }