fix: address review-bot findings on merged PRs (#593)

- parallel: the error path waits only on futures it could not cancel, so a
  pool shutdown that cancelled queued items (or ExitRequested in the inline
  fallback) no longer hangs exit (#582)
- send: Send to All keeps the node-identity part of the revision after the
  first write and relaxes only the graph version, so a workflow loaded
  between sends is refused (#578)
- send: every node of Send to All takes a fresh ticket, so a window with
  more than 100 nodes no longer refuses node 1 (#578)
- quality: close test file handles, drop unused locals, replace the
  looked-up flag with a sentinel, single unittest.mock import, document the
  BaseException re-raise and the startup banner print

__init__.py

@@ -448,8 +448,11 @@ except Exception as e:
 
 __all__ = ["NODE_CLASS_MAPPINGS", "NODE_DISPLAY_NAME_MAPPINGS", "WEB_DIRECTORY"]
 
-# One startup line. ASCII only: a Windows console using a legacy code page
-# cannot print emoji, and a UnicodeEncodeError here would abort loading the node.
+# One startup line, printed at import on purpose: it is the package's banner on
+# stdout, as ComfyUI's other custom nodes print theirs. Not a log record: the
+# console log handler hides INFO by default, so a logger would drop it. ASCII
+# only: a Windows console using a legacy code page cannot print emoji, and a
+# UnicodeEncodeError here would abort loading the node.
 print(
     f"\033[94m[ComfyUI-PromptManager] Version:\033[0m {get_version()}"
     f" - {len(NODE_CLASS_MAPPINGS)} tools loaded"

tests/js/prompt-sender.test.js

@@ -255,7 +255,8 @@ test('All is offered only with two or more eligible nodes and is never selected
 test('Apply to All previews every node first, then sends one after another in node id order',async()=>{
     const {createSendSession,ALL_NODES}=await source;const s=await allSession(createSendSession);
     s.session.selectTarget(ALL_NODES);await s.session.apply();
-    assert.deepEqual(order(s.calls),['preview 2','preview 9','preview 10','apply 2','preview 9','apply 9','preview 10','apply 10']);
+    // Each node takes a fresh ticket, the first too: the host keeps only its latest 100 (#593).
+    assert.deepEqual(order(s.calls),['preview 2','preview 9','preview 10','preview 2','apply 2','preview 9','apply 9','preview 10','apply 10']);
     assert.equal(s.session.state.phase,'success');assert.equal(s.session.state.message,'Sent to 3 of 3 nodes in Flow - ComfyUI.');
     await s.session.apply();assert.equal(s.calls.filter(call=>call[0]==='apply').length,3);s.session.dispose();
 });
@@ -363,3 +364,48 @@ test('a changed prepend or append after the review refuses that node',async()=>{
     assert.deepEqual(s.calls.filter(call=>call[0]==='apply').map(call=>call[1]),['2','10']);
     assert.match(s.session.state.message,/^Sent to 2 of 3 nodes in Flow - ComfyUI\. #9 Node 9: Its prompt changed since the preview\./);s.session.dispose();
 });
+
+// #593: Send to All through the real Toolkit bridge and the real adapter, as the review bot reproduced it.
+function realWindow(count) {
+    const listeners=new Set();
+    const channelFactory=()=>{const own=new Set();return{
+        addEventListener(type,listener){if(type==='message'){own.add(listener);listeners.add(listener);}},removeEventListener(type,listener){listeners.delete(listener);},
+        postMessage(data){for(const listener of [...listeners])if(!own.has(listener))queueMicrotask(()=>listener({data:structuredClone(data),origin:'https://comfy.example'}));},close(){},
+    };};
+    // Like LiteGraph, a widget write moves the version of the graph it is in.
+    const graph=()=>{const made={_version:1,setDirtyCanvas(){made._version++;}};made._nodes=Array.from({length:count},(_,index)=>{
+        const widgets=['text','prepend_text','append_text'].map(name=>({name,value:name==='text'?'old':'',inputEl:{value:name==='text'?'old':''}}));
+        return {id:index+1,type:'PromptManager',title:`Node ${index+1}`,widgets,widgets_values:widgets.map(widget=>widget.value),inputs:[],setDirtyCanvas(){}};
+    });return made;};
+    return {channelFactory,graph};
+}
+async function realSend(count,{afterFirstApply}={}) {
+    const {createSendSession,ALL_NODES}=await source;
+    const {createBridgeHost,createBridgeClient}=await import('../../web/js/shared/toolkit/prompt-bridge.js');
+    const {createPromptAdapter}=await import('../../web/js/prompt-bridge-adapter.js');
+    const win=realWindow(count);const app={graph:win.graph(),canvas:{}};
+    const adapter=createPromptAdapter(app,{runtime:{}});let applied=0;
+    const watched={...adapter,apply(snapshot){const result=adapter.apply(snapshot);if(++applied===1)afterFirstApply?.(app,win);return result;}};
+    const options={namespace:'/prompt_manager/',origin:'https://comfy.example',channelFactory:win.channelFactory};
+    const host=createBridgeHost({...options,label:'Flow - ComfyUI',adapter:watched});
+    const client=createBridgeClient(options);
+    const session=createSendSession({client,roles:{positive:'new'},onChange:()=>{}});
+    try {
+        await session.discover();session.selectTarget(ALL_NODES);await session.apply();
+        return {state:session.state,app};
+    } finally { session.dispose();host.dispose(); }
+}
+test('All refuses the later nodes when another workflow is loaded after the first send (#593)',async()=>{
+    let first;
+    const {state,app}=await realSend(3,{afterFirstApply:(target,win)=>{first=target.graph;target.graph=win.graph();target.graph._version=first._version;}});
+    // The new workflow has the same ids, text and composition, and even the same graph version: only node identity differs.
+    assert.equal(first._nodes[0].widgets[0].value,'new');
+    assert.deepEqual(app.graph._nodes.map(node=>node.widgets[0].value),['old','old','old']);
+    assert.equal(state.phase,'error');
+    assert.match(state.message,/^Sent to 1 of 3 nodes in Flow - ComfyUI\. #2 Node 2: The workflow changed since this window was found\./);
+});
+test('All sends to every node of a window with more than 100 of them (#593)',async()=>{
+    const {state,app}=await realSend(101);
+    assert.equal(state.message,'Sent to 101 of 101 nodes in Flow - ComfyUI.');
+    assert.ok(app.graph._nodes.every(node=>node.widgets[0].value==='new'));
+});

tests/test_maintenance_pair_metadata.py

@@ -171,7 +171,7 @@ class TestClassification(PairMetadataCase):
         self.assertEqual(self.preview().targets, [[prompt, True, True]])
 
     def test_category_is_case_sensitive(self):
-        _a, prompt = self.pair("case", "n1", category="Portrait")
+        self.pair("case", "n1", category="Portrait")
         self.pair("case", "n2", category="portrait")
         draft = self.preview()
         self.assertEqual(draft.targets, [])
@@ -292,7 +292,7 @@ class TestClassification(PairMetadataCase):
         )
 
     def test_protected_rows_come_first_within_fifty(self):
-        _c, conflict = self.pair("zz conflict", "n1", category="x")
+        self.pair("zz conflict", "n1", category="x")
         self.pair("zz conflict", "n2", category="y")
         for n in range(60):
             self.pair(f"fill {n:02d}", category="portrait")

tests/test_native_memory.py

@@ -2,7 +2,7 @@
 
 import sys
 import unittest
-from unittest import mock
+import unittest.mock as mock
 
 from utils import native_memory
 
@@ -10,7 +10,9 @@ from utils import native_memory
 class ReleaseFreeMemoryTests(unittest.TestCase):
     def setUp(self):
         # Each test looks the symbol up again.
-        patcher = mock.patch.multiple(native_memory, _trim=None, _looked_up=False)
+        patcher = mock.patch.object(
+            native_memory, "_trim", native_memory._NOT_LOOKED_UP
+        )
         patcher.start()
         self.addCleanup(patcher.stop)
 
@@ -22,7 +24,7 @@ class ReleaseFreeMemoryTests(unittest.TestCase):
     def test_other_platforms_do_nothing(self):
         for platform in ("win32", "darwin", "cygwin"):
             with self.subTest(platform), mock.patch.object(sys, "platform", platform):
-                native_memory._looked_up = False
+                native_memory._trim = native_memory._NOT_LOOKED_UP
                 with mock.patch.object(native_memory.ctypes, "CDLL") as cdll:
                     self.assertFalse(native_memory.release_free_memory())
                 cdll.assert_not_called()

tests/test_saved_instructions.py

@@ -153,7 +153,7 @@ class ConfigTestCase(unittest.TestCase):
             path = os.path.join(tmpdir, "config.json")
             with open(path, "w", encoding="utf-8") as handle:
                 json.dump({"autotag": {"defaults": defaults}}, handle)
-            before = os.stat(path).st_mtime_ns, open(path, "rb").read()
+            before = os.stat(path).st_mtime_ns, self.read_bytes(path)
             if warnings:
                 with self.assertLogs("prompt_manager.config", level="WARNING") as logs:
                     PromptManagerConfig.load_from_file(path)
@@ -162,9 +162,13 @@ class ConfigTestCase(unittest.TestCase):
             else:
                 PromptManagerConfig.load_from_file(path)
             # Nothing is written at load.
-            self.assertEqual(
-                (os.stat(path).st_mtime_ns, open(path, "rb").read()), before
-            )
+            self.assertEqual((os.stat(path).st_mtime_ns, self.read_bytes(path)), before)
+
+    @staticmethod
+    def read_bytes(path):
+        # Closed at once: Windows cannot delete the temp dir with it open.
+        with open(path, "rb") as handle:
+            return handle.read()
 
     def state(self):
         return [dict(e) for e in AutotagConfig.INSTRUCTIONS], dict(

tests/test_utils_parallel.py

@@ -12,10 +12,12 @@ import unittest
 
 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 
-from utils import parallel  # noqa: E402
+from utils import exit_signal, parallel  # noqa: E402
 from utils.parallel import JOB_THREAD_PREFIX, map_parallel  # noqa: E402
 
 WAIT = 30
+# A map that has not returned by then is stuck (#593).
+HANG_LIMIT = 5
 
 
 def _job_threads():
@@ -163,5 +165,68 @@ class TestMapParallel(unittest.TestCase):
             map_parallel(boom, [1, 2, 3], workers=3)
 
 
+class TestErrorAfterShutdownCancel(unittest.TestCase):
+    """#593: ``shutdown_job_pool`` cancels the call's queued items without
+    notifying them, so the error path must not wait on them (exit hung)."""
+
+    def setUp(self):
+        exit_signal.reset()
+        self.addCleanup(exit_signal.reset)
+        self.assertTrue(parallel.shutdown_job_pool(WAIT))
+        self.addCleanup(parallel.shutdown_job_pool, WAIT)
+
+    def _run_in_thread(self, func, items):
+        """``map_parallel`` on a daemon thread: (thread, outcome list), so a
+        hang fails the test instead of hanging the suite."""
+        outcome = []
+
+        def call():
+            try:
+                outcome.append(("result", map_parallel(func, items, 1)))
+            except Exception as error:  # the outcome the test checks
+                outcome.append(("error", error))
+
+        thread = threading.Thread(target=call, daemon=True)
+        thread.start()
+        return thread, outcome
+
+    def _hold_first_then_cancel_rest(self, first):
+        """One worker runs item 0 (``first``) while items 1 and 2 queue;
+        shutdown cancels the queued two, then item 0 is released."""
+        entered, release = threading.Event(), threading.Event()
+
+        def item(n):
+            if n == 0:
+                entered.set()
+                self.assertTrue(release.wait(WAIT))
+                return first()
+            return n
+
+        thread, outcome = self._run_in_thread(item, [0, 1, 2])
+        self.assertTrue(entered.wait(WAIT))
+        self.assertFalse(parallel.shutdown_job_pool(0))  # item 0 still runs
+        release.set()
+        thread.join(HANG_LIMIT)
+        self.assertFalse(thread.is_alive(), "map_parallel hung on cancelled items")
+        return outcome
+
+    def test_a_running_item_raising_after_shutdown_cancel_returns(self):
+        def fail():
+            raise ValueError("zero")
+
+        outcome = self._hold_first_then_cancel_rest(fail)
+        self.assertEqual(outcome[0][0], "error")
+        self.assertIsInstance(outcome[0][1], ValueError)
+
+    def test_exit_requested_inline_after_shutdown_cancel_returns(self):
+        def signal_exit():
+            exit_signal.request()  # the cancelled items then run inline
+            return 0
+
+        outcome = self._hold_first_then_cancel_rest(signal_exit)
+        self.assertEqual(outcome[0][0], "error")
+        self.assertIsInstance(outcome[0][1], exit_signal.ExitRequested)
+
+
 if __name__ == "__main__":
     unittest.main()

utils/image_monitor.py

@@ -113,7 +113,10 @@ def _start_daemon_observer(observer) -> None:
     def start() -> None:
         try:
             observer.start()
-        except BaseException as e:  # re-raised on the caller's thread
+        # BaseException on purpose: a KeyboardInterrupt or SystemExit raised
+        # by start() is not lost on this thread; _start_daemon_observer
+        # re-raises it on the caller's thread below.
+        except BaseException as e:
             raised.append(e)
 
     starter = threading.Thread(

utils/native_memory.py

@@ -16,8 +16,9 @@ import sys
 import threading
 
 _lock = threading.Lock()
-_trim = None
-_looked_up = False
+# Not looked up yet; after the lookup, the function or None.
+_NOT_LOOKED_UP = object()
+_trim = _NOT_LOOKED_UP
 
 
 def _glibc() -> bool:
@@ -33,10 +34,9 @@ def _glibc() -> bool:
 
 def _malloc_trim():
     """glibc's ``malloc_trim``, looked up once; None where there is none."""
-    global _trim, _looked_up
+    global _trim
     with _lock:
-        if not _looked_up:
-            _looked_up = True
+        if _trim is _NOT_LOOKED_UP:
             _trim = _lookup()
         return _trim
 

utils/parallel.py

@@ -165,10 +165,12 @@ def _collect(func, items, futures):
     and wait for its running ones, as the old per-call pool's exit did."""
     try:
         return _run_rest(func, items, futures)
-    except BaseException:
-        for future in futures:
-            future.cancel()
-        wait(futures)
+    except BaseException:  # re-raised below, after the running items stop
+        # Wait only for what could not be cancelled (running or done). A
+        # future ``shutdown(cancel_futures=True)`` already cancelled stays
+        # CANCELLED, never notified, and ``wait`` would block on it forever
+        # (#593); ``cancel()`` is True for those as well.
+        wait([future for future in futures if not future.cancel()])
         raise
 
 

web/js/workspace/prompt-sender.mjs

@@ -63,15 +63,19 @@ export function createSendSession({client, roles:rawRoles, onChange}) {
         const target=host.targets.find(item=>item.id===state.targetId);
         return target ? [target] : [];
     }
-    // checkRevision is off only for a later node of "All", whose revision our own earlier write may have moved.
-    async function fetchPreview(request,hostId,target,checkRevision=true) {
+    // The adapter's revision is "<node identity>:<graph version>". Loading a workflow makes new node objects, so a new
+    // identity; our own write moves only the graph version. A string without the separator is all identity.
+    const nodeIdentity = revision => typeof revision==='string' && revision.includes(':') ? revision.slice(0,revision.indexOf(':')) : revision;
+    // graphMayMove is on only for a later node of "All", whose graph version our own earlier write may have moved; the node
+    // identity is still checked, so a write never lands in another workflow loaded in between (#593).
+    async function fetchPreview(request,hostId,target,graphMayMove=false) {
         const preview=await client.preview(hostId,target.id,roles,{signal:request.signal});
         if (!current(request)) return null;
         const fields=preview?.fields;
         if(typeof preview?.ticket!=='string' || !preview.ticket || preview.target?.id!==target.id || !Array.isArray(fields) || fields.length!==Object.keys(roles).length || new Set(fields.map(field=>field?.role)).size!==fields.length || Array.from(fields).some(field=>!field || !Object.hasOwn(roles,field.role) || typeof field.before!=='string' || field.after!==roles[field.role])) throw Error('ComfyUI returned an unexpected preview. Find the destination and preview again.');
         // Undefined on a tab still running older code, which has nothing to compare.
         const found=target.revision;
-        if(checkRevision && found!==undefined && preview.revision!==found) throw Error('The workflow changed since this window was found. Nothing was sent. Find windows again.');
+        if(found!==undefined && (graphMayMove ? nodeIdentity(preview.revision)!==nodeIdentity(found) : preview.revision!==found)) throw Error('The workflow changed since this window was found. Nothing was sent. Find windows again.');
         return preview;
     }
     function checkAcknowledgement(acknowledgement,targetId) {
@@ -95,25 +99,21 @@ export function createSendSession({client, roles:rawRoles, onChange}) {
         return reviewed;
     }
     /**
-     * Sends the reviewed nodes one after another, best effort past refusals. Only the first node of a pass taken in this
-     * click keeps its ticket: a held review may have expired at the host (30 s), and our own writes may make later
-     * tickets stale, so every other node takes a fresh ticket that must show exactly the reviewed text.
+     * Sends the reviewed nodes one after another, best effort past refusals. Every node takes a fresh ticket that must
+     * show exactly the reviewed text: a held review may have expired at the host (30 s), our own writes make later
+     * tickets stale, and the host keeps only its latest 100 tickets, so a pass over more nodes has evicted the first (#593).
      */
-    async function applyEach(request,hostId,reviewed,where,held) {
+    async function applyEach(request,hostId,reviewed,where) {
         const sent=[],refused=[];let written=false;
         for (const [index,{target,preview}] of reviewed.entries()) {
             let posted=false;
             try {
-                let ticket=preview.ticket;
-                if (index>0 || held) {
-                    // Until something was posted the revision is still the discovered one; after that our own write may move it.
-                    const fresh=await fetchPreview(request,hostId,target,!written);
-                    if (!fresh) return;
-                    if (!sameChanges(preview,fresh)) throw Error('Its prompt changed since the preview. Preview again.');
-                    ticket=fresh.ticket;
-                }
+                // Until something was posted the revision is still the discovered one; after that our own write may move it.
+                const fresh=await fetchPreview(request,hostId,target,written);
+                if (!fresh) return;
+                if (!sameChanges(preview,fresh)) throw Error('Its prompt changed since the preview. Preview again.');
                 posted=written=true;
-                const acknowledgement=await client.apply(hostId,ticket,{signal:request.signal});
+                const acknowledgement=await client.apply(hostId,fresh.ticket,{signal:request.signal});
                 if (!current(request)) return;
                 checkAcknowledgement(acknowledgement,target.id);
                 sent.push(target);
@@ -191,7 +191,7 @@ export function createSendSession({client, roles:rawRoles, onChange}) {
                 let reviewed;
                 try { reviewed=held?.length ? held : await previewEach(request,hostId,targets); }
                 catch(error) { if(current(request)) emit({phase:'error',message:error.message}); return; }
-                if (reviewed) await applyEach(request,hostId,reviewed,windowLabel,Boolean(held?.length));
+                if (reviewed) await applyEach(request,hostId,reviewed,windowLabel);
                 return;
             }
             const [target]=targets;