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"

database/health.py

@@ -199,23 +199,6 @@ def link_states(rows, roots, readable):
     check's ``missing_files`` and the maintenance missing-link plan both
     judge links this way, so their counts agree.
     """
-    for row, owner, state in _placed_links(rows, roots, readable):
-        if state is None:
-            state = "present" if os.path.exists(row[1]) else "missing"
-        yield row, owner, state
-
-
-def held_links(rows, roots, readable):
-    """``(row, owner, state)`` for each link ``link_states`` holds back
-    (``unavailable`` or ``outside``), without looking at any file (#586)."""
-    for row, owner, state in _placed_links(rows, roots, readable):
-        if state is not None:
-            yield row, owner, state
-
-
-def _placed_links(rows, roots, readable):
-    """``link_states`` before any file is looked at: state None means the
-    link lies under a readable source."""
     prefixes = [_root_prefix(root) for root in roots]
     for row in rows:
         owner = _image_owner(row[1], row[3], prefixes)
@@ -223,8 +206,10 @@ def _placed_links(rows, roots, readable):
             yield row, None, "outside"
         elif not readable[owner]:
             yield row, owner, "unavailable"
+        elif os.path.exists(row[1]):
+            yield row, owner, "present"
         else:
-            yield row, owner, None
+            yield row, owner, "missing"
 
 
 def _unavailable_line(unavailable, *, partial):

database/maintenance.py

@@ -43,7 +43,7 @@ from dataclasses import dataclass, field
 from pathlib import Path
 
 from .execution_contexts import LIVE_HINT_CTES
-from .health import BATCH_ROWS, held_links, link_states, readable_roots, readonly_uri
+from .health import BATCH_ROWS, link_states, readable_roots, readonly_uri
 from .image_removal_schema import TABLE as MARKS_TABLE
 from .image_removal_schema import clear_marks, has_marks_table
 from .pairs import PairDependencyError
@@ -130,16 +130,6 @@ _UNLINKED_PROTECTIONS_SQL = (
     + "), 0) FROM prompts p WHERE NOT EXISTS "
     "(SELECT 1 FROM generated_images gi WHERE gi.prompt_id = p.id)"
 )
-# Each prompt no image links to, with which protections keep it (#586); the
-# ones with none are filtered out in Python, so each predicate runs once.
-_UNLINKED_KEPT_SQL = (
-    "WITH "
-    + LIVE_HINT_CTES
-    + " SELECT p.id, "
-    + ", ".join(f"({predicate})" for predicate, _many, _one in PROTECTIONS)
-    + " FROM prompts p WHERE NOT EXISTS "
-    "(SELECT 1 FROM generated_images gi WHERE gi.prompt_id = p.id) ORDER BY p.id"
-)
 
 # Duplicate prompt groups, as the legacy cleanup finds them (case-insensitive
 # text; GROUP_CONCAT order is sorted by created_at afterwards).
@@ -169,14 +159,6 @@ _PROMPT_STATE_SQL = (
     "ORDER BY id",
 )
 
-# Why the missing-link cleanup holds a link back: (one, many) (#586).
-HELD_LINK_REASONS = {
-    "unavailable": (
-        "link under an unavailable source",
-        "links under unavailable sources",
-    ),
-    "outside": ("link outside the image sources", "links outside the image sources"),
-}
 NO_FILES_DELETED = "No image file is deleted."
 LINKS_ONLY = "Database links only; no image file is deleted."
 UNUSED_DETAIL = "No image, saved pair, Library state or recorded run uses it"
@@ -190,7 +172,6 @@ CONFLICT_REASONS = {
         "groups have their own Library titles",
     ),
 }
