fix(maintenance): list every protected item with its reason (#586)

Squashed from PR #596. Closes #586.

__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,6 +199,23 @@ 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)
@@ -206,10 +223,8 @@ def link_states(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, "missing"
+            yield row, owner, None
 
 
 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, link_states, readable_roots, readonly_uri
+from .health import BATCH_ROWS, held_links, 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,6 +130,16 @@ _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).
@@ -159,6 +169,14 @@ _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"
@@ -172,6 +190,7 @@ 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",
@@ -413,6 +432,35 @@ 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."""
 
@@ -592,7 +640,7 @@ class MaintenanceOperations:
             + [
                 {
                     "label": _excerpt(texts.get(group[0])),
-                    "detail": "Kept: belongs to a saved pair",
+                    "detail": PAIRED_DETAIL,
                 }
                 for group in paired
             ]
@@ -670,6 +718,24 @@ 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")
@@ -689,23 +755,11 @@ class MaintenanceOperations:
                 unavailable += 1
             elif state == "outside":
                 outside += 1
-        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",
-                )
-            )
+        reasons = [
+            _plural(count, *HELD_LINK_REASONS[state])
+            for state, count in (("unavailable", unavailable), ("outside", outside))
+            if count
+        ]
         count = len(candidates)
         parts = []
         if count:
@@ -809,7 +863,10 @@ 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).
+        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``.
         """
         if operation not in MAINTENANCE_OPERATIONS or operation == "workflow_storage":
             raise ValueError(f"Unknown maintenance operation {operation!r}")
@@ -835,10 +892,11 @@ 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
@@ -846,7 +904,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]}
+        return {"changed": changed, "rows": rows[:MAX_RESULT_ROWS], "kept": kept}
 
     def _apply_unused_components(self, conn, fingerprint, targets, _roots):
         ids, current = self._unused_fingerprint(conn)
@@ -858,7 +916,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}
+        return {"changed": changed, "rows": rows, "kept": _kept_unused(conn)}
 
     def _apply_missing_links(self, conn, fingerprint, targets, roots, marks):
         rows = []

database/workflow_storage.py

@@ -112,20 +112,22 @@ QUERY_PLAN_STATEMENTS = {
 
 
 def _rewrite_row(conn, image_id, row, verified, checked):
-    """Store the row's inline texts as documents; returns (changed, protected, size).
+    """Store the row's inline texts as documents; returns (changed, kept, size).
+
+    ``kept`` names the columns whose value stays inline (not text).
 
     ``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 = {}, []
-    protected = size = 0
+    changes, replaced, kept = {}, [], []
+    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):
-            protected += 1
+            kept.append(column)
             continue
         document_id = store_text(conn, value, checked)
         if document_id not in verified:
@@ -143,7 +145,7 @@ def _rewrite_row(conn, image_id, row, verified, checked):
             [*changes.values(), image_id],
         )
     release_documents(conn, replaced)
-    return bool(changes), protected, size
+    return bool(changes), kept, size
 
 
 def migrate_batch(
@@ -160,9 +162,10 @@ 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, 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.
+    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.
     """
     if conn.in_transaction:
         raise RuntimeError("The migration needs its own transaction")
@@ -174,6 +177,7 @@ def migrate_batch(
             "last_id": after_id,
             "rows": 0,
             "protected": 0,
+            "kept": [],
             "errors": [],
             "done": not ids,
         }
@@ -182,7 +186,7 @@ def migrate_batch(
             conn.execute("SAVEPOINT workflow_row")
             try:
                 row = conn.execute(ROW_SQL, (image_id,)).fetchone()
-                changed, protected, row_size = _rewrite_row(
+                changed, kept, row_size = _rewrite_row(
                     conn, image_id, row, verified, checked
                 )
             except Exception as error:
@@ -194,11 +198,12 @@ def migrate_batch(
                 checked.clear()
                 logger.warning(f"Image record {image_id} kept inline: {error}")
                 result["errors"].append(image_id)
-                changed, protected, row_size = False, 0, 0
+                changed, kept, row_size = False, [], 0
             else:
                 conn.execute("RELEASE workflow_row")
             result["rows"] += int(changed)
-            result["protected"] += protected
+            result["protected"] += len(kept)
+            result["kept"].extend((image_id, column) for column in kept)
             result["last_id"] = image_id
             size += row_size
             if size >= max_bytes or _clock() - started >= max_seconds:

py/api/duplicate_files.py

@@ -65,6 +65,7 @@ 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,
@@ -434,6 +435,7 @@ class DuplicateFilesMixin:
             {
                 "targets": review["targets"],
                 "kept": review["kept"],
+                "held": review["held"],
                 "protected": review["public"]["protected"]["count"],
             },
         )
@@ -446,7 +448,7 @@ class DuplicateFilesMixin:
         job hashes each file again right before it unlinks it.
         """
         listed = self._listed_sources()
-        targets, kept, protected = [], [], {}
+        targets, kept, protected, held = [], [], {}, []
         for group, keepers in selection.values():
             removals = [r for r in group["files"] if r not in keepers]
             for record in removals:
@@ -467,6 +469,7 @@ 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:
@@ -488,6 +491,7 @@ class DuplicateFilesMixin:
         return {
             "targets": targets,
             "kept": kept,
+            "held": held,
             "fingerprint": _fingerprint(targets),
             "public": _public_plan(targets, protected, records),
         }
@@ -584,6 +588,13 @@ 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,6 +70,18 @@ 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,15 +288,16 @@ 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(
-                {
-                    "path": path,
-                    "root": str(row.get("root") or ""),
-                    "root_index": index,
-                    "detail": str(row.get("detail") or ""),
-                }
-            )
+            rows.append(shaped)
         self.log_line(log_line)
 
     def record_counted(self, group, count, rows=(), *, complete=False):

py/api/log_cleanup.py

@@ -68,6 +68,23 @@ 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")
@@ -129,7 +146,7 @@ class LogCleanupMixin:
             token = store.issue(
                 ROTATED_LOGS,
                 json.dumps(files, sort_keys=True),
-                {"files": files, "protected": protected},
+                {"files": files, "protected": protected, "kept": _kept_files(rows)},
                 (),
             )
         except ScopeTooLarge:
@@ -198,7 +215,11 @@ class LogCleanupMixin:
             job.record(
                 "errors", {"path": name, "detail": detail}, f"! {name} · {detail}"
             )
-        job.record_counted("protected", plan.payload["protected"])
+        kept = [
+            {"path": row["name"], "detail": KEPT_DETAILS[row["kind"]]}
+            for row in plan.payload["kept"]
+        ]
+        job.record_counted("protected", len(kept), kept, complete=True)
         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,19 +697,13 @@ class MaintenanceRoutesMixin:
             rows,
             complete=plan.operation in COMPLETE_RECORD,
         )