-PAIRED_DETAIL = "Kept: belongs to a saved pair"
 CONFLICT_DETAILS = {
     "collision": "Kept: two copies link the same image",
     "title": "Kept: a copy has its own Library title",
@@ -432,35 +413,6 @@ def _source_name(root, index):
     return os.path.basename(str(root).rstrip("\\/")) or f"Source {index + 1}"
 
 
-def _kept_duplicates(paired, conflicted):
-    """``{prompt_id, reasons}`` for each prompt of a group kept whole (#586).
-
-    The reasons are the review's details without "Kept: ".
-    """
-    held = [(ids, PAIRED_DETAIL) for ids in paired] + [
-        (ids, CONFLICT_DETAILS[reason]) for ids, reason in conflicted
-    ]
-    return [
-        {"prompt_id": prompt_id, "reasons": [detail.removeprefix("Kept: ")]}
-        for ids, detail in held
-        for prompt_id in ids
-    ]
-
-
-def _kept_unused(conn):
-    """``{prompt_id, reasons}`` for each prompt the unused prune keeps (#586).
-
-    The reasons are the review's, one per protection that applies, in the
-    order the plan names them.
-    """
-    kept = []
-    for row in conn.execute(_UNLINKED_KEPT_SQL):
-        reasons = [one for flag, (_p, _m, one) in zip(row[1:], PROTECTIONS) if flag]
-        if reasons:
-            kept.append({"prompt_id": row[0], "reasons": reasons})
-    return kept
-
-
 class MaintenanceOperations:
     """Mixin for PromptDatabase: maintenance previews and their apply."""
 
@@ -640,7 +592,7 @@ class MaintenanceOperations:
             + [
                 {
                     "label": _excerpt(texts.get(group[0])),
-                    "detail": PAIRED_DETAIL,
+                    "detail": "Kept: belongs to a saved pair",
                 }
                 for group in paired
             ]
@@ -718,24 +670,6 @@ class MaintenanceOperations:
             ],
         )
 
-    def kept_links(self, roots):
-        """``{image_path, owner, reason}`` for each link the missing-link
-        cleanup holds back under ``roots`` (blocking, reads only, no file
-        opened); ``owner`` indexes ``roots`` (None outside every source).
-
-        ``reason`` is the review's (``HELD_LINK_REASONS``), one link each.
-        """
-        roots = [str(root) for root in roots]
-        rows = _read_snapshot(self.model.db_path, self._link_rows)
-        return [
-            {
-                "image_path": row[1],
-                "owner": owner,
-                "reason": HELD_LINK_REASONS[state][0],
-            }
-            for row, owner, state in held_links(rows, roots, readable_roots(roots))
-        ]
-
     @staticmethod
     def _links_fingerprint(candidates, stale=()):
         fingerprint = _Fingerprint("missing_links")
@@ -755,11 +689,23 @@ class MaintenanceOperations:
                 unavailable += 1
             elif state == "outside":
                 outside += 1
-        reasons = [
-            _plural(count, *HELD_LINK_REASONS[state])
-            for state, count in (("unavailable", unavailable), ("outside", outside))
-            if count
-        ]
+        reasons = []
+        if unavailable:
+            reasons.append(
+                _plural(
+                    unavailable,
+                    "link under an unavailable source",
+                    "links under unavailable sources",
+                )
+            )
+        if outside:
+            reasons.append(
+                _plural(
+                    outside,
+                    "link outside the image sources",
+                    "links outside the image sources",
+                )
+            )
         count = len(candidates)
         parts = []
         if count:
@@ -863,10 +809,7 @@ class MaintenanceOperations:
         (``{prompt_id[, kept_id]}`` or ``{image_path}``). ``marks`` are a
         missing-links plan's stale removal marks; that apply also returns
         ``marks``, how many it forgot. ``pair_metadata`` returns every filled
-        prompt, uncapped (its job keeps the complete record, #550). The
-        prompt operations also return ``kept``: every prompt the plan keeps,
-        ``{prompt_id, reasons}`` in the review's wording (#550, #586); the
-        links a missing-links plan keeps are read by ``kept_links``.
+        prompt, uncapped (its job keeps the complete record, #550).
         """
         if operation not in MAINTENANCE_OPERATIONS or operation == "workflow_storage":
             raise ValueError(f"Unknown maintenance operation {operation!r}")
@@ -892,11 +835,10 @@ class MaintenanceOperations:
             return appliers[operation](conn, fingerprint, targets, roots)
 
     def _apply_duplicate_prompts(self, conn, fingerprint, targets, _roots):
-        groups, paired, conflicted, current = self._duplicates_fingerprint(conn)
+        groups, _paired, _conflicted, current = self._duplicates_fingerprint(conn)
         if current != fingerprint or groups != targets:
             raise PlanChanged("duplicate_prompts")
         changed, rows = 0, []
-        kept = _kept_duplicates(paired, conflicted)
         for sorted_ids in groups:
             removed, _moved = self._merge_duplicate_group(conn, sorted_ids, strict=True)
             changed += removed
@@ -904,7 +846,7 @@ class MaintenanceOperations:
                 {"prompt_id": duplicate_id, "kept_id": sorted_ids[0]}
                 for duplicate_id in sorted_ids[1:]
             )
-        return {"changed": changed, "rows": rows[:MAX_RESULT_ROWS], "kept": kept}
+        return {"changed": changed, "rows": rows[:MAX_RESULT_ROWS]}
 
     def _apply_unused_components(self, conn, fingerprint, targets, _roots):
         ids, current = self._unused_fingerprint(conn)
@@ -916,7 +858,7 @@ class MaintenanceOperations:
             {"prompt_id": prompt_id, "text": _excerpt(texts.get(prompt_id))}
             for prompt_id in ids[:MAX_RESULT_ROWS]
         ]
-        return {"changed": changed, "rows": rows, "kept": _kept_unused(conn)}
+        return {"changed": changed, "rows": rows}
 
     def _apply_missing_links(self, conn, fingerprint, targets, roots, marks):
         rows = []

database/workflow_storage.py

@@ -112,22 +112,20 @@ QUERY_PLAN_STATEMENTS = {
 
 
 def _rewrite_row(conn, image_id, row, verified, checked):
-    """Store the row's inline texts as documents; returns (changed, kept, size).
-
-    ``kept`` names the columns whose value stays inline (not text).
+    """Store the row's inline texts as documents; returns (changed, protected, size).
 
     ``verified`` maps document ids already read back in this batch to their
     text, and ``checked`` holds the reused ids ``store_text`` already
     checked, so a document shared by many rows is decompressed once.
     """
-    changes, replaced, kept = {}, [], []
-    size = 0
+    changes, replaced = {}, []
+    protected = size = 0
     for index, (column, reference) in enumerate(TEXT_COLUMNS):
         value, current = row[index], row[index + len(TEXT_COLUMNS)]
         if value is None:
             continue
         if not storable(value):
-            kept.append(column)
+            protected += 1
             continue
         document_id = store_text(conn, value, checked)
         if document_id not in verified:
@@ -145,7 +143,7 @@ def _rewrite_row(conn, image_id, row, verified, checked):
             [*changes.values(), image_id],
         )
     release_documents(conn, replaced)
-    return bool(changes), kept, size
+    return bool(changes), protected, size
 
 
 def migrate_batch(
@@ -162,10 +160,9 @@ def migrate_batch(
     savepoint: a row that cannot be rewritten (``_row_error``) is rolled
     back alone, stays inline and is listed in ``errors``; anything else rolls
     the whole batch back and propagates. Returns ``{last_id, rows,
-    protected, kept, errors, done}``: the last row looked at, rows
-    rewritten, inline values kept (not text) and each of them as ``(image
-    id, column)`` (#586), the ids kept by an error and whether there was
-    nothing left to look at.
+    protected, errors, done}``: the last row looked at, rows rewritten,
+    inline values kept (not text), the ids kept by an error and whether
+    there was nothing left to look at.
     """
     if conn.in_transaction:
         raise RuntimeError("The migration needs its own transaction")
@@ -177,7 +174,6 @@ def migrate_batch(
             "last_id": after_id,
             "rows": 0,
             "protected": 0,
-            "kept": [],
             "errors": [],
             "done": not ids,
         }
@@ -186,7 +182,7 @@ def migrate_batch(
             conn.execute("SAVEPOINT workflow_row")
             try:
                 row = conn.execute(ROW_SQL, (image_id,)).fetchone()
-                changed, kept, row_size = _rewrite_row(
+                changed, protected, row_size = _rewrite_row(
                     conn, image_id, row, verified, checked
                 )
             except Exception as error:
@@ -198,12 +194,11 @@ def migrate_batch(
                 checked.clear()
                 logger.warning(f"Image record {image_id} kept inline: {error}")
                 result["errors"].append(image_id)
-                changed, kept, row_size = False, [], 0
+                changed, protected, row_size = False, 0, 0
             else:
                 conn.execute("RELEASE workflow_row")
             result["rows"] += int(changed)
-            result["protected"] += len(kept)
-            result["kept"].extend((image_id, column) for column in kept)
+            result["protected"] += protected
             result["last_id"] = image_id
             size += row_size
             if size >= max_bytes or _clock() - started >= max_seconds:

py/api/duplicate_files.py

@@ -65,7 +65,6 @@ from .duplicate_review import (  # noqa: F401 (CHANGED, OUTSIDE_SOURCE: public)
     NOT_ATTEMPTED,
     OUTSIDE_SOURCE,
     RECORDS_FAILED,
-    REVIEW_KEPT_DETAILS,
     ROW_DETAILS,
     SAME_FILE,
     SOURCES_CHANGED,
@@ -435,7 +434,6 @@ class DuplicateFilesMixin:
             {
                 "targets": review["targets"],
                 "kept": review["kept"],
-                "held": review["held"],
                 "protected": review["public"]["protected"]["count"],
             },
         )
@@ -448,7 +446,7 @@ class DuplicateFilesMixin:
         job hashes each file again right before it unlinks it.
         """
         listed = self._listed_sources()
-        targets, kept, protected, held = [], [], {}, []
+        targets, kept, protected = [], [], {}
         for group, keepers in selection.values():
             removals = [r for r in group["files"] if r not in keepers]
             for record in removals:
@@ -469,7 +467,6 @@ class DuplicateFilesMixin:
                     problem = _bound_problem(record, group, keeper)
                 if problem:
                     protected[problem] = protected.get(problem, 0) + 1
-                    held.append(self._copy_row(record, REVIEW_KEPT_DETAILS[problem]))
                     continue
                 reviewed.append(self._target(group, record, keeper))
             if reviewed:
@@ -491,7 +488,6 @@ class DuplicateFilesMixin:
         return {
             "targets": targets,
             "kept": kept,
-            "held": held,
             "fingerprint": _fingerprint(targets),
             "public": _public_plan(targets, protected, records),
         }
@@ -588,13 +584,6 @@ class DuplicateFilesMixin:
 
     async def _remove_duplicates(self, job, db, plan, backup):
         targets = plan.payload["targets"]
-        # #586: every copy the review protected, with its reason, and the
-        # whole group in the report.
-        held = plan.payload["held"]
-        job.record_counted("kept", len(held), held, complete=True)
-        if held:
-            copies = _plural(len(held), "copy", "copies")
-            job.log_line(f"Kept {copies} the review protected")
         for row in plan.payload["kept"]:
             job.record("kept", row, f"= {row['root']}/{row['path']} · {KEPT_COPY}")
         job.update(total=len(targets), status_line="Removing duplicate files")

py/api/duplicate_review.py

@@ -70,18 +70,6 @@ PROTECTED_REASONS = {
         "files kept: their kept copy changed or is gone",
     ),
 }
-# The job's row for a copy the review protected (#586): the review's reason.
-REVIEW_KEPT_DETAILS = {
-    MAYBE_SAME: "Kept: may be the same file reached through another mount or share",
-    NO_FILE_ID: "Kept: can't confirm this is a separate file on this drive",
-    SAME: "Kept: the same file as its kept copy",
-    OUTSIDE_SOURCE: "Kept: not inside a configured source",
-    LINK: "Kept: a link, not a regular file",
-    CHANGED_CONTENT: "Kept: changed since the search",
-    GONE: "Kept: no longer exists",
-    UNREADABLE: "Kept: could not be read",
-    KEEPER: "Kept: its kept copy changed or is gone",
-}
 # Job outcome rows (never an OS error text).
 CHANGED = "The file changed since the review; it was kept."
 KEEPER_CHANGED = "The kept copy changed or is gone; this file was kept."

py/api/jobs.py

@@ -288,16 +288,15 @@ class Job:
         index = row.get("root_index")
         if isinstance(index, bool) or not isinstance(index, int) or index < 0:
             index = None
-        shaped = {
-            "path": path,
-            "root": str(row.get("root") or ""),
-            "root_index": index,
-            "detail": str(row.get("detail") or ""),
-        }
-        if group in self.complete:  # a group kept whole for the report (#586)
-            self.complete[group].append(dict(shaped))
         if len(rows) < MAX_OUTCOME_ROWS:
-            rows.append(shaped)
+            rows.append(
+                {
+                    "path": path,
+                    "root": str(row.get("root") or ""),
+                    "root_index": index,
+                    "detail": str(row.get("detail") or ""),
+                }
+            )
         self.log_line(log_line)
 
     def record_counted(self, group, count, rows=(), *, complete=False):

py/api/log_cleanup.py

@@ -68,23 +68,6 @@ def _count(count, singular, plural):
     return f"{count} {singular if count == 1 else plural}"
 
 
-# Each kept entry's row in the job (#586), in the review's "Kept: ..." words.
-KEPT_DETAILS = {
-    "active": "Kept: the active log",
-    "link": "Kept: a link",
-    "other": "Kept: another file",
-}
-
-
-def _kept_files(rows):
-    """``{name, kind}`` of each entry the cleanup keeps, in folder order."""
-    return [
-        {"name": row["name"], "kind": row["kind"]}
-        for row in rows
-        if row["kind"] in KEPT_DETAILS
-    ]
-
-
 def _protected_reasons(rows):
     """The plan's Protected reasons: what the cleanup keeps, zero counts left out."""
     active = sum(1 for row in rows if row["kind"] == "active")
@@ -146,7 +129,7 @@ class LogCleanupMixin:
             token = store.issue(
                 ROTATED_LOGS,
                 json.dumps(files, sort_keys=True),
-                {"files": files, "protected": protected, "kept": _kept_files(rows)},
+                {"files": files, "protected": protected},
                 (),
             )
         except ScopeTooLarge:
@@ -215,11 +198,7 @@ class LogCleanupMixin:
             job.record(
                 "errors", {"path": name, "detail": detail}, f"! {name} · {detail}"
             )
-        kept = [
-            {"path": row["name"], "detail": KEPT_DETAILS[row["kind"]]}
-            for row in plan.payload["kept"]
-        ]
-        job.record_counted("protected", len(kept), kept, complete=True)
+        job.record_counted("protected", plan.payload["protected"])
         removed, failed = len(result["removed"]), len(result["failed"])
         line = f"Deleted {_count(removed, 'rotated log copy', 'rotated log copies')}"
         if failed:

py/api/maintenance.py

@@ -697,13 +697,19 @@ class MaintenanceRoutesMixin:
             rows,
             complete=plan.operation in COMPLETE_RECORD,
         )
-        kept = await self._run_in_executor(self._kept_rows, db, plan, result)
-        if kept is None:
-            job.record_counted("protected", payload["protected"])  # none kept
-        else:
-            # #550, #586: each kept record with its reason, like the changed
-            # ones, and every one of them in the report.
+        if "kept" in result:
+            # #550: each kept record with its reason, like the changed ones.
+            kept = [
+                {
+                    "path": "",
+                    "detail": f"Prompt {item['prompt_id']}: kept, "
+                    + "; ".join(item["reasons"]),
+                }
+                for item in result["kept"]
+            ]
             job.record_counted("protected", len(kept), kept, complete=True)
+        else:
+            job.record_counted("protected", payload["protected"])
         line = _changed_line(
             plan.operation, result["changed"], result.get("marks", 0), result
         )
@@ -711,47 +717,6 @@ class MaintenanceRoutesMixin:
         job.update(progress=100, status_line=f"{line} · {backup_note(backup)}")
         return None
 
-    def _kept_rows(self, db, plan, result):
-        """Outcome rows for the records the apply kept (blocking), or None.
-
-        Prompts are records (no path); a missing-links plan's kept links
-        name their file, relative to its output root, so Activity can open
-        and locate it (#586); one under a source that is gone keeps its path
-        under that source's label. None for an operation that keeps nothing
-        (Optimize).
-        """
-        if plan.operation == "missing_links":
-            roots = _scan_roots(self._get_all_output_dirs())
-            judged = plan.payload["roots"]
-            rows = []
-            for item in db.kept_links(judged):
-                path, label, index = _scan_location(item["image_path"], roots)
-                if index is None and item["owner"] is not None:
-                    # A source that is gone is not listed: name the file
-                    # under it, with no listing index to open it by.
-                    source = judged[item["owner"]]
-                    where = [(os.path.abspath(source), self._public_path(source), None)]
-                    path, label, index = _scan_location(item["image_path"], where)
-                rows.append(
-                    {
-                        "path": path,
-                        "root": label,
-                        "root_index": index,
-                        "detail": f"Kept: {item['reason']}",
-                    }
-                )
-            return rows
-        if "kept" not in result:
-            return None
-        return [
-            {
-                "path": "",
-                "detail": f"Prompt {item['prompt_id']}: kept, "
-                + "; ".join(item["reasons"]),
-            }
-            for item in result["kept"]
-        ]
-
     def _maintenance_rows(self, operation, described):
         """Outcome rows for the changed records (root-relative paths only)."""
         if operation == "missing_links":

py/api/workflow_storage.py

@@ -59,9 +59,6 @@ REWRITE_FAILED = (
 )
 RECLAIM_FAILED = "Reclaim space failed; the database is unchanged."
 ROW_KEPT = "could not be rewritten; it keeps its inline text and reads as before"
-# A protected value's row (#586), in the review's "not text; kept as is".
-VALUE_KEPT = "stored value not text; kept as is"
-KEPT_COLUMNS = {"workflow_data": "workflow", "prompt_metadata": "prompt graph"}
 
 
 NOTHING_LEFT = "Every image record already stores its workflow once"
@@ -180,19 +177,7 @@ class WorkflowStorageRoutesMixin:
             protected += batch["protected"]
             errors += len(batch["errors"])
             job.record_counted("changed", batch["rows"])
-            job.record_counted(
-                "protected",
-                batch["protected"],
-                [
-                    {
-                        "path": "",
-                        "detail": f"Image record {image_id}: "
-                        f"{KEPT_COLUMNS.get(column, column)}, {VALUE_KEPT}",
-                    }
-                    for image_id, column in batch["kept"]
-                ],
-                complete=True,
-            )
+            job.record_counted("protected", batch["protected"])
             for image_id in batch["errors"]:
                 job.record(
                     "errors",

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_api_duplicate_files.py

@@ -452,40 +452,6 @@ class TestReview(DuplicateFilesTestCase):
         )
 
 
-class TestReviewProtectedRows(DuplicateFilesTestCase):
-    async def test_cleanup_lists_each_copy_the_review_kept(self):
-        if not TestReview._can_symlink(self.tmpdir):  # pragma: no cover
-            self.skipTest("symlinks are not available")
-        self.trio()
-        job, group = await self.found_group()
-        # b.png becomes a link after the find: the review protects it (#586).
-        (self.output_dir / "b.png").unlink()
-        os.symlink(str(self.output_dir / "a.png"), str(self.output_dir / "b.png"))
-        plan = await self.review(job.id, [self.keep_of(group, "a.png")])
-        self.assertEqual(plan["protected"]["count"], 1)
-        body = await self.apply(plan["token"])
-        cleanup = self.api._jobs.get(body["job"]["id"])
-        await asyncio.wait_for(asyncio.shield(cleanup.task), DISK_BOUND_SECONDS)
-        self.assertEqual(cleanup.state, "finished", cleanup.log)
-        rows = {row["path"]: row for row in cleanup.outcomes["kept"]}
-        self.assertEqual(
-            rows["b.png"],
-            {
-                "path": "b.png",
-                "root": rows["a.png"]["root"],
-                "root_index": 0,
-                "detail": "Kept: a link, not a regular file",
-            },
-        )
-        self.assertEqual(rows["a.png"]["detail"], "Kept copy")
-        self.assertEqual(cleanup.counters, {"removed": 1, "kept": 2, "errors": 0})
-        report = cleanup.report()["outcomes"]["kept"]
-        self.assertEqual(
-            sorted(row["path"] for row in report["rows"]), ["a.png", "b.png"]
-        )
-        self.assertFalse(report["truncated"])
-
-
 class TestCleanupJob(DuplicateFilesTestCase):
     async def test_cleanup_is_a_records_job_with_a_safety_backup(self):
         self.trio()

tests/test_api_log_cleanup.py

@@ -266,30 +266,6 @@ class TestApply(LogCleanupTestCase):
         report = json.dumps(job.report())
         self.assertNotIn(self.tmpdir, report)
 
-    async def test_apply_lists_each_kept_file(self):
-        linked = self.seed()
-        plan = await self.preview()
-        _summary, job = await self.run_apply(plan)
-        expected = [
-            ("prompt_manager.log", "Kept: the active log"),
-            ("prompt_manager.log.9", "Kept: another file"),
-            ("prompt_manager.log.bak", "Kept: another file"),
-        ]
-        if linked:
-            expected.append(("prompt_manager.log.4", "Kept: a link"))
-        rows = job.outcomes["protected"]
-        self.assertEqual(
-            sorted((row["path"], row["detail"]) for row in rows), sorted(expected)
-        )
-        self.assertEqual(
-            {(row["root"], row["root_index"]) for row in rows}, {("", None)}
-        )
-        report = job.report()["outcomes"]["protected"]
-        self.assertEqual(
-            (report["count"], len(report["rows"]), report["truncated"]),
-            (plan["protected"]["count"], len(expected), False),
-        )
-
     async def test_apply_rows_name_each_file(self):
         self.seed()
         plan = await self.preview()

tests/test_api_maintenance_protected_rows.py

@@ -1,140 +0,0 @@
-"""Protected maintenance items are listed in Activity, not only counted (#586).
-
-Each record a maintenance job keeps gets one ``protected`` row with its
-reason, in the wording of that operation's review, and the job's
-downloadable report lists every one of them (never cut at 200 rows). A kept
-image link names its file, so Activity can still open and locate it.
-"""
-
-import os
-import shutil
-import sys
-from pathlib import Path
-from unittest.mock import MagicMock
-
-_mock_server = MagicMock()
-_mock_server.PromptServer.instance.routes = MagicMock()
-sys.modules.setdefault("server", _mock_server)
-
-sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
-
-from py.api.jobs import MAX_OUTCOME_ROWS  # noqa: E402
-from py.config import GalleryConfig, PromptManagerConfig  # noqa: E402
-
-try:
-    from tests.test_api_maintenance import ApplyTestCase  # noqa: E402
-except ImportError:  # discovered with tests/ as the top-level directory
-    from test_api_maintenance import ApplyTestCase  # noqa: E402
-
-
-def _record(detail):
-    return {"path": "", "root": "", "root_index": None, "detail": detail}
-
-
-class TestProtectedRows(ApplyTestCase):
-    def duplicate(self, text, created_at="2999-01-01"):
-        with self.api.db.model.get_connection() as conn:
-            return conn.execute(
-                "INSERT INTO prompts(text, created_at) VALUES(?, ?)",
-                (text.upper(), created_at),
-            ).lastrowid
-
-    def assert_complete(self, job, count):
-        self.assertEqual(job.counters["protected"], count)
-        report = job.report()["outcomes"]["protected"]
-        self.assertEqual(
-            (report["count"], len(report["rows"]), report["truncated"]),
-            (count, count, False),
-        )
-
-    async def test_duplicate_prompts_lists_each_kept_copy(self):
-        paired = self.save("A dog")
-        self.api.db.save_prompt_pair("A dog", "")
-        paired_copy = self.duplicate("A dog")
-        titled = self.save("A bird")
-        titled_copy = self.duplicate("A bird")
-        with self.api.db.model.get_connection() as conn:
-            conn.execute(
-                "INSERT INTO prompt_library_state(prompt_id,title) VALUES(?,?)",
-                (titled_copy, "Bird recipe"),
-            )
-        plan, _summary, job = await self.run_apply("duplicate_prompts")
-        self.assertEqual(job.state, "finished", job.log)
-        self.assertEqual(
-            sorted(row["detail"] for row in job.outcomes["protected"]),
-            sorted(
-                [
-                    f"Prompt {paired}: kept, belongs to a saved pair",
-                    f"Prompt {paired_copy}: kept, belongs to a saved pair",
-                    f"Prompt {titled}: kept, a copy has its own Library title",
-                    f"Prompt {titled_copy}: kept, a copy has its own Library title",
-                ]
-            ),
-        )
-        self.assertEqual(plan["protected"]["count"], 4)
-        self.assert_complete(job, 4)
-
-    async def test_unused_components_lists_each_kept_prompt_with_its_reasons(self):
-        self.save("Orphan")
-        tagged = self.save("Kept", tags=["__protected__"])
-        self.api.db.save_prompt_pair("Paired", "")
-        paired = self.api.db.search_prompts(text="Paired")[0]["id"]
-        with self.api.db.model.get_connection() as conn:
-            conn.execute(
-                "INSERT INTO prompt_tags(prompt_id, tag_id) "
-                "SELECT ?, id FROM tags WHERE name='__protected__'",
-                (paired,),
-            )
-        plan, _summary, job = await self.run_apply("unused_components")
-        self.assertEqual(job.state, "finished", job.log)
-        self.assertEqual(
-            sorted(job.outcomes["protected"], key=lambda row: row["detail"]),
-            sorted(
-                [
-                    _record(f"Prompt {tagged}: kept, tagged __protected__"),
-                    _record(
-                        f"Prompt {paired}: kept, in a saved pair; "
-                        "tagged __protected__"
-                    ),
-                ],
-                key=lambda row: row["detail"],
-            ),
-        )
-        self.assertEqual(plan["protected"]["count"], 2)
-        self.assert_complete(job, 2)
-
-    async def test_report_lists_every_kept_prompt(self):
-        PromptManagerConfig.SAFETY_BACKUP = False
-        count = MAX_OUTCOME_ROWS + 5
-        for n in range(count):
-            self.save(f"Kept {n:03d}", tags=["__protected__"])
-        _plan, _summary, job = await self.run_apply("unused_components")
-        self.assertEqual(len(job.outcomes["protected"]), MAX_OUTCOME_ROWS)
-        self.assert_complete(job, count)
-
-    async def test_missing_links_lists_each_kept_link_with_its_file(self):
-        unmounted = Path(self.tmpdir) / "unmounted"
-        unmounted.mkdir()
-        GalleryConfig.MONITORING_DIRECTORIES = [str(self.output_dir), str(unmounted)]
-        prompt = self.save("Links")
-        self.image(prompt, "gone.png", exists=False)
-        self.image(prompt, "sub/away.png", folder=unmounted, exists=False)
-        self.image(prompt, "x.png", folder=Path(self.tmpdir) / "elsewhere")
-        shutil.rmtree(unmounted)
-        plan, _summary, job = await self.run_apply("missing_links")
-        self.assertEqual(job.state, "finished", job.log)
-        self.assertEqual(plan["protected"]["count"], 2)
-        rows = sorted(job.outcomes["protected"], key=lambda row: row["path"])
-        self.assertEqual(
-            [(row["path"], row["detail"]) for row in rows],
-            [
-                ("sub/away.png", "Kept: link under an unavailable source"),
-                ("x.png", "Kept: link outside the image sources"),
-            ],
-        )
-        # The gone source keeps the path under its label; outside every
-        # source the row names the file only.
-        self.assertEqual((rows[0]["root"], rows[0]["root_index"]), ("unmounted", None))
-        self.assertEqual((rows[1]["root"], rows[1]["root_index"]), ("", None))
-        self.assertNotIn(self.tmpdir, str(job.report()))
-        self.assert_complete(job, 2)

tests/test_api_workflow_storage.py

@@ -363,11 +363,11 @@ class TestWorkflowStorageOutcomes(WorkflowStorageTestCase):
         prompt_id = self.old_library(count=4)
         conn = self.api.db.model.get_connection()
         with conn:
-            image_id = conn.execute(
+            conn.execute(
                 "INSERT INTO generated_images (prompt_id, image_path, filename, "
                 "file_path, workflow_data) VALUES (?, '/x/b', 'b', '/x/b', x'00')",
                 (prompt_id,),
-            ).lastrowid
+            )
             store_text(conn, "a document no record references")
         _plan, _summary, job = await self.start()
         await self.finish(job)
@@ -377,13 +377,6 @@ class TestWorkflowStorageOutcomes(WorkflowStorageTestCase):
             "Rewrote 4 image records; 1 stored value not text, kept as is", log
         )
         self.assertIn("Removed 1 unused workflow document", log)