-        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)
+        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:
-            job.record_counted("protected", payload["protected"])
+            # #550, #586: each kept record with its reason, like the changed
+            # ones, and every one of them in the report.
+            job.record_counted("protected", len(kept), kept, complete=True)
         line = _changed_line(
             plan.operation, result["changed"], result.get("marks", 0), result
         )
@@ -717,6 +711,47 @@ 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,6 +59,9 @@ 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"
@@ -177,7 +180,19 @@ class WorkflowStorageRoutesMixin:
             protected += batch["protected"]
             errors += len(batch["errors"])
             job.record_counted("changed", batch["rows"])
-            job.record_counted("protected", batch["protected"])
+            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,
+            )
             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,6 +452,40 @@ 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,6 +266,30 @@ 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

@@ -0,0 +1,140 @@
+"""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:
-            conn.execute(
+            image_id = 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,6 +377,13 @@ 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,6 +506,20 @@ 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_lora_import.py

@@ -1061,17 +1061,78 @@ class LoraImportRefusalTests(LoraImportAPITestCase):
             job = await self._failed_job()
         self.assertEqual(job.scope["failure"], "backup-failed")
 
-    async def test_cancel_right_after_the_start_writes_nothing(self):
+    async def test_cancel_while_it_stages_writes_nothing(self):
+        # #575: held at a download, so the cancel lands while it stages.
         self._two_loras()
         before = _snapshot(self.api.db)
-        resp, plan = await self._post(PREVIEW)
-        resp, body = await self._post(APPLY, {"token": plan["token"]})
-        job = self.api._jobs.get(body["job"]["id"])
-        self.api._jobs.cancel(job.id)  # at once, while it stages
+        self.gate = threading.Event()
+        _plan, job = await self._apply_reviewed()
+        loop = asyncio.get_running_loop()
+        self.assertTrue(await loop.run_in_executor(None, self.downloading.wait, 10))
+        self.assertTrue(self.api._jobs.cancel(job.id))
+        self.gate.set()
         await job.task
         self.assertEqual(job.state, "cancelled")
         self.assertEqual(_snapshot(self.api.db), before)
 
+    async def test_a_cancel_accepted_as_staging_ends_writes_nothing(self):
+        # #575: staging already passed its last cancel check and returned; a
+        # cancel accepted before the job resumes still stops it before the
+        # commit, so a request answered "cancellation_requested" holds.
+        self._two_loras()
+        before = _snapshot(self.api.db)
+        staged, release = threading.Event(), threading.Event()
+        real_stage = lora_import.stage
+
+        def held_stage(*args, **kwargs):
+            result = real_stage(*args, **kwargs)
+            staged.set()
+            release.wait(10)
+            return result
+
+        loop = asyncio.get_running_loop()
+        with patch.object(lora_import, "stage", held_stage):
+            _plan, job = await self._apply_reviewed()
+            try:
+                self.assertTrue(await loop.run_in_executor(None, staged.wait, 10))
+                self.assertTrue(self.api._jobs.cancel(job.id))
+            finally:
+                release.set()
+            await job.task
+        self.assertEqual(job.state, "cancelled", job.log)
+        self.assertEqual(_snapshot(self.api.db), before)
+
+    async def test_a_cancel_after_the_commit_point_is_refused(self):
+        # #575: once staging is done the job is no longer cancellable; a
+        # late cancel is refused (the route answers it as not requested) and
+        # the job truthfully ends "finished" with the import written.
+        self._two_loras()
+        before = _snapshot(self.api.db)
+        replacing, release = threading.Event(), threading.Event()
+        real_replace = self.api.db.replace_category_prompts
+
+        def held_replace(*args, **kwargs):
+            replacing.set()
+            release.wait(10)
+            return real_replace(*args, **kwargs)
+
+        loop = asyncio.get_running_loop()
+        with patch.object(self.api.db, "replace_category_prompts", held_replace):
+            _plan, job = await self._apply_reviewed()
+            try:
+                self.assertTrue(await loop.run_in_executor(None, replacing.wait, 10))
+                self.assertFalse(job.summary()["cancellable"])
+                resp = await self.client.request(
+                    "POST", f"/prompt_manager/jobs/{job.id}/cancel"
+                )
+                body = json.loads(await resp.text())
+            finally:
+                release.set()
+            await job.task
+        self.assertFalse(body["cancellation_requested"])
+        self.assertEqual(job.state, "finished", job.log)
+        self.assertNotEqual(_snapshot(self.api.db), before)
+
 
 def plan_title():
     return lora_import.JOB_TITLE

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;