-        # #586: the kept value has its row, in the review's wording.
-        self.assertEqual(
-            [row["detail"] for row in job.outcomes["protected"]],
-            [f"Image record {image_id}: workflow, stored value not text; kept as is"],
-        )
-        report = job.report()["outcomes"]["protected"]
-        self.assertEqual((report["count"], report["truncated"]), (1, False))
 
     async def reclaim_with(self, **patches):
         self.old_library(count=4)

tests/test_jobs.py

@@ -506,20 +506,6 @@ class TestJobOutcomesAndLog(unittest.TestCase):
         self.assertFalse(report["changed"]["truncated"])
         self.assertEqual(report["errors"], {"count": 0, "rows": [], "truncated": False})
 
-    def test_record_into_a_complete_group_keeps_the_row_for_the_report(self):
-        # #586: a group kept whole stays whole when later rows are recorded.
-        job = Job("duplicate_cleanup", "Remove", groups=("kept",))
-        rows = [
-            {"path": f"f{i}.png", "detail": "Kept"} for i in range(MAX_OUTCOME_ROWS)
-        ]
-        job.record_counted("kept", len(rows), rows, complete=True)
-        job.record("kept", {"path": "late.png", "detail": "Kept copy"}, "= late.png")
-        self.assertEqual(len(job.outcomes["kept"]), MAX_OUTCOME_ROWS)
-        report = job.report()["outcomes"]["kept"]
-        self.assertEqual(report["count"], MAX_OUTCOME_ROWS + 1)
-        self.assertEqual(report["rows"][-1]["path"], "late.png")
-        self.assertFalse(report["truncated"])
-
     def test_thumbnail_job_groups(self):
         self.assertEqual(
             jobs_module.THUMBNAIL_GROUP_NAMES, ("generated", "skipped", "errors")

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;