refactor(thumbnails): keep the after-scan queue's store rule and pin the write lock (#320)

A planned target keeps the plan's own containment rule (inside the
store, as the thumbnail job writes it); captures keep the serve rule.
The structural test also covers PREVIEW_FOLDER joins and the
repo-root modules; the write-lock test asserts the lock instead of
waiting on a timer; new tests cover nested roots in the after-scan
queue and a legacy upper-case preview served through the resolver.

TESTING.md

@@ -349,9 +349,12 @@ and the time left. The steps, in order:
    observer join get what is left, which can be 0 s. `stop()` joins watchdog's
    emitters without a limit, so it runs on a daemon thread and is waited for
    only until the deadline; a later stop waits for that same call (#561).
-3. `close_auto_thumbnails(final=True)` stops the preview worker (30 s) and
-   kills ffmpeg.
-4. `PromptTracker.shutdown` stops the prompt cleanup thread (2 s).
+3. `close_auto_thumbnails(final=True)` stops the preview worker (30 s),
+   kills ffmpeg and shuts the interactive image pool down.
+4. `shutdown_job_pool` (`utils/parallel.py`) cancels the long-lived job pool's
+   queued items and joins its threads (5 s; part `workers`, only when a job
+   used the pool, #523).
+5. `PromptTracker.shutdown` stops the prompt cleanup thread (2 s).
 
 A failing step, or a second Ctrl+C, is logged with its traceback and does not
 skip the later steps. A clean run logs one INFO line to PromptManager's log:
@@ -469,6 +472,25 @@ differ are files whose pixel data is undecodable or cut short: legacy found
 nothing, the walk keeps the text before the pixels. Animated PNGs still use
 `.text`.
 
+Image work runs on PromptManager's own threads (#523). glibc keeps each thread's
+malloc arena at its high-water mark, so image buffers spread over the default
+executor's threads stay resident. Interactive image work (view previews behind their
+gate, workspace card facts, both inspector routes) runs on the API's fixed
+`pm-image` pool (`IMAGE_POOL_SIZE` 3, `_run_image`, `_map_image`; card facts already
+cached answer without a trip; while the gate's 3 preview writes hold every thread, a
+card or inspector read waits for one of them, by design of the merged pool; after the
+package teardown the pool is not recreated and late calls use the default executor). Jobs (scan metadata and file facts, thumbnail and
+convert jobs, the legacy gallery generation, capture previews) run through
+`map_parallel` on one long-lived `pm-worker` pool of Worker threads, reused between
+jobs; a new Worker threads value replaces it (the teardown also joins a replaced pool
+still finishing), and a call made from a job thread runs inline. After a scan, thumbnail or convert job, or the legacy gallery generation,
+`release_free_memory` (`utils/native_memory.py`) calls glibc's `malloc_trim(0)` once;
+it does nothing on Windows, macOS and musl and never raises. No allocator setting
+changes. `tests/test_image_pools.py` asserts the thread of each read and write and the
+trim; `tests/test_utils_parallel.py` the pool's reuse, resize, nesting and shutdown.
+`tools/perf/memory_drive.py` (manual, Linux, never in CI) measures RSS, heap, threads
+and arenas under the #523 load against the disposable browser test server.
+
 The `perf budget` job in `.github/workflows/test.yml` is the reference
 measurement: `ubuntu-latest`, one Python, no coverage, only the wall-clock tests.
 It runs on every push to `main` and on `workflow_dispatch`, never on pull
@@ -3871,8 +3893,8 @@ workspace never asks; previews are a cache the server keeps:
 - `GET /images/serve/thumbnails/<rel>_thumb<ext>` for a missing preview is refused as
   text first (traversal, a backslash, a drive or stream colon, empty or dot segments, so
   `D:/x`, `\\server\share` and a leading `/` never reach a joined path), then admitted
-  on the event loop (`_PreviewGate`): 3 previews are written at once on the API's own
-  `pm-preview` pool (never the default executor), one flight runs per file (the history
+  on the event loop (`_PreviewGate`): 3 previews are written at once on the API's
+  interactive `pm-image` pool (never the default executor; #523), one flight runs per file (the history
   route shares the flight key and the admission), and at most 64 requests wait; one
   more gets its fallback at once without listing a folder. Inside the admission the
   preview maps back to its media file under the same root (exact names, then one folder
@@ -3905,7 +3927,7 @@ workspace never asks; previews are a cache the server keeps:
   `exclusive` job runs (`_auto_thumbnail_blocker`), checked when a write starts (after
   its slot) and again just before its rename.
 - Shutdown is bounded and logged: the worker stops at a file boundary, ffmpeg is
-  killed, an idle worker leaves (the next output starts a fresh one), the view pool
+  killed, an idle worker leaves (the next output starts a fresh one), the image pool
   shuts down, and after the app's cleanup a late monitor event
   starts nothing. A package reload stops the workers the previous import left.
 - Cards and strips load previews only: Images in every layout and size, Library cards,
@@ -4382,6 +4404,23 @@ the per-section hand-off panel, was retired by S12 (#329): every section is nati
   choice. The host's checks still refuse a linked field or a changed workflow, and the
   sender also refuses if the destination's `revision` (node object plus graph version)
   moved since discovery; it is undefined on an older tab, which skips the comparison. A host error reads as a refusal, not "may have applied".
+- **D191** (#542, beyond the concept; maintainer 2026-10-31, Q542-1 A to 5 A): a window with two or more
+  eligible nodes adds a last Destination node option "All N eligible nodes" (`ALL_NODES` in
+  `prompt-sender.mjs`), never selected for you (D173). Eligible is D173's filter; the nodes it leaves out
+  stay counted in the discovery sentence. One window only. Preview stays optional (D175): Preview changes
+  shows one `section.prompt-sender-node` per node, headed "#<id> <title>", in send order (node id,
+  numeric). Apply previews every node first and writes nothing if any refuses ("#<id> <title>: <reason>
+  Nothing was sent."). Then it sends one node after another. Only the first node of a pass taken in that
+  click keeps its ticket; every other node, and every node of a held Preview (host tickets expire after
+  30 s), takes a fresh ticket that is refused unless its Before texts and composition equal the reviewed
+  ones. That fresh preview checks the discovered revision until something has been posted, and not after,
+  since our own write may move the graph version. Refusals are best effort: "Sent to 2 of 3 nodes in
+  <window>. #12 <title>: <host's refusal>."; Apply stays available and a retry sends to every node again
+  (rewriting the same text). A lost reply stops the run, names that node as unknown and lists the nodes
+  not sent. Closing the dialog stops after the current node. The review sections are labelled by their
+  `h4` node heading. Unit tests in `tests/js/prompt-sender.test.js`, a fake three-node host in
+  `tests/browser/prompt-inspector.spec.mjs`, real Classic and Nodes 2.0 widgets in
+  `tests/comfyui/prompt-sender.spec.mjs`.
 - **D106** (Integrations, S10, Q8): the send bridge always asks for the ComfyUI window
   and node, and the nodes always follow ComfyUI's own renderer, so "Default send
   destination" offers only "Ask each time" and "Host compatibility" only "Follow host

__init__.py

@@ -280,11 +280,11 @@ def _teardown_logger():
 
 
 def _stop_background_work():
-    """Stop the image monitor, the preview worker and the prompt cleanup
-    thread against one TEARDOWN_SECONDS deadline, once.
+    """Stop the image monitor, the preview worker, the job worker threads
+    and the prompt cleanup thread against one TEARDOWN_SECONDS deadline, once.
 
-    Returns ``{"monitor", "thumbnails", "tracker"}`` booleans (True: it
-    stopped), leaving out parts that never started. Later calls return a
+    Returns ``{"monitor", "thumbnails", "workers", "tracker"}`` booleans
+    (True: it stopped), leaving out parts that never started. Later calls return a
     copy of the first result. A Ctrl+C inside a step is logged and the
     step is False; one between steps is re-raised after the result (parts
     not reached are False) is stored and the ``atexit`` entry removed.
@@ -298,11 +298,14 @@ def _stop_background_work():
             api,
             _global_prompt_tracker,
         )
+        from .utils import parallel
+
         # A part whose global is set has started; False until its step says
         # otherwise, so an interrupt between steps leaves an honest result.
         started = (
             ("monitor", monitor),
             ("thumbnails", thumbnails),
+            ("workers", parallel.job_pool_started() or None),
             ("tracker", tracker),
         )
         result = {name: False for name, part in started if part is not None}
@@ -345,6 +348,16 @@ def _stop_background_work():
                 result["thumbnails"] = _teardown_step(
                     "the preview worker", close_previews, logger
                 )
+            if "workers" in result:
+
+                def stop_workers():
+                    from .utils.parallel import JOB_POOL_JOIN_SECONDS
+
+                    return parallel.shutdown_job_pool(left(JOB_POOL_JOIN_SECONDS))
+
+                result["workers"] = _teardown_step(
+                    "the job worker threads", stop_workers, logger
+                )
             if tracker is not None:
 
                 def stop_tracker():

py/api/__init__.py

@@ -85,6 +85,7 @@ try:
     from ...database.operations import PromptDatabase
     from ...utils.file_prompt import parse_comfyui_prompt
     from ...utils.logging_config import get_logger
+    from ...utils.native_memory import release_free_memory
     from ...utils.png_text import png_text
     from ...utils.video_metadata import is_video_path, read_video_metadata
 except ImportError:
@@ -96,6 +97,7 @@ except ImportError:
     from database.operations import PromptDatabase
     from utils.file_prompt import parse_comfyui_prompt
     from utils.logging_config import get_logger
+    from utils.native_memory import release_free_memory
     from utils.png_text import png_text
     from utils.video_metadata import is_video_path, read_video_metadata
 
@@ -528,6 +530,14 @@ class PromptManagerAPI(
             return await loop.run_in_executor(None, call)
         return await loop.run_in_executor(None, func, *args)
 
+    async def _release_job_memory(self):
+        """After a scan or thumbnail job: glibc's free heap pages go back to
+        the OS once (``malloc_trim(0)``; nothing elsewhere, #523)."""
+        try:
+            await self._run_in_executor(release_free_memory)
+        except Exception:  # best effort: the job's result already stands
+            self.logger.debug("Releasing free memory failed", exc_info=True)
+
     def _root_listing(self, root) -> RootListing:
         """The shared listing record of one output root (created empty)."""
         return self._root_listings.setdefault(str(root), RootListing())

py/api/admin.py

@@ -1454,6 +1454,7 @@ class AdminRoutesMixin:
             verb = "cancelled" if state == "cancelled" else "finished"
             job.log_line(f"Scan {verb} · {_plural(job.processed, 'file')} processed")
         job.finish(state)
+        await self._release_job_memory()
         if state == "finished":
             # After the scan, so the guarded thumbnail job is admitted (#249).
             await self._hand_off_scan_thumbnails(job, output_dirs)
@@ -1668,7 +1669,8 @@ class AdminRoutesMixin:
     def _extract_batch_metadata_sync(self, file_batch):
         """Extract metadata from a batch of files in one executor call.
 
-        Files are read on ``PromptManagerConfig.WORKER_THREADS`` threads.
+        Files are read on the long-lived job pool of
+        ``PromptManagerConfig.WORKER_THREADS`` threads (#523).
         PNG text is read chunk by chunk without decoding pixels (#523); file
         reads and ffprobe release the GIL.
         """

py/api/autotag_routes.py

@@ -20,7 +20,7 @@ from .jobs import (
 
 from .maintenance import disk_plan
 from .prompts import thumbnail_url_for
-from . import thumbnail_settings
+from . import preview_store, thumbnail_settings
 from .. import model_gate
 
 try:
@@ -1371,7 +1371,8 @@ class AutotagRoutesMixin:
             files = await self._run_in_executor(
                 self._without_removed, output_path, files
             )
-            has_thumbnails = (output_path / "thumbnails").is_dir()
+            store = preview_store.store_dir(output_path)
+            has_thumbnails = store is not None and store.is_dir()
 
             images = []
             for image_path in files:

py/api/history.py

@@ -843,7 +843,10 @@ class HistoryRoutesMixin:
                 raise
 
         async def execute():
-            pending = asyncio.create_task(self._run_in_executor(project))
+            # An inspection reads the file's header and text: the
+            # interactive image pool (#523); media stays on the default.
+            run = self._run_image if kind == "inspection" else self._run_in_executor
+            pending = asyncio.create_task(run(project))
             try:
                 return await asyncio.shield(pending)
             except asyncio.CancelledError:

py/api/history_current.py

@@ -15,10 +15,10 @@ from typing import BinaryIO
 
 from aiohttp import web
 
-from . import thumbnail_settings
+from . import preview_store, thumbnail_settings
 from .images import _canonical, _public_path, SERVABLE_EXTENSIONS, VIDEO_EXTENSIONS
 from .images import in_trash_below_roots
-from .images import existing_media_thumbnail, media_thumbnail_rel
+from .images import existing_media_thumbnail, media_preview_path
 
 try:
     from ...utils.execution_outputs import (
@@ -326,12 +326,10 @@ class _OpenedCurrentMedia:
 
 
 def _preview_writable(root: Path, preview: Path) -> bool:
-    """A preview may be written: ``thumbnails`` and every link segment of the
-    preview's folder resolve inside *root* (the serve route's rule, #249)."""
-    folder = root / "thumbnails"
-    return _canonical(folder).is_relative_to(_canonical(root)) and _canonical(
-        preview
-    ).is_relative_to(_canonical(folder))
+    """A preview may be written: the store of *root* and every link segment
+    of the preview's folder resolve inside it (the serve route's rule, #249;
+    ``preview_store.owns``)."""
+    return preview_store.owns(root, preview)
 
 
 def _is_jxl(path) -> bool:
@@ -381,8 +379,12 @@ def _cover_target(resolved, ensure_preview, placeholder, accepts_jxl):
         return _canonical(preview)
     if preview is not None:
         return _canonical(preview)
-    preview = resolved.gallery_root / media_thumbnail_rel(relative)
-    if ensure_preview is not None and _preview_writable(resolved.gallery_root, preview):
+    preview = media_preview_path(resolved.gallery_root, relative)
+    if (
+        ensure_preview is not None
+        and preview is not None
+        and _preview_writable(resolved.gallery_root, preview)
+    ):
         ensure_preview(target, preview, is_video)
     if is_video and placeholder is not None:
         return _placeholder_media(placeholder)

py/api/images.py

@@ -34,7 +34,7 @@ from .gallery_listing import (
     WalkOutcome,
     walk_root,
 )
-from . import thumbnail_settings
+from . import preview_store, thumbnail_settings
 from .thumbnail_encode import (
     PREVIEW_HELD,
     EncoderClosed,
@@ -1068,13 +1068,20 @@ def _iter_media_files(output_path, max_files):
 
 
 def media_thumbnail_rel(rel_path, settings=None):
-    """``thumbnails/<rel>_thumb<ext>`` (POSIX) for the media file *rel_path*:
-    the name the preview gets when it is written now (#317), images and
-    videos alike (``writer_settings``' extension)."""
+    """``thumbnails/<rel>_thumb<ext>`` (POSIX URL text) for the media file
+    *rel_path*: the name the preview gets when it is written now (#317),
+    images and videos alike (``writer_settings``' extension). Where it is on
+    disk is ``media_preview_path``."""
     settings = settings or writer_settings()
     return thumbnail_settings.thumbnail_rel(rel_path, settings.extension)
 
 
+def media_preview_path(output_path, rel_path, settings=None):
+    """Where the preview of *rel_path* is written now (``preview_store``)."""
+    settings = settings or writer_settings()
+    return preview_store.preview_path(output_path, rel_path, settings.extension)
+
+
 def existing_media_thumbnail(output_path, rel_path):
     """The existing preview *rel_path* owns that a listing or cover shows
     (the format written now first, then older ones), or None. The Images
@@ -2798,13 +2805,32 @@ class ImageRoutesMixin:
                     output_dirs[0], filepath, int(root_param), len(outputs)
                 )
 
+            # Preview text is mapped by the resolver (#320); any other text,
+            # ``Thumbnails/...`` media on a case-sensitive disk included,
+            # keeps the plain join.
+            preview_parts = (
+                tuple(filepath.split("/"))
+                if preview_store.is_preview_text(filepath)
+                else None
+            )
             for output_path in output_dirs:
-                image_path = output_path / filepath
+                if preview_parts is None:
+                    image_path = output_path / filepath
+                else:
+                    image_path = preview_store.request_path(output_path, preview_parts)
+                    if image_path is None:
+                        continue
                 if not image_path.exists():
                     continue
-                # A symlink that points outside the root is a hard deny, not
-                # a fall-through to the next root.
-                if not _is_within(image_path, output_path):
+                # A symlink that points outside the root (a preview: outside
+                # the folder holding its store) is a hard deny, not a
+                # fall-through to the next root.
+                container = (
+                    output_path
+                    if preview_parts is None
+                    else preview_store.store_container(output_path)
+                )
+                if not _is_within(image_path, container):
                     return _forbidden()
                 if in_trash_below_roots(image_path, output_dirs):
                     return _file_not_found()
@@ -2975,6 +3001,7 @@ class ImageRoutesMixin:
             result = await self._run_in_executor(
                 self._generate_gallery_thumbnails_sync, output_dirs
             )
+            await self._release_job_memory()
             return web.json_response(result)
 
         except Exception as e:
@@ -2987,6 +3014,7 @@ class ImageRoutesMixin:
         """Blocking thumbnail generation loop (run in executor)."""
         settings = writer_settings(settings)
         start = _time.monotonic()
+        thumbnails_dir = preview_store.given_store(output_path, thumbnails_dir)
         plan = self._plan_thumbnails(output_path, thumbnails_dir, settings=settings)
         payload = self._execute_thumbnail_plan(plan, settings, start)
         payload["thumbnails_path"] = _public_path(thumbnails_dir)
@@ -2999,7 +3027,8 @@ class ImageRoutesMixin:
         start = _time.monotonic()
         plan = self._plan_gallery_thumbnails(output_dirs, settings=settings)
         payload = self._execute_thumbnail_plan(plan, settings, start)
-        payload["thumbnails_path"] = _public_path(output_dirs[0] / "thumbnails")
+        store = preview_store.store_dir(output_dirs[0])
+        payload["thumbnails_path"] = None if store is None else _public_path(store)
         return payload
 
     def _execute_thumbnail_plan(self, plan: dict, settings, start: float) -> dict:
@@ -3042,9 +3071,12 @@ class ImageRoutesMixin:
             if stop is not None and stop():
                 break
             root = Path(root)
+            store = preview_store.store_dir(root)
+            if store is None:
+                continue
             plan = self._plan_thumbnails(
                 root,
-                root / "thumbnails",
+                store,
                 mode=mode,
                 stop=stop,
                 settings=settings,
@@ -3128,7 +3160,7 @@ class ImageRoutesMixin:
         """
         settings = writer_settings(settings)
         output_path = Path(output_path)
-        thumbnails_dir = Path(thumbnails_dir)
+        thumbnails_dir = preview_store.given_store(output_path, thumbnails_dir)
         plan = {key: 0 for key in _PLAN_COUNTS}
         plan.update(targets=[], fresh_files=[])
         folders = {}
@@ -3220,6 +3252,9 @@ class ImageRoutesMixin:
             "dir": src.parent.name,
             "type": "video" if is_video else "image",
         }
+        # Taken when the write starts: a thumbnail location switched since
+        # then drops the write at its rename (#320).
+        guard = preview_store.write_guard(guard)
         dst = Path(dst)
         if dst.suffix != settings.extension:
             dst = dst.with_suffix(settings.extension)
@@ -3500,8 +3535,8 @@ class ImageRoutesMixin:
             self.logger.debug(f"No roots to sweep for preview temp files: {exc}")
             return 0
         for output_path in output_dirs:
-            folder = Path(output_path) / "thumbnails"
-            if folder.is_symlink() or not folder.is_dir():
+            folder = preview_store.store_dir(output_path)
+            if folder is None or folder.is_symlink() or not folder.is_dir():
                 continue
             removed += sweep_stray_temps(folder)[0]
         if removed:
@@ -3519,13 +3554,16 @@ class ImageRoutesMixin:
         directories = []
         for output_path in output_dirs:
             output_path = Path(output_path)
-            thumbnails_dir = output_path / "thumbnails"
+            thumbnails_dir = preview_store.store_dir(output_path)
+            if thumbnails_dir is None:
+                continue
             if not thumbnails_dir.exists() and not thumbnails_dir.is_symlink():
                 continue
+            container = preview_store.store_container(output_path)
             if (
                 thumbnails_dir.is_symlink()
-                or not _is_within(thumbnails_dir, output_path)
-                or _canonical(thumbnails_dir) == _canonical(output_path)
+                or not _is_within(thumbnails_dir, container)
+                or _canonical(thumbnails_dir) == _canonical(container)
             ):
                 raise ValueError(
                     "Safety check failed: invalid thumbnails directory path"
@@ -3997,7 +4035,8 @@ class ImageRoutesMixin:
                     **observed,
                 )
 
-            body = await self._run_in_executor(inspect)
+            # Header and text reads: the interactive image pool (#523).
+            body = await self._run_image(inspect)
             if body is None:
                 return web.json_response(
                     {
@@ -4210,6 +4249,22 @@ class ImageRoutesMixin:
         # (which Windows refuses on open files) is not held up by it.
         return _CardFacts(width, height, *self._card_parse(path, records))
 
+    def _cached_card_facts(self, entry: tuple) -> _CardFacts | None:
+        path, mtime, root, index = entry[:4]
+        return self._workspace_card_cache.get((index, str(root)), {}).get((path, mtime))
+
+    def _card_facts_of(self, entries: list, *, parse: bool) -> list[_CardFacts]:
+        """Card facts of *entries*, in order; files not cached yet are read on
+        the interactive image pool (#523), cached ones answer here."""
+        found = [self._cached_card_facts(entry) for entry in entries]
+        missing = [i for i, facts in enumerate(found) if facts is None]
+        read = self._map_image(
+            lambda i: self._card_facts(entries[i], parse=parse), missing
+        )
+        for i, facts in zip(missing, read):
+            found[i] = facts
+        return found
+
     def _card_facts(self, entry: tuple, *, parse: bool) -> _CardFacts:
         """One listed image's card facts, read once per ``(path, mtime)``.
 
@@ -4218,11 +4273,11 @@ class ImageRoutesMixin:
         and drops the rest, and a roots change drops them all. Only files of
         the current scan are stored, so the cache is bounded by the listing.
         """
-        path, mtime, root, index = entry[:4]
-        root_key, key = (index, str(root)), (path, mtime)
-        cached = self._workspace_card_cache.get(root_key, {}).get(key)
+        cached = self._cached_card_facts(entry)
         if cached is not None:
             return cached
+        path, mtime, root, index = entry[:4]
+        root_key, key = (index, str(root)), (path, mtime)
         facts = self._read_card_facts(path, parse=parse)
         if facts is None:
             # Not stored: a locked or half-written file may read next time.
@@ -4248,15 +4303,17 @@ class ImageRoutesMixin:
             if linked
             else {}
         )
+        # Videos: the card reads no video metadata. Library previews show
+        # no card facts, so they skip the parse (``parse=linked``).
+        stills = [image for image in images if not image["is_video"]]
+        read = self._card_facts_of(
+            [entries[(image["root_index"], image["path"])] for image in stills],
+            parse=linked,
+        )
+        facts_of = {id(image): facts for image, facts in zip(stills, read)}
         for image in images:
             entry = entries[(image["root_index"], image["path"])]
-            # Videos: the card reads no video metadata. Library previews show
-            # no card facts, so they skip the parse (``parse=linked``).
-            facts = (
-                _NO_CARD_FACTS
-                if image["is_video"]
-                else self._card_facts(entry, parse=linked)
-            )
+            facts = facts_of.get(id(image), _NO_CARD_FACTS)
             image["width"], image["height"] = facts.width, facts.height
             if not linked:
                 continue

py/api/preview_store.py

@@ -0,0 +1,206 @@
+"""Where previews live on disk (#320): the one resolver.
+
+Every mapping between an image source (a "root"), one of its media files and
+that file's preview on disk goes through this module. URL text stays in
+``thumbnail_settings`` (``thumbnail_rel``: ``thumbnails/<rel>_thumb<ext>``,
+the shape the serve route, ``inspection-state.mjs`` and ``admin.js`` share);
+this module answers where such a preview is on disk.
+
+The store of a source is where its previews live. With the default location
+(mode ``source``, "Inside each source") it is ``<source>/thumbnails``, exactly
+the folder every caller joined by hand before, so nothing moves.
+
+Writes and the switch: every preview writer takes ``write_guard`` when it
+starts (it records the location's ``generation``); the final rename runs
+under ``write_lock`` and only when the generation is unchanged and no hold
+applies (``thumbnail_encode.replace_unless_held``). Changing the location
+bumps the generation under the same lock, so no write that began before can
+land in the old store afterwards. Nothing reads or commits config while
+``write_lock`` is held. The hold check runs under it too, so hold checks take
+no lock (the job registry's ``running()`` is a lock-free snapshot) and
+``bump_generation`` is never called with a registry or config lock held.
+
+All functions are blocking: call them off the event loop.
+"""
+
+import os
+import threading
+from contextlib import contextmanager
+from dataclasses import dataclass
+from pathlib import Path
+
+from . import thumbnail_settings
+
+try:
+    from ...utils.media_folders import PREVIEW_FOLDER
+except ImportError:  # tests import the package top-level
+    from utils.media_folders import PREVIEW_FOLDER
+
+SOURCE_MODE = "source"
+
+
+@dataclass(frozen=True)
+class Location:
+    """One snapshot of the thumbnail location (take one per request or job)."""
+
+    mode: str = SOURCE_MODE
+    path: str | None = None
+    available: bool = True
+    problem: str | None = None
+    generation: int = 0
+
+
+write_lock = threading.Lock()
+_generation = 0
+
+
+def current():
+    """The thumbnail location now."""
+    return Location(generation=_generation)
+
+
+def generation():
+    return _generation
+
+
+def bump_generation():
+    """Invalidate every write that started before (under ``write_lock``)."""
+    global _generation
+    with write_lock:
+        _generation += 1
+        return _generation
+
+
+def _canonical(path):
+    """Resolved and case-folded, so paths compare on every OS."""
+    return Path(os.path.normcase(os.path.realpath(str(path))))
+
+
+def _within(path, folder):
+    return _canonical(path).is_relative_to(_canonical(folder))
+
+
+# ── stores ───────────────────────────────────────────────────────────
+
+
+def store_container(root, location=None):
+    """The folder a store of *root* must stay inside: the source itself."""
+    return Path(root)
+
+
+def store_dir(root, location=None):
+    """The store of *root* (``<source>/thumbnails``)."""
+    return Path(root) / PREVIEW_FOLDER
+
+
+def ensure_store(root, location=None):
+    """``store_dir`` for a writer. Preview folders below it are made by the
+    write itself."""
+    return store_dir(root, location)
+
+
+def given_store(root, store):
+    """A caller's explicit store argument: the default spelling
+    (``<root>/thumbnails``) means the resolver's store; another folder (only
+    tests pass one) is used as given."""
+    store = Path(store)
+    if store == Path(root) / PREVIEW_FOLDER:
+        return store_dir(root)
+    return store
+
+
+def preview_path(root, media_rel, ext=None, location=None):
+    """Where the preview of *media_rel* (relative to *root*) is, or is to be
+    written, in *ext* (the extension written now when None)."""
+    store = store_dir(root, location)
+    if store is None:
+        return None
+    if ext is None:
+        from .thumbnail_encode import writer_settings
+
+        ext = writer_settings().extension
+    return store / (thumbnail_settings.preview_stem(media_rel) + ext)
+
+
+def request_path(root, parts, location=None):
+    """The disk path of the URL parts ``("thumbnails", ...)``, or None when
+    *parts* is not under an exact ``thumbnails`` first part. The text is
+    checked before (``is_preview_text``)."""
+    parts = tuple(parts)
+    if len(parts) < 2 or parts[0] != PREVIEW_FOLDER:
+        return None
+    store = store_dir(root, location)
+    if store is None:
+        return None
+    return store.joinpath(*parts[1:])
+
+
+def owns(root, path, location=None):
+    """Whether *path* lies inside the store of *root*, resolved: the store
+    stays inside its container and no link takes *path* out of the store."""
+    store = store_dir(root, location)
+    if store is None:
+        return False
+    return _within(store, store_container(root, location)) and _within(path, store)
+
+
+# ── preview text ─────────────────────────────────────────────────────
+
+
+def is_preview_text(filepath):
+    """URL text naming a preview: ``thumbnails/<...>/<name>_thumb<ext>``.
+
+    Text only, before any join: no backslash, drive or stream colon, and no
+    empty, ``.`` or ``..`` segment; ``thumbnails`` exactly as the first part
+    and nowhere else among the folders; the extension, in any case, one a
+    preview may carry (``X_thumb.JPG`` of an older release included).
+    """
+    if not filepath or "\\" in filepath or ":" in filepath:
+        return False
+    parts = filepath.split("/")
+    if len(parts) < 2 or parts[0] != PREVIEW_FOLDER:
+        return False
+    if any(part in ("", ".", "..") for part in parts):
+        return False
+    if PREVIEW_FOLDER in parts[1:-1]:
+        return False
+    stem, extension = os.path.splitext(parts[-1])
+    return (
+        stem.endswith(thumbnail_settings.THUMB_SUFFIX)
+        and len(stem) > len(thumbnail_settings.THUMB_SUFFIX)
+        and extension.lower() in thumbnail_settings.THUMBNAIL_EXTENSIONS
+    )
+
+
+# ── writes ───────────────────────────────────────────────────────────
+
+
+class _WriteGuard:
+    """``guard()`` for one preview write: true while the location has not
+    changed since the write started and *hold* (if any) allows it."""
+
+    __slots__ = ("generation", "_hold")
+
+    def __init__(self, hold=None):
+        self.generation = _generation
+        self._hold = hold
+
+    def __call__(self):
+        if self.generation != _generation:
+            return False
+        return self._hold is None or bool(self._hold())
+
+
+def write_guard(hold=None):
+    """Take when a preview write starts; hand to its final rename."""
+    if isinstance(hold, _WriteGuard):
+        return hold
+    return _WriteGuard(hold)
+
+
+@contextmanager
+def committed(guard=None):
+    """Around a preview's final rename: holds ``write_lock`` and yields
+    whether the write may land (``guard()``, re-checked under the lock)."""
+    with write_lock:
+        yield guard is None or bool(guard())

py/api/thumbnail_auto.py

@@ -8,7 +8,8 @@ previews behind those URLs are made here, in three ways:
 - **On first view.** ``GET /prompt_manager/images/serve/thumbnails/<rel>_thumb
   <ext>`` for a preview that does not exist yet is admitted on the event loop
   (``_PreviewGate``): at most ``ON_DEMAND_LIMIT`` previews are written at once,
-  on this API's own small pool (never the default executor), one flight runs
+  on this API's interactive image pool (``IMAGE_POOL_SIZE`` threads, shared
+  with card facts and inspector reads; never the default executor), one flight runs
   per file, and the requests waiting are capped (``MAX_WAITING``); past the
   cap a request gets its fallback at once. Inside the admission the preview is
   mapped back to its media file under the same root (the request text is
@@ -24,7 +25,8 @@ previews behind those URLs are made here, in three ways:
 - **On capture.** The image monitor hands every new output it settles to
   ``queue_capture_thumbnail``, which queues its preview for one background
   worker thread (``AutoThumbnailer``). Each wake takes at most Worker threads
-  (``WORKER_THREADS``) files and writes them on that many threads; enqueuing
+  (``WORKER_THREADS``) files and writes them on the long-lived job pool
+  (``utils.parallel``, shared with scans and thumbnail jobs); enqueuing
   is O(1), so the monitor and ComfyUI's queue never wait. A file already
   queued is not queued again, the queue is bounded (``MAX_QUEUED``) and a
   preview left out is made on its first view instead.
@@ -49,7 +51,7 @@ write starts (after its slot) and again just before its rename.
 
 Shutdown: ``close_auto_thumbnails`` stops the worker at a file boundary,
 kills a running ffmpeg, waits a bounded time (and logs when the wait ran
-out) and shuts the view pool down; after the app's cleanup
+out) and shuts the image pool down; after the app's cleanup
 (``final=True``) nothing starts a worker again. Importing this module again
 (a package reload) stops the workers the previous import left.
 """
@@ -60,12 +62,12 @@ import functools
 import io
 import os
 import threading
-from concurrent.futures import ThreadPoolExecutor
+from concurrent.futures import CancelledError, ThreadPoolExecutor
 from pathlib import Path
 
 from aiohttp import web
 
-from . import thumbnail_settings
+from . import preview_store, thumbnail_settings
 from .images import (
     in_trash_below_roots,
     MEDIA_EXTENSIONS,
@@ -76,7 +78,7 @@ from .images import (
     _thumbnail_is_fresh,
     _video_placeholder_image,
     _worker_threads,
-    media_thumbnail_rel,
+    media_preview_path,
 )
 from .jobs import (
     DUPLICATE_CLEANUP_JOB_KIND,
@@ -104,8 +106,11 @@ PLACEHOLDER_SIZE = (thumbnail_settings.DEFAULT_EDGE, thumbnail_settings.DEFAULT_
 PLACEHOLDER_QUALITY = 85
 # Source extensions in the order a shared preview name looks for them (R2).
 _SOURCE_ORDER = thumbnail_settings.IMAGE_SOURCE_ORDER + tuple(sorted(VIDEO_EXTENSIONS))
-# Previews views may write at once (the size of the view pool).
-ON_DEMAND_LIMIT = 3
+# Threads of the interactive image pool (#523): view previews, card facts and
+# inspector reads. Fixed, so image buffers stay in a few threads' arenas.
+IMAGE_POOL_SIZE = 3
+# Previews views may write at once (the whole image pool).
+ON_DEMAND_LIMIT = IMAGE_POOL_SIZE
 # View requests that may wait on the gate; one more gets its fallback at once.
 MAX_WAITING = 64
 # How long a pool thread waits for the capture worker's flight of the same file.
@@ -124,7 +129,7 @@ AUTO_REFUSED_KINDS = frozenset(
     {RESTORE_JOB_KIND, DUPLICATE_CLEANUP_JOB_KIND, SOURCE_CHANGE_JOB_KIND}
 )
 WORKER_NAME = "PromptManagerThumbnailWorker"
-POOL_PREFIX = "pm-preview"
+POOL_PREFIX = "pm-image"
 # Identifies this import's workers: a reload stops the ones it did not start.
 _MODULE_TOKEN = object()
 
@@ -153,6 +158,36 @@ def _dst_key(dst):
     return flight_key(Path(dst).parent, (Path(dst).name,))
 
 
+class DeferredPreview:
+    """The preview of *rel* under *root*, resolved when its write starts
+    (#320): a queued capture follows the thumbnail location at write time,
+    not the one when it was queued."""
+
+    __slots__ = ("root", "rel", "strict")
+
+    def __init__(self, root, rel, *, strict=True):
+        self.root = Path(root)
+        self.rel = Path(rel)
+        # A capture refuses a store linked out of its container (the serve
+        # route's rule); a planned target keeps the plan's rule: inside the
+        # store, as the thumbnail job writes it.
+        self.strict = strict
+
+    def resolve(self):
+        """The preview's path now, or None when nothing may be written."""
+        dst = media_preview_path(self.root, self.rel)
+        if dst is None:
+            return None
+        if self.strict:
+            return dst if preview_store.owns(self.root, dst) else None
+        store = preview_store.store_dir(self.root)
+        return dst if _is_within(dst, store) else None
+
+
+def _resolved(dst):
+    return dst.resolve() if isinstance(dst, DeferredPreview) else Path(dst)
+
+
 @functools.lru_cache(maxsize=1)
 def video_placeholder_jpeg():
     """The placeholder frame as JPEG bytes (a video whose preview is missing)."""
@@ -210,10 +245,19 @@ def _candidates(parts):
 def _current_preview(root, parts):
     """The preview *parts* names, in the format written now (#317): a request
     for an older name (``_thumb.png`` before a conversion) is answered with
-    the current file."""
+    the current file. None when the store of *root* has no place for it."""
     _folder, base, _wanted = _candidates(parts)
     name = base + thumbnail_settings.THUMB_SUFFIX + writer_settings().extension
-    return Path(root).joinpath(*parts[:-1], name)
+    return preview_store.request_path(root, (*parts[:-1], name))
+
+
+def _request_key(root, parts):
+    """The flight key of the preview *parts* under *root*: the key of its
+    disk path (so the serve and history routes share one flight)."""
+    path = preview_store.request_path(root, parts)
+    if path is None:
+        return flight_key(root, parts)
+    return _dst_key(path)
 
 
 def _quick_source(root, parts):
@@ -254,12 +298,10 @@ def _thumbnail_source(root, parts):
     if in_trash_below_roots(found, [root]):
         return None
     rel = Path(*folder_parts, found.name)
-    dst = root / media_thumbnail_rel(rel)
-    if dst != _current_preview(root, parts):
+    dst = media_preview_path(root, rel)
+    if dst is None or dst != _current_preview(root, parts):
         return None
-    if not _is_within(root / "thumbnails", root) or not _is_within(
-        dst, root / "thumbnails"
-    ):
+    if not preview_store.owns(root, dst):
         dst = None
     return found, dst, found.suffix.lower() in VIDEO_EXTENSIONS
 
@@ -279,6 +321,21 @@ def _scan_folder(folder, base, extensions):
     return None
 
 
+def _deferred(roots, src, dst):
+    """The ``DeferredPreview`` of a planned ``(src, dst)``: the root whose
+    store the plan wrote *dst* into (nested roots plan one each); *dst* as
+    planned when none matches."""
+    src = Path(src)
+    for root in roots:
+        root = Path(root)
+        if not src.is_relative_to(root):
+            continue
+        rel = src.relative_to(root)
+        if media_preview_path(root, rel) == Path(dst):
+            return DeferredPreview(root, rel, strict=False)
+    return dst
+
+
 def _stop_stale_workers():
     """Stop capture workers a previous import of this module started."""
     for thread in threading.enumerate():
@@ -294,7 +351,7 @@ class AutoThumbnailer:
     ``generate(src, dst, is_video, *, stop, guard, track)`` writes one
     preview and returns ``_generate_one``'s result dict; ``may_write()``
     says whether automatic previews may be written now. Views call
-    ``ensure`` on the view pool, the monitor calls ``enqueue`` from its
+    ``ensure`` on the image pool, the monitor calls ``enqueue`` from its
     thread.
     """
 
@@ -420,16 +477,23 @@ class AutoThumbnailer:
 
     def enqueue(self, src, dst, is_video):
         """Queue *dst*'s preview for the worker; False when already queued,
-        full, closed or its worker is refused at interpreter exit. Never
-        blocks on a write."""
-        key = _dst_key(dst)
+        full, closed, nowhere to write it or its worker is refused at
+        interpreter exit. Never blocks on a write. *dst* may be a
+        ``DeferredPreview``: its path is resolved again when the write
+        starts."""
+        now = _resolved(dst)
+        if now is None:
+            return False
+        key = _dst_key(now)
+        if not isinstance(dst, DeferredPreview):
+            dst = now
         with self._changed:
             if self._closed or key in self._queued:
                 return False
             if len(self._queue) >= self._max_queued:
                 return False
             self._queued.add(key)
-            self._queue.append((key, Path(src), Path(dst), bool(is_video)))
+            self._queue.append((key, Path(src), dst, bool(is_video)))
             self._busy = True
             started = True
             if self._start_worker and not self._worker_running:
@@ -491,12 +555,15 @@ class AutoThumbnailer:
                     self._worker_running = False
                     self._changed.notify_all()
                     return
-                take = min(len(self._queue), max(1, int(self._workers() or 1)))
+                workers = max(1, int(self._workers() or 1))
+                take = min(len(self._queue), workers)
                 batch = [self._queue.popleft() for _ in range(take)]
                 self._batch = len(batch)
                 self._changed.notify_all()
             try:
-                map_parallel(self._write_queued, batch, len(batch))
+                # The job pool's size is the setting; a smaller batch uses
+                # fewer of its threads (#523: one long-lived pool).
+                map_parallel(self._write_queued, batch, workers)
             finally:
                 with self._changed:
                     self._batch = 0
@@ -506,6 +573,9 @@ class AutoThumbnailer:
     def _write_queued(self, item):
         _key, src, dst, is_video = item
         try:
+            dst = _resolved(dst)
+            if dst is None:
+                return  # nowhere to write it now: made on its first view
             self.ensure(src, dst, is_video)
         except Exception:  # one file never stops the worker
             self._logger.warning(
@@ -567,7 +637,7 @@ class _PreviewGate:
     """Admission of view requests on the event loop (#249 I2).
 
     Waiters are coroutines: at most ``limit`` previews are written at once
-    on the view pool, one flight runs per key, and past ``MAX_WAITING``
+    on the image pool, one flight runs per key, and past ``MAX_WAITING``
     waiting requests ``run`` answers ``OVER`` at once.
     """
 
@@ -622,13 +692,17 @@ class _PreviewGate:
                 flight.set_result(result)
 
 
+def _on_image_thread():
+    return threading.current_thread().name.startswith(POOL_PREFIX + "_")
+
+
 class AutoThumbnailMixin:
     """Automatic previews on the API: views, the monitor's queue, scans."""
 
     _auto_thumbnailer = None
     _auto_thumbnails_shut = False
     _preview_gate_obj = None
-    _preview_pool_obj = None
+    _image_pool_obj = None
     _auto_thumbnailer_lock = threading.Lock()
 
     def _auto_thumbnails(self):
@@ -651,7 +725,7 @@ class AutoThumbnailMixin:
             self._auto_thumbnails_shut = True
 
     def close_auto_thumbnails(self, final=False, timeout=None):
-        """Stop the capture worker and the view pool, and forget them.
+        """Stop the capture worker and the image pool, and forget them.
 
         Bounded by ``timeout`` (see AutoThumbnailer.close). ``final`` (the
         package teardown) also keeps a late monitor event from starting a
@@ -662,7 +736,7 @@ class AutoThumbnailMixin:
             if final:
                 self._auto_thumbnails_shut = True
             thumbnailer, self._auto_thumbnailer = self._auto_thumbnailer, None
-            pool, self._preview_pool_obj = self._preview_pool_obj, None
+            pool, self._image_pool_obj = self._image_pool_obj, None
             self._preview_gate_obj = None
         closed = True
         if thumbnailer is not None:
@@ -706,16 +780,67 @@ class AutoThumbnailMixin:
             gate = self._preview_gate_obj = _PreviewGate(ON_DEMAND_LIMIT, self.logger)
         return gate
 
-    def _preview_pool(self):
-        with AutoThumbnailMixin._auto_thumbnailer_lock:
-            if self._preview_pool_obj is None:
-                self._preview_pool_obj = ThreadPoolExecutor(
-                    ON_DEMAND_LIMIT, thread_name_prefix=POOL_PREFIX
-                )
-            return self._preview_pool_obj
+    def _image_pool(self):
+        """The interactive image pool (#523): IMAGE_POOL_SIZE threads for view
+        previews (behind their gate), card facts and inspector reads, so
+        image buffers never spread over the default executor's threads.
 
-    def _preview_pool_threads(self):
-        pool = self._preview_pool_obj
+        None after the package teardown (``close_auto_thumbnails(final=True)``):
+        a late request then runs on the default executor (``run_in_executor``
+        with None) and starts no thread the teardown no longer stops.
+        """
+        with AutoThumbnailMixin._auto_thumbnailer_lock:
+            if self._image_pool_obj is None and not self._auto_thumbnails_shut:
+                self._image_pool_obj = ThreadPoolExecutor(
+                    IMAGE_POOL_SIZE, thread_name_prefix=POOL_PREFIX
+                )
+            return self._image_pool_obj
+
+    async def _run_image(self, func, *args, **kwargs):
+        """``_run_in_executor`` for interactive image work: the image pool.
+
+        After the teardown, or with the pool shut down in between (a test
+        reset), the call runs on the default executor as before.
+        """
+        call = functools.partial(func, *args, **kwargs)
+        loop = asyncio.get_running_loop()
+        try:
+            future = loop.run_in_executor(self._image_pool(), call)
+        except RuntimeError:  # submit refused: the pool was shut down
+            future = loop.run_in_executor(None, call)
+        return await future
+
+    def _map_image(self, func, items):
+        """``func`` over *items* on the image pool, in order (blocking).
+
+        Called from an image pool thread, after the teardown, or with the
+        pool shut down in between (its queued items are cancelled), the
+        items it did not run run on the calling thread.
+        """
+        items = list(items)
+        pool = None if not items or _on_image_thread() else self._image_pool()
+        if pool is None:
+            return [func(item) for item in items]
+        futures = []
+        try:
+            for item in items:
+                futures.append(pool.submit(func, item))
+        except RuntimeError:
+            pass  # shut down in between: the rest run here
+        results = []
+        for index, item in enumerate(items):
+            future = futures[index] if index < len(futures) else None
+            if future is None or future.cancelled():
+                results.append(func(item))
+                continue
+            try:
+                results.append(future.result())
+            except CancelledError:  # cancelled while this caller waited
+                results.append(func(item))
+        return results
+
+    def _image_pool_threads(self):
+        pool = self._image_pool_obj
         return [] if pool is None else list(getattr(pool, "_threads", ()))
 
     def _video_placeholder_bytes(self):
@@ -730,7 +855,7 @@ class AutoThumbnailMixin:
         return await self._preview_gate().run(
             flight_key(Path(dst).parent, (Path(dst).name,)),
             functools.partial(self._ensure_view, src, dst, is_video),
-            self._preview_pool(),
+            self._image_pool(),
         )
 
     def _ensure_view(self, src, dst, is_video):
@@ -774,17 +899,17 @@ class AutoThumbnailMixin:
         The flight answers only the outcome (a history flight may be the one
         joined); the file it previews is derived here: the preview path is
         the request's own, and a fallback looks its source up again on the
-        view pool (bounded by it, never the default executor).
+        image pool (bounded by it, never the default executor).
         """
         if _has_traversal(filepath) or not _is_preview_request(filepath):
             return None
         parts = tuple(filepath.split("/"))
         gate = self._preview_gate()
-        pool = self._preview_pool()
+        pool = self._image_pool()
         for root in output_dirs:
             root = Path(root)
             outcome = await gate.run(
-                flight_key(root, parts),
+                _request_key(root, parts),
                 functools.partial(self._preview_on_demand, root, parts),
                 pool,
             )
@@ -797,7 +922,7 @@ class AutoThumbnailMixin:
                     continue
                 return self._preview_response(BUSY, quick[0], None, quick[1])
             dst = _current_preview(root, parts)
-            if outcome == READY:
+            if outcome == READY and dst is not None:
                 return self._file_response(dst)
             located = await gate.loop.run_in_executor(
                 pool, _thumbnail_source, root, parts
@@ -820,17 +945,31 @@ class AutoThumbnailMixin:
         """
         if self._auto_thumbnails_shut:
             return "skipped"
-        located = self._capture_target(image_path)
+        located = self._capture_source(image_path)
         if located is None:
             return "skipped"
-        src, dst, is_video = located
+        src, root, rel, is_video = located
+        dst = DeferredPreview(root, rel).resolve()
+        if dst is None:
+            return "skipped"
         if _thumbnail_is_fresh(src, dst):
             return "fresh"
-        if self._auto_thumbnails().enqueue(src, dst, is_video):
+        if self._auto_thumbnails().enqueue(src, DeferredPreview(root, rel), is_video):
             return "queued"
         return "skipped"
 
     def _capture_target(self, image_path):
+        """``(src, dst, is_video)`` of a new output's preview now, or None."""
+        located = self._capture_source(image_path)
+        if located is None:
+            return None
+        src, root, rel, is_video = located
+        dst = DeferredPreview(root, rel).resolve()
+        return None if dst is None else (src, dst, is_video)
+
+    def _capture_source(self, image_path):
+        """``(src, root, rel, is_video)`` of a media file under a current
+        root (not a preview, a link or in a trash folder), or None."""
         path = Path(os.path.abspath(str(image_path)))
         if path.suffix.lower() not in MEDIA_EXTENSIONS or not _regular_file(path):
             return None
@@ -843,12 +982,7 @@ class AutoThumbnailMixin:
             rel = Path(os.path.relpath(path, root))
             if "thumbnails" in rel.parts[:-1] or in_trash_below_roots(path, [root]):
                 return None
-            dst = root / media_thumbnail_rel(rel)
-            if not _is_within(root / "thumbnails", root) or not _is_within(
-                dst, root / "thumbnails"
-            ):
-                return None
-            return path, dst, path.suffix.lower() in VIDEO_EXTENSIONS
+            return path, root, rel, path.suffix.lower() in VIDEO_EXTENSIONS
         return None
 
     # ── after a scan ──────────────────────────────────────────────────
@@ -890,7 +1024,7 @@ class AutoThumbnailMixin:
             return "none"
         thumbnailer = self._auto_thumbnails()
         for src, dst, is_video in plan["targets"]:
-            thumbnailer.enqueue(src, dst, is_video)
+            thumbnailer.enqueue(src, _deferred(output_dirs, src, dst), is_video)
         self.logger.info(f"{missing:,} previews missing after the scan; queued")
         return "queued"
 

py/api/thumbnail_convert.py

@@ -153,6 +153,7 @@ class ThumbnailConvertMixin:
             job.log_line("Conversion failed · internal error")
             state = "failed"
         job.finish(state)
+        await self._release_job_memory()
 
     async def _convert_job_body(self, job, options):
         output_dirs = await self._run_in_executor(self._get_all_output_dirs)

py/api/thumbnail_encode.py

@@ -46,7 +46,7 @@ from pathlib import Path
 
 from PIL import Image
 
-from . import thumbnail_settings
+from . import preview_store, thumbnail_settings
 
 try:
     from ...utils.logging_config import get_logger
@@ -101,10 +101,15 @@ def temp_sibling(dst, tag="part"):
 
 
 def replace_unless_held(tmp, dst, guard=None):
-    """Rename *tmp* over *dst*, unless ``guard()`` is false (PreviewWriteHeld)."""
-    if guard is not None and not guard():
-        raise PreviewWriteHeld(PREVIEW_HELD)
-    os.replace(tmp, dst)
+    """Rename *tmp* over *dst*, unless ``guard()`` is false (PreviewWriteHeld).
+
+    The check and the rename run under ``preview_store.write_lock`` (#320): a
+    guard from ``preview_store.write_guard`` also refuses a write that started
+    before the thumbnail location changed."""
+    with preview_store.committed(guard) as allowed:
+        if not allowed:
+            raise PreviewWriteHeld(PREVIEW_HELD)
+        os.replace(tmp, dst)
 
 
 # ``temp_sibling``'s names: ``.<stem>.<tag>-<pid>-<thread id><ext>``.

py/api/thumbnail_jobs.py

@@ -226,6 +226,7 @@ class ThumbnailJobsMixin:
             )
             state = "failed"
         job.finish(state)
+        await self._release_job_memory()
 
     async def _thumbnail_job_body(self, job, options):
         """Plan, then generate in worker batches; returns the final state.

py/api/thumbnail_settings.py

@@ -114,9 +114,16 @@ def validate_edge(value):
 # ── names ────────────────────────────────────────────────────────────
 
 
+def preview_stem(rel_path):
+    """``<rel without extension>_thumb`` (POSIX): a preview's name inside the
+    store of its source (``preview_store``)."""
+    return f"{Path(rel_path).with_suffix('').as_posix()}{THUMB_SUFFIX}"
+
+
 def thumbnail_stem_rel(rel_path):
-    """``thumbnails/<rel without extension>_thumb`` (POSIX)."""
-    return f"thumbnails/{Path(rel_path).with_suffix('').as_posix()}{THUMB_SUFFIX}"
+    """``thumbnails/<rel without extension>_thumb`` (POSIX): URL text only.
+    Where the preview is on disk is ``preview_store``'s answer."""
+    return f"thumbnails/{preview_stem(rel_path)}"
 
 
 def thumbnail_rel(rel_path, extension=None):
@@ -207,8 +214,13 @@ def owned_in_folder(folder, stem, source_extension, preferred=None):
 
 
 def _owned(root, rel_path, preferred):
+    from . import preview_store
+
     rel_path = Path(rel_path)
-    stem = Path(root) / thumbnail_stem_rel(rel_path)
+    store = preview_store.store_dir(root)
+    if store is None:
+        return iter(())
+    stem = store / preview_stem(rel_path)
     return owned_in_folder(stem.parent, stem.name, rel_path.suffix, preferred)
 
 

tests/browser/prompt-inspector.spec.mjs

@@ -320,3 +320,44 @@ test('copy toast and fallback dialog',async({page})=>{
     await expect(fallback).toContainText('Automatic clipboard access requires a secure browser context.');
     await page.keyboard.press('Escape');await expect(fallback).toBeHidden();await expect(copy).toBeFocused();
 });
+
+// #542 D191: "All N eligible nodes" previews every node, sends one after another and names each refusal.
+test('All eligible nodes previews each node, sends in node id order and names a refusal',async({page,context},testInfo)=>{
+    const host=await context.newPage();await host.goto('/prompt_manager/workspace#/overview');
+    await host.evaluate(async()=>{
+        const {createBridgeHost}=await import('/prompt_manager/js/shared/toolkit/prompt-bridge.js');
+        const targets=['12','3','7'].map(id=>({id,label:id==='7'?'Edited in ComfyUI':'Prompt Manager (+/-)',type:'PromptManagerPair',roles:['positive','negative']}));
+        window.hostValues=Object.fromEntries(targets.map(target=>[target.id,{positive:`Existing ${target.id}`,negative:'Existing negative'}]));window.applied=[];
+        window.fixtureHost=createBridgeHost({origin:location.origin,namespace:'/',label:'Test ComfyUI',adapter:{
+            listTargets:()=>targets,
+            preview:(id,roles)=>{const target=targets.find(item=>item.id===id);return{preview:{target,fields:Object.entries(roles).map(([role,after])=>({role,before:window.hostValues[id][role],after}))},snapshot:{id,roles}};},
+            apply:({id,roles})=>{if(id==='7')throw new Error('Workflow or prompt changed. Preview again.');Object.assign(window.hostValues[id],roles);window.applied.push(id);return{targetId:id,values:roles};},
+        }});
+    });
+    try {
+        await page.goto(workspace);await select(page,'0:shared/same.png');await details(page).getByRole('button',{name:/^Send prompt pair to ComfyUI: /}).click();
+        const sender=dialog(page);const nodes=sender.getByRole('combobox',{name:'Destination node',exact:true});
+        await sender.getByRole('combobox',{name:'ComfyUI window',exact:true}).selectOption({label:'Test ComfyUI'});
+        await expect(nodes.locator('option')).toHaveText(['Choose a destination node','#12 Prompt Manager (+/-)','#3 Prompt Manager (+/-)','#7 Edited in ComfyUI','All 3 eligible nodes']);
+        await expect(nodes).toHaveValue('');
+        await nodes.selectOption({label:'All 3 eligible nodes'});
+        await expect(sender.getByRole('status')).toHaveText('Sending to all 3 eligible nodes in Test ComfyUI. Apply to send it, or preview the changes first.');
+        await page.screenshot({path:testInfo.outputPath('send-all-option.png')});
+        await sender.getByRole('button',{name:'Preview changes',exact:true}).click();
+        const reviews=sender.locator('section.prompt-sender-node');
+        await expect(reviews.locator('> h4')).toHaveText(['#3 Prompt Manager (+/-)','#7 Edited in ComfyUI','#12 Prompt Manager (+/-)']);
+        await expect(sender.getByRole('region',{name:'#12 Prompt Manager (+/-)',exact:true})).toContainText('Existing 12');
+        expect(await host.evaluate(()=>window.applied)).toEqual([]);
+        await sender.getByRole('button',{name:'Apply to ComfyUI',exact:true}).click();
+        await expect(sender.getByRole('alert')).toHaveText('Sent to 2 of 3 nodes in Test ComfyUI. #7 Edited in ComfyUI: Workflow or prompt changed. Preview again.');
+        expect(await host.evaluate(()=>window.applied)).toEqual(['3','12']);
+        expect(await host.evaluate(()=>window.hostValues['7'].positive)).toBe('Existing 7');
+        await page.screenshot({path:testInfo.outputPath('send-all-result.png')});
+        await page.setViewportSize({width:390,height:844});
+        await sender.getByRole('button',{name:'Preview changes',exact:true}).click();await expect(reviews).toHaveCount(3);
+        await reviews.last().scrollIntoViewIfNeeded();
+        expect(await page.evaluate(()=>document.querySelector('dialog.prompt-sender').scrollWidth<=document.querySelector('dialog.prompt-sender').clientWidth+1)).toBe(true);
+        await page.screenshot({path:testInfo.outputPath('send-all-mobile.png')});
+        const apply=sender.getByRole('button',{name:'Apply to ComfyUI',exact:true});await apply.focus();await expect(apply).toBeFocused();
+    } finally {await host.evaluate(()=>window.fixtureHost.dispose());await host.close();}
+});

tests/browser/workspace-system-backups.spec.mjs

@@ -284,6 +284,23 @@ test('a restore seen finishing elsewhere closes an open restore review and disca
     expect((await serverJson(request, '/__test__/system/backups')).stages).toBe(0);
 });
 
+// #583: in WebKit, while a page.route is armed, a file part of a FormData body is read only when the request goes on;
+// a garbage collection in between dropped the form once nothing referenced it, and the server got an empty file
+// ("Uploaded file is empty"). The page keeps the form until its upload is answered. The route holds the upload
+// through a collection; without the fix WebKit lost the file every time (0/10 in the pinned image). Chromium and
+// Firefox keep the body anyway, so there it only checks that a held upload still opens the review.
+test('a review upload survives a garbage collection while it is in flight', async ({ page, request }) => {
+    const bytes = await backupBytes(request);
+    await page.route(`**${PREVIEW_URL}`, async route => {
+        await page.requestGC();
+        await route.continue();
+    });
+    await openBackups(page);
+    expect((await chooseBackup(page, bytes)).status()).toBe(200);
+    await expect(dialog(page).locator('#modal-title')).toHaveText('Restore preview');
+    expect((await serverJson(request, '/__test__/system/backups')).stages).toBe(1);
+});
+
 test('invalid file shows the validator message; nothing is staged', async ({ page, request }) => {
     const before = await serverJson(request, '/__test__/system/digest');
     await openBackups(page);

tests/comfyui/prompt-sender.spec.mjs

@@ -131,11 +131,40 @@ test('native image inspector previews and sends a pair into real visible widgets
         const twinId=await page.evaluate(async()=>{const {app}=await import('/scripts/app.js');const node=LiteGraph.createNode('PromptManagerPair');app.graph.add(node);return node.id;});
         await find0.click();
         const targets=sender.getByRole('combobox',{name:'Destination node',exact:true});
-        await expect(targets.locator('option')).toHaveCount(3);
+        await expect(targets.locator('option')).toHaveCount(4);
         await expect(targets).toHaveValue('');
         await expect(apply).toBeDisabled();
         await expect(targets.locator('option').nth(2)).toHaveText(new RegExp(`^#${twinId} `));
-        await page.evaluate(async twinId=>{const {app}=await import('/scripts/app.js');app.graph.remove(app.graph.getNodeById(twinId));},twinId);
+        // #542 D191: "All 2 eligible nodes" writes both real nodes, one after another; a linked third node is not offered.
+        const linkedId=await page.evaluate(async()=>{
+            const {app}=await import('/scripts/app.js');const {convertToInput}=await import('/extensions/core/widgetInputs.js');
+            const node=LiteGraph.createNode('PromptManagerPair');app.graph.add(node);
+            const widget=node.widgets.find(widget=>widget.name==='text');
+            if(!node.inputs.some(input=>input.name==='text'))convertToInput(node,widget);
+            const source=LiteGraph.createNode('PrimitiveString');app.graph.add(source);
+            source.connect(0,node,node.inputs.findIndex(input=>input.name==='text'));
+            for(const each of app.graph._nodes.filter(item=>item.type==='PromptManagerPair')){const text=each.widgets.find(widget=>widget.name==='text');if(text)text.value=`Before all ${each.id}`;}
+            return node.id;
+        });
+        await find0.click();
+        await expect(targets.locator('option')).toHaveText(['Choose a destination node',new RegExp(`^#${id} `),new RegExp(`^#${twinId} `),'All 2 eligible nodes']);
+        await expect(sender.getByRole('status')).toContainText('1 other Prompt Manager node cannot receive this prompt.');
+        await targets.selectOption({label:'All 2 eligible nodes'});
+        await sender.getByRole('button',{name:'Preview changes',exact:true}).click();
+        await expect(sender.locator('section.prompt-sender-node > h4')).toHaveText([new RegExp(`^#${id} `),new RegExp(`^#${twinId} `)]);
+        await apply.click();
+        // The tab title gains ComfyUI's unsaved mark once the workflow changed, so the window is matched loosely.
+        await expect(sender.getByRole('status')).toHaveText(/^Sent to 2 of 2 nodes in .*ComfyUI\.$/);
+        await expect(apply).toBeDisabled();
+        const all=await page.evaluate(async ids=>{
+            const {app}=await import('/scripts/app.js');const output=(await app.graphToPrompt()).output;
+            return ids.map(nodeId=>{const node=app.graph.getNodeById(nodeId);const serialized=node.serialize();const index=node.widgets.findIndex(widget=>widget.name==='negative_text');
+                return{prompt:output[String(nodeId)]?.inputs.negative_text,serialized:serialized.widgets_values[index],text:node.widgets.find(widget=>widget.name==='text')?.value};});
+        },[id,twinId,linkedId]);
+        expect(all.slice(0,2)).toEqual([{prompt:negative,serialized:negative,text:positive},{prompt:negative,serialized:negative,text:positive}]);
+        expect(all[2].text).toBe(`Before all ${linkedId}`);
+        expect(await page.evaluate(()=>window.senderEffects.queue-window.senderSendBaseline.queue)).toBe(0);
+        await page.evaluate(async ids=>{const {app}=await import('/scripts/app.js');for(const nodeId of ids)app.graph.remove(app.graph.getNodeById(nodeId));},[twinId,linkedId]);
         // #535: the status line says why a window cannot receive.
         const find=sender.getByRole('button',{name:'Find ComfyUI windows',exact:true});
         const preview=sender.getByRole('button',{name:'Preview changes',exact:true});

tests/js/prompt-sender.test.js

@@ -225,3 +225,141 @@ test('acknowledgement cannot claim an extra role beyond approved changes',async(
     const {createSendSession}=await source;const s=setup(createSendSession,{apply:async()=>({targetId:'one',values:{positive:'new',negative:'also changed'}})});
     await s.session.discover();s.session.selectHost('host');s.session.selectTarget('one');await s.session.preview();await s.session.apply();assert.equal(s.session.state.phase,'error');assert.match(s.session.state.message,/unexpected acknowledgement/);s.session.dispose();
 });
+// #542: "All N eligible nodes" sends to every eligible node of one window, previewing every node first (Q542-2 C).
+function allHost({nodes=['10','9','2'],revision,preview,apply,discover}={}) {
+    const calls=[];const sent=new Set();
+    const targets=nodes.map(id=>pm(id,['positive'],revision?{revision}:{}));
+    const client={
+        discover:discover ?? (async()=>[window1(targets)]),
+        preview:async(host,id,roles)=>{calls.push(['preview',id]);if(preview)return preview(id,calls,sent);return{ticket:`t-${id}-${calls.length}`,revision,target:{id,label:`Node ${id}`},fields:[{role:'positive',before:sent.has(id)?'new':'old',after:roles.positive}]};},
+        apply:async(host,ticket)=>{const id=ticket.split('-')[1];calls.push(['apply',id,ticket]);if(apply)return apply(id,calls);sent.add(id);return{targetId:id,values:{positive:'new'}};},
+        dispose:()=>calls.push(['dispose']),
+    };
+    return {client,calls,sent};
+}
+async function allSession(createSendSession,options) {
+    const host=allHost(options);const session=createSendSession({client:host.client,roles:{positive:'new'},onChange:()=>{}});
+    await session.discover();return {...host,session};
+}
+const order=calls=>calls.filter(call=>call[0]!=='dispose').map(call=>`${call[0]} ${call[1]}`);
+test('All is offered only with two or more eligible nodes and is never selected for you',async()=>{
+    const {createSendSession,ALL_NODES}=await source;
+    const one=await allSession(createSendSession,{nodes:['1']});
+    one.session.selectTarget(ALL_NODES);assert.equal(one.session.state.targetId,'');assert.equal(one.calls.length,0);one.session.dispose();
+    const two=await allSession(createSendSession,{nodes:['1','2']});
+    assert.equal(two.session.state.targetId,'');
+    two.session.selectTarget(ALL_NODES);assert.equal(two.session.state.targetId,ALL_NODES);
+    assert.match(two.session.state.message,/all 2 eligible nodes in Flow - ComfyUI/);
+    await two.session.discover();assert.equal(two.session.state.targetId,'');two.session.dispose();
+});
+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']);
+    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();
+});
+test('a double click on Apply to All sends to each node once',async()=>{
+    const {createSendSession,ALL_NODES}=await source;const s=await allSession(createSendSession,{nodes:['1','2']});
+    s.session.selectTarget(ALL_NODES);await Promise.all([s.session.apply(),s.session.apply()]);
+    assert.deepEqual(s.calls.filter(call=>call[0]==='apply').map(call=>call[1]),['1','2']);s.session.dispose();
+});
+test('a node refusing its preview stops All before anything is written and names the node',async()=>{
+    const {createSendSession,ALL_NODES}=await source;
+    const s=await allSession(createSendSession,{preview:id=>{if(id==='9')throw Error('positive is connected to an upstream input.');return{ticket:`t-${id}`,target:{id},fields:[{role:'positive',before:'old',after:'new'}]};}});
+    s.session.selectTarget(ALL_NODES);await s.session.apply();
+    assert.equal(s.calls.filter(call=>call[0]==='apply').length,0);assert.equal(s.session.state.phase,'error');
+    assert.match(s.session.state.message,/^#9 Node 9: positive is connected to an upstream input\. Nothing was sent\./);
+    assert.equal(s.session.state.outcomeUnknown,false);s.session.dispose();
+});
+test('a refusal while sending to All is reported by node and the other nodes are still sent',async()=>{
+    const {createSendSession,ALL_NODES}=await source;
+    const s=await allSession(createSendSession,{apply:id=>{if(id==='9')throw Error('Workflow or prompt changed. Preview again.');return{targetId:id,values:{positive:'new'}};}});
+    s.session.selectTarget(ALL_NODES);await s.session.apply();
+    assert.deepEqual(s.calls.filter(call=>call[0]==='apply').map(call=>call[1]),['2','9','10']);
+    assert.equal(s.session.state.phase,'error');assert.equal(s.session.state.outcomeUnknown,false);
+    assert.equal(s.session.state.message,'Sent to 2 of 3 nodes in Flow - ComfyUI. #9 Node 9: Workflow or prompt changed. Preview again.');s.session.dispose();
+});
+test('a lost reply on one node of All leaves the later nodes unsent and that outcome unknown',async()=>{
+    const {createSendSession,ALL_NODES}=await source;
+    const s=await allSession(createSendSession,{apply:id=>{if(id==='9')throw Error('Bridge request timed out');return{targetId:id,values:{positive:'new'}};}});
+    s.session.selectTarget(ALL_NODES);await s.session.apply();
+    assert.deepEqual(s.calls.filter(call=>call[0]==='apply').map(call=>call[1]),['2','9']);
+    assert.equal(s.session.state.outcomeUnknown,true);
+    assert.match(s.session.state.message,/^Sent to 1 of 3 nodes in Flow - ComfyUI\. No confirmation received from #9 Node 9; the update may have applied\. Not sent: #10 Node 10\. Check ComfyUI, then preview again\. Bridge request timed out$/);
+    await s.session.apply();assert.equal(s.calls.filter(call=>call[0]==='apply').length,2);s.session.dispose();
+});
+test('closing the dialog during All stops after the current node',async()=>{
+    const {createSendSession,ALL_NODES}=await source;let release;
+    const s=await allSession(createSendSession,{apply:id=>id==='2'?new Promise(done=>release=()=>done({targetId:id,values:{positive:'new'}})):{targetId:id,values:{positive:'new'}}});
+    s.session.selectTarget(ALL_NODES);const pending=s.session.apply();
+    for(let i=0;i<50 && !release;i++)await tick();assert.ok(release,'the first node was sent');
+    s.session.dispose();release();await pending;
+    assert.deepEqual(s.calls.filter(call=>call[0]==='apply').map(call=>call[1]),['2']);
+});
+test('All accepts the revision our own write moved, but not a workflow changed before the run',async()=>{
+    const {createSendSession,ALL_NODES}=await source;
+    // The host moves the graph version on every write: later nodes preview with the new revision.
+    let version=4;
+    const bumped=await allSession(createSendSession,{revision:'1:4',preview:id=>({ticket:`t-${id}`,revision:`1:${version}`,target:{id},fields:[{role:'positive',before:'old',after:'new'}]}),apply:id=>{version++;return{targetId:id,values:{positive:'new'}};}});
+    bumped.session.selectTarget(ALL_NODES);await bumped.session.apply();
+    assert.equal(bumped.session.state.message,'Sent to 3 of 3 nodes in Flow - ComfyUI.');bumped.session.dispose();
+    const moved=await allSession(createSendSession,{revision:'1:4',preview:id=>({ticket:`t-${id}`,revision:'1:5',target:{id},fields:[{role:'positive',before:'old',after:'new'}]})});
+    moved.session.selectTarget(ALL_NODES);await moved.session.apply();
+    assert.match(moved.session.state.message,/workflow changed since this window was found/i);assert.equal(moved.calls.filter(call=>call[0]==='apply').length,0);moved.session.dispose();
+});
+test('a node whose text changed after the preview pass is refused, not overwritten',async()=>{
+    const {createSendSession,ALL_NODES}=await source;let edited=false;
+    const s=await allSession(createSendSession,{preview:id=>({ticket:`t-${id}`,target:{id},fields:[{role:'positive',before:id==='10'&&edited?'typed in ComfyUI':'old',after:'new'}]}),apply:id=>{edited=true;return{targetId:id,values:{positive:'new'}};}});
+    s.session.selectTarget(ALL_NODES);await s.session.apply();
+    assert.deepEqual(s.calls.filter(call=>call[0]==='apply').map(call=>call[1]),['2','9']);
+    assert.match(s.session.state.message,/^Sent to 2 of 3 nodes in Flow - ComfyUI\. #10 Node 10: Its prompt changed since the preview\./);s.session.dispose();
+});
+test('Preview with All shows every node in order, and Apply sends fresh tickets that match the review',async()=>{
+    const {createSendSession,ALL_NODES}=await source;const s=await allSession(createSendSession);
+    s.session.selectTarget(ALL_NODES);await s.session.preview();
+    assert.equal(s.session.state.phase,'ready');assert.equal(s.session.state.preview,null);
+    assert.deepEqual(s.session.state.previews.map(item=>item.label),['#2 Node 2','#9 Node 9','#10 Node 10']);
+    assert.match(s.session.state.message,/Sending to all 3 eligible nodes in Flow - ComfyUI\. Review the complete changes, then apply\./);
+    const first=s.session.state.previews[0].preview.ticket;
+    await s.session.apply();
+    // A held review may have outlived the host's 30 s tickets, so even the first node takes a fresh one.
+    assert.deepEqual(order(s.calls),['preview 2','preview 9','preview 10','preview 2','apply 2','preview 9','apply 9','preview 10','apply 10']);
+    assert.notEqual(s.calls.find(call=>call[0]==='apply')[2],first);assert.equal(s.session.state.phase,'success');
+    s.session.selectTarget('2');assert.deepEqual(s.session.state.previews,[]);s.session.dispose();
+});
+test('the dialog offers All for two or more nodes and reviews each node under its name',async()=>{
+    const {createPromptSender,ALL_NODES}=await source;const {createScope}=await import('../../web/js/shared/toolkit/core.js');
+    const dom=fakeDOM();const scope=createScope();
+    const {client}=allHost({nodes:['1','2']});
+    try {
+        createPromptSender({scope,clientFactory:()=>client}).open({positive:'new'});await tick();
+        const dialog=dom.body.children.at(-1);const target=dom.find(dialog,node=>node.id==='prompt-sender-target');
+        assert.deepEqual(target.children.map(option=>[option.value,option.textContent]),[['','Choose a destination node'],['1','#1 Node 1'],['2','#2 Node 2'],[ALL_NODES,'All 2 eligible nodes']]);
+        assert.equal(target.value,'');target.value=ALL_NODES;target.dispatchEvent(new Event('change'));
+        dom.find(dialog,node=>node.textContent==='Preview changes').dispatchEvent(new Event('click'));for(let i=0;i<5;i++)await tick();
+        const sections=[];(function walk(node){if(node.className==='prompt-sender-node')sections.push(node);node.children.forEach(walk);})(dialog);
+        assert.deepEqual(sections.map(section=>section.children[0].textContent),['#1 Node 1','#2 Node 2']);
+        dom.find(dialog,node=>node.textContent==='Apply to ComfyUI').dispatchEvent(new Event('click'));for(let i=0;i<10;i++)await tick();
+        assert.ok(dom.find(dialog,node=>node.textContent==='Sent to 2 of 2 nodes in Flow - ComfyUI.'));
+    } finally {scope.dispose();dom.restore();}
+});
+test('an expired review still sends every node, and a held review checks the discovered revision first',async()=>{
+    const {createSendSession,ALL_NODES}=await source;let reviewing=true;
+    // The host refuses every ticket taken during the review, as after its 30 s expiry.
+    const s=await allSession(createSendSession,{preview:(id,calls)=>({ticket:`t-${id}-${reviewing?'old':calls.length}`,target:{id},fields:[{role:'positive',before:'old',after:'new'}]}),apply:id=>({targetId:id,values:{positive:'new'}})});
+    const realApply=s.client.apply;s.client.apply=async(host,ticket,options)=>{if(ticket.endsWith('-old'))throw Error('Preview expired or unavailable; preview again');return realApply(host,ticket,options);};
+    s.session.selectTarget(ALL_NODES);await s.session.preview();reviewing=false;await s.session.apply();
+    assert.equal(s.session.state.message,'Sent to 3 of 3 nodes in Flow - ComfyUI.');s.session.dispose();
+    const moved=await allSession(createSendSession,{revision:'1:4',preview:(id,calls)=>({ticket:`t-${id}`,revision:calls.length>3?'1:5':'1:4',target:{id},fields:[{role:'positive',before:'old',after:'new'}]})});
+    moved.session.selectTarget(ALL_NODES);await moved.session.preview();assert.equal(moved.session.state.phase,'ready');await moved.session.apply();
+    assert.match(moved.session.state.message,/^Sent to 0 of 3 nodes in Flow - ComfyUI\. #2 Node 2: The workflow changed since this window was found/);
+    assert.equal(moved.calls.filter(call=>call[0]==='apply').length,0);moved.session.dispose();
+});
+test('a changed prepend or append after the review refuses that node',async()=>{
+    const {createSendSession,ALL_NODES}=await source;let edited=false;
+    const s=await allSession(createSendSession,{preview:id=>({ticket:`t-${id}`,target:{id},fields:[{role:'positive',before:'old',after:'new'}],composition:{prepend:id==='9'&&edited?'typed':'trigger',append:'',before:'trigger old',after:'trigger new'}}),apply:id=>{edited=true;return{targetId:id,values:{positive:'new'}};}});
+    s.session.selectTarget(ALL_NODES);await s.session.apply();
+    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();
+});

tests/js/workspace-system-backups.test.js

@@ -128,6 +128,24 @@ test('invalid file shows the validator message; a refusal names its job; a lost
     assert.deepEqual(await uploadRestore(file, { ApiError, fetchUrl: async () => response(200, PREVIEW) }), PREVIEW);
 });
 
+// #583: WebKit reads a FormData's file part by reference, possibly after a garbage collection; the page holds the form
+// until the upload is answered, then lets it go, whether the server staged it, refused it or the request failed.
+test('a review upload keeps its form referenced until the upload is answered', async () => {
+    const { uploadRestore, uploadsInFlight } = await backups;
+    const file = new File(['bytes'], 'b.db');
+    for (const answer of [async () => response(200, PREVIEW), async () => response(400, { success: false, error: 'x' }),
+        async () => { throw new TypeError('offline'); }]) {
+        let held;
+        const upload = uploadRestore(file, { ApiError, fetchUrl: async (url, init) => {
+            held = uploadsInFlight.has(init.body) && init.body.get('database_file').size === file.size;
+            return answer();
+        } }).catch(() => {});
+        await upload;
+        assert.equal(held, true);
+        assert.equal(uploadsInFlight.size, 0);
+    }
+});
+
 test('Cancel discards the stage through the api client and never throws', async () => {
     const { discardRestore, PREVIEW_URL } = await backups;
     const calls = [];

tests/test_api_workspace_images.py

@@ -908,20 +908,20 @@ class TestWorkspaceImages(ImageAPITestCase):
 
     async def test_one_cards_parse_error_spares_the_page(self):
         prompt = self._save_prompt("a fox")
-        for name, mtime in (("bad.png", 200), ("ok.png", 100)):
+        # Cards are read on the image pool (#523), in no fixed order: the
+        # parser fails on bad.png's own record (seed 6), not on a call count.
+        for name, mtime, seed in (("bad.png", 200, 6), ("ok.png", 100, 5)):
             path = records_png(
-                self.output_dir / name, seeded_records("a fox", 5), mtime
+                self.output_dir / name, seeded_records("a fox", seed), mtime
             )
             self.api.db.link_image_to_prompt(prompt, str(path))
         real_parse = header_params_of_images()
 
         def parsing(info):
-            if parsing.calls == 0:
-                parsing.calls += 1
+            if '"seed": 6' in info["prompt"]:
                 raise RuntimeError("unexpected parser bug")
             return real_parse(info)
 
-        parsing.calls = 0
         with (
             patch("py.api.images.header_params", side_effect=parsing),
             self.assertLogs(self.api.logger, "WARNING") as logged,

tests/test_exit_window_threads.py

@@ -146,6 +146,18 @@ class TestPreviewQueueRefusedAtShutdown(unittest.TestCase):
 
 
 class TestMapParallelRefusedAtShutdown(unittest.TestCase):
+    def setUp(self):
+        # The job pool is long-lived (#523): start each test without one, so
+        # the patched pool class is the one created.
+        self.assertTrue(parallel.shutdown_job_pool(30))
+        self.addCleanup(parallel.shutdown_job_pool, 30)
+
+    def test_the_refused_pool_is_not_reused(self):
+        with mock.patch.object(parallel, "ThreadPoolExecutor", _ShutPool):
+            parallel.map_parallel(lambda n: n, [1, 2], 2)
+        self.assertFalse(parallel.job_pool_started())
+        self.assertEqual(parallel.map_parallel(lambda n: n + 1, [1, 2], 2), [2, 3])
+
     def test_a_shut_pool_runs_the_items_on_the_calling_thread(self):
         records = _capture(self, "prompt_manager.parallel")
         with mock.patch.object(parallel, "ThreadPoolExecutor", _ShutPool):

tests/test_image_pools.py

@@ -0,0 +1,241 @@
+"""Image work runs on PromptManager's own pools, never the default executor (#523).
+
+glibc keeps each thread's malloc arena at its high-water mark, so image work
+spread over the default executor's threads (min(32, cpus + 4)) kept an
+image-sized buffer per thread. Interactive image work (card facts, inspector
+reads, view previews) runs on the fixed interactive image pool
+(``pm-image``); jobs (scan metadata, thumbnail jobs, capture previews) on the
+long-lived Worker threads pool (``pm-worker``). After a scan or thumbnail
+job, glibc's free pages are released once (``malloc_trim``).
+"""
+
+import asyncio
+import threading
+from unittest.mock import patch
+
+import py.api as api_package
+import py.api.images as images_module
+import py.api.inspection_service as inspection_service
+import py.api.thumbnail_auto as auto
+from utils import parallel
+
+try:
+    from tests.test_api_images import make_png
+    from tests.test_thumbnail_auto import WAIT, AutoThumbnailCase, _entry
+except ImportError:  # discovered with tests/ as the top-level directory
+    from test_api_images import make_png
+    from test_thumbnail_auto import WAIT, AutoThumbnailCase, _entry
+
+IMAGE_THREAD = auto.POOL_PREFIX + "_"
+JOB_THREAD = parallel.JOB_THREAD_PREFIX + "_"
+
+
+class ImagePoolCase(AutoThumbnailCase):
+    async def asyncSetUp(self):
+        await super().asyncSetUp()
+        self.assertTrue(parallel.shutdown_job_pool(WAIT))
+        self.addCleanup(parallel.shutdown_job_pool, WAIT)
+        self.threads = []
+        self.lock = threading.Lock()
+
+    def spy(self, owner, name):
+        """Record the thread every call of ``owner.name`` runs on."""
+        original = getattr(owner, name)
+
+        def recording(*args, **kwargs):
+            with self.lock:
+                self.threads.append(threading.current_thread().name)
+            return original(*args, **kwargs)
+
+        patcher = patch.object(owner, name, recording)
+        patcher.start()
+        self.addCleanup(patcher.stop)
+
+    def assert_threads(self, prefix):
+        self.assertTrue(self.threads, "the spied call never ran")
+        self.assertTrue(
+            all(name.startswith(prefix) for name in self.threads), self.threads
+        )
+
+    async def run_job(self, path, body=None):
+        response = await self.client.post(path, json=body or {})
+        self.assertEqual(response.status, 202, await response.text())
+        job = self.api._jobs.get((await response.json())["job"]["id"])
+        await asyncio.wait_for(asyncio.shield(job.task), WAIT)
+        return job
+
+
+class TestInteractiveImagePool(ImagePoolCase):
+    async def test_workspace_card_facts_are_read_on_the_image_pool(self):
+        for index in range(5):
+            make_png(self.output_dir / f"c{index}.png", size=(32, 32))
+        self.spy(images_module, "_card_header")
+        response = await self.client.get(
+            "/prompt_manager/images/output?mode=workspace&limit=100"
+        )
+        self.assertEqual(response.status, 200, await response.text())
+        images = (await response.json())["images"]
+        self.assertEqual([image["width"] for image in images], [32] * 5)
+        self.assert_threads(IMAGE_THREAD)
+        self.assertEqual(len(self.threads), 5)
+
+    async def test_cached_card_facts_need_no_image_pool_trip(self):
+        make_png(self.output_dir / "c.png", size=(32, 32))
+        url = "/prompt_manager/images/output?mode=workspace&limit=100"
+        await (await self.client.get(url)).read()
+        self.spy(images_module, "_card_header")
+        response = await self.client.get(url)
+        self.assertEqual((await response.json())["images"][0]["width"], 32)
+        self.assertEqual(self.threads, [])
+
+    async def test_the_inspector_reads_on_the_image_pool(self):
+        make_png(self.output_dir / "i.png", size=(32, 32))
+        self.spy(inspection_service, "read_inspection_file")
+        response = await self.client.get("/prompt_manager/images/inspect/i.png?root=0")
+        self.assertEqual(response.status, 200, await response.text())
+        self.assert_threads(IMAGE_THREAD)
+
+    async def test_view_previews_and_card_reads_share_three_threads(self):
+        names = [f"v{i}.png" for i in range(auto.ON_DEMAND_LIMIT + 3)]
+        for name in names:
+            make_png(self.output_dir / name, size=(64, 64))
+        calls = []
+        original = self.api._generate_one
+
+        def recording(*args, **kwargs):
+            calls.append(threading.current_thread().name)
+            return original(*args, **kwargs)
+
+        self.api._generate_one = recording
+        for name in names:
+            url = _entry(self.api, self.output_dir, name)["thumbnail_url"]
+            await (await self.client.get(url)).read()
+        self.assertEqual(len(calls), len(names))
+        self.assertTrue(all(name.startswith(IMAGE_THREAD) for name in calls), calls)
+        self.assertLessEqual(len(self.api._image_pool_threads()), auto.IMAGE_POOL_SIZE)
+
+    def test_map_image_keeps_order_and_runs_inline_on_a_pool_thread(self):
+        self.assertEqual(self.api._map_image(lambda n: n * 2, [3, 1, 2]), [6, 2, 4])
+        nested = self.api._image_pool().submit(
+            self.api._map_image, lambda _: threading.current_thread().name, [1]
+        )
+        (name,) = nested.result(WAIT)
+        self.assertTrue(name.startswith(IMAGE_THREAD), name)
+
+    def test_map_image_after_a_close_runs_on_the_calling_thread(self):
+        pool = self.api._image_pool()
+        pool.shutdown(wait=True)
+        here = threading.current_thread().name
+        with patch.object(self.api, "_image_pool", return_value=pool):
+            names = self.api._map_image(
+                lambda _: threading.current_thread().name, [1, 2]
+            )
+        self.assertEqual(names, [here, here])
+
+    async def test_after_the_final_close_no_image_thread_starts(self):
+        await asyncio.get_running_loop().run_in_executor(
+            None, lambda: self.api.close_auto_thumbnails(final=True)
+        )
+        self.assertIsNone(self.api._image_pool())
+        here = threading.current_thread().name
+        names = self.api._map_image(lambda _: threading.current_thread().name, [1])
+        self.assertEqual(names, [here])
+        name = await self.api._run_image(lambda: threading.current_thread().name)
+        self.assertFalse(name.startswith(IMAGE_THREAD), name)
+        self.assertEqual(self.api._image_pool_threads(), [])
+
+    def test_map_image_runs_items_a_shutdown_cancelled_here(self):
+        entered = threading.Barrier(auto.IMAGE_POOL_SIZE + 1)
+        release = threading.Event()
+        self.addCleanup(release.set)
+        pool = self.api._image_pool()
+        for _ in range(auto.IMAGE_POOL_SIZE):  # occupy every pool thread
+            pool.submit(lambda: (entered.wait(WAIT), release.wait(WAIT)))
+        entered.wait(WAIT)
+        here = threading.current_thread().name
+        results, submitted = {}, threading.Event()
+
+        class Counting:
+            calls = 0
+
+            def submit(self, *args):
+                future = pool.submit(*args)
+                Counting.calls += 1
+                if Counting.calls == 2:
+                    submitted.set()
+                return future
+
+        def call():
+            with patch.object(self.api, "_image_pool", return_value=Counting()):
+                results["names"] = self.api._map_image(
+                    lambda _: threading.current_thread().name, [1, 2]
+                )
+
+        caller = threading.Thread(target=call, name="map-image-caller")
+        caller.start()
+        # The caller's two items are queued behind the busy threads; the
+        # shutdown cancels them and the caller runs them itself.
+        self.assertTrue(submitted.wait(WAIT))
+        pool.shutdown(wait=False, cancel_futures=True)
+        caller.join(WAIT)
+        release.set()
+        self.assertFalse(caller.is_alive())
+        self.assertEqual(results["names"], ["map-image-caller"] * 2)
+        self.assertNotEqual(here, "map-image-caller")
+
+    async def test_run_image_after_a_close_uses_the_default_executor(self):
+        pool = self.api._image_pool()
+        pool.shutdown(wait=True)
+        with patch.object(self.api, "_image_pool", return_value=pool):
+            name = await self.api._run_image(lambda: threading.current_thread().name)
+        self.assertFalse(name.startswith(IMAGE_THREAD), name)
+
+
+class TestJobPool(ImagePoolCase):
+    def test_scan_metadata_is_read_on_the_job_pool(self):
+        paths = [
+            make_png(self.output_dir / f"s{i}.png", size=(16, 16)) for i in range(4)
+        ]
+        self.spy(self.api, "_extract_comfyui_metadata")
+        results = self.api._extract_batch_metadata_sync(paths)
+        self.assertEqual([path for path, _ in results], paths)
+        self.assert_threads(JOB_THREAD)
+
+    async def test_thumbnail_job_writes_on_the_job_pool_and_trims_after(self):
+        for index in range(3):
+            make_png(self.output_dir / f"t{index}.png", size=(64, 64))
+        self.spy(self.api, "_generate_one")
+        with patch.object(api_package, "release_free_memory") as trim:
+            job = await self.run_job("/prompt_manager/jobs/thumbnails")
+        self.assertEqual(job.state, "finished")
+        self.assert_threads(JOB_THREAD)
+        trim.assert_called_once_with()
+
+    async def test_scan_job_trims_after_it_finishes(self):
+        make_png(self.output_dir / "s.png", size=(16, 16))
+        states = []
+
+        def trim():
+            states.append(self.api._jobs.active("scan"))
+            return True
+
+        with patch.object(api_package, "release_free_memory", side_effect=trim):
+            job = await self.run_job("/prompt_manager/jobs/scan")
+        self.assertEqual(job.state, "finished")
+        self.assertGreaterEqual(len(states), 1)
+        self.assertIsNone(states[0], "trimmed after the scan finished")
+
+    async def test_a_failing_trim_never_fails_the_job(self):
+        make_png(self.output_dir / "t.png", size=(64, 64))
+        with patch.object(
+            api_package, "release_free_memory", side_effect=RuntimeError("no")
+        ):
+            job = await self.run_job("/prompt_manager/jobs/thumbnails")
+        self.assertEqual(job.state, "finished")
+
+    async def test_capture_previews_write_on_the_job_pool(self):
+        path = make_png(self.output_dir / "a.png", size=(64, 64))
+        self.spy(self.api, "_generate_one")
+        self.assertEqual(self.api.queue_capture_thumbnail(str(path)), "queued")
+        await self._idle()
+        self.assert_threads(JOB_THREAD)

tests/test_init_package.py

@@ -799,6 +799,45 @@ class TestPackageTeardown(PackageTeardownCase):
         self.assertIsNone(seen["worker"])
         self.assertTrue(all(result.values()), result)
 
+    def test_teardown_stops_the_job_pool_once_a_job_used_it(self):
+        """#523: the long-lived job pool is a teardown part of its own."""
+        package = self.load()
+        parallel = sys.modules[package.__name__ + ".utils.parallel"]
+        self.assertEqual(parallel.map_parallel(lambda n: n * 2, [1, 2], 2), [2, 4])
+        # This package's own pool (the suite's top-level one may also run).
+        threads = list(parallel._pool[1]._threads)
+        self.assertTrue(threads)
+
+        result = package._stop_background_work()
+
+        self.assertEqual(list(result), ["monitor", "thumbnails", "workers", "tracker"])
+        self.assertTrue(all(result.values()), result)
+        self.assertFalse(parallel.job_pool_started())
+        self.assertFalse(any(t.is_alive() for t in threads))
+
+    def test_teardown_reports_a_job_thread_past_its_bound(self):
+        package = self.load(require_stopped=False)
+        parallel = sys.modules[package.__name__ + ".utils.parallel"]
+        entered, release = threading.Event(), threading.Event()
+
+        def hold(_):
+            entered.set()
+            release.wait(30)
+
+        caller = threading.Thread(target=parallel.map_parallel, args=(hold, [1], 1))
+        caller.start()
+        self.addCleanup(caller.join, 30)
+        self.addCleanup(release.set)
+        self.assertTrue(entered.wait(30))
+        with (
+            mock.patch.object(parallel, "JOB_POOL_JOIN_SECONDS", 0.05),
+            self.assertLogs("prompt_manager.init", "WARNING") as logs,
+        ):
+            result = package._stop_background_work()
+        self.assertFalse(result["workers"])
+        self.assertTrue(result["tracker"])
+        self.assertRegex("\n".join(logs.output), r"not stopped in time .*workers")
+
     def test_teardown_kills_ffmpeg_and_stops_the_preview_worker(self):
         package = self.load()
         held = self.hold_a_preview(package, advance=4.0, kill_releases=True)

tests/test_native_memory.py

@@ -0,0 +1,82 @@
+"""``release_free_memory``: glibc's malloc_trim(0), a no-op elsewhere (#523)."""
+
+import sys
+import unittest
+from unittest import mock
+
+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.start()
+        self.addCleanup(patcher.stop)
+
+    @unittest.skipUnless(native_memory._glibc(), "glibc only")
+    def test_glibc_trims_once_per_call(self):
+        self.assertTrue(native_memory.release_free_memory())
+        self.assertTrue(native_memory.release_free_memory())
+
+    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
+                with mock.patch.object(native_memory.ctypes, "CDLL") as cdll:
+                    self.assertFalse(native_memory.release_free_memory())
+                cdll.assert_not_called()
+
+    def test_musl_does_nothing(self):
+        with (
+            mock.patch.object(sys, "platform", "linux"),
+            mock.patch.object(
+                native_memory.os, "confstr", side_effect=ValueError("unknown name")
+            ),
+            mock.patch.object(native_memory.ctypes, "CDLL") as cdll,
+        ):
+            self.assertFalse(native_memory.release_free_memory())
+        cdll.assert_not_called()
+
+    def test_a_missing_symbol_is_false(self):
+        library = mock.Mock(spec=[])  # no malloc_trim attribute
+        with (
+            mock.patch.object(native_memory, "_glibc", return_value=True),
+            mock.patch.object(native_memory.ctypes, "CDLL", return_value=library),
+        ):
+            self.assertFalse(native_memory.release_free_memory())
+
+    def test_an_unloadable_libc_is_false(self):
+        with (
+            mock.patch.object(native_memory, "_glibc", return_value=True),
+            mock.patch.object(native_memory.ctypes, "CDLL", side_effect=OSError),
+        ):
+            self.assertFalse(native_memory.release_free_memory())
+
+    def test_a_failing_trim_never_raises(self):
+        trim = mock.Mock(side_effect=OSError("boom"))
+        library = mock.Mock(malloc_trim=trim)
+        with (
+            mock.patch.object(native_memory, "_glibc", return_value=True),
+            mock.patch.object(native_memory.ctypes, "CDLL", return_value=library),
+        ):
+            self.assertFalse(native_memory.release_free_memory())
+        trim.assert_called_once_with(0)
+
+    def test_the_symbol_is_looked_up_once(self):
+        trim = mock.Mock(return_value=1)
+        library = mock.Mock(malloc_trim=trim)
+        with (
+            mock.patch.object(native_memory, "_glibc", return_value=True),
+            mock.patch.object(
+                native_memory.ctypes, "CDLL", return_value=library
+            ) as cdll,
+        ):
+            self.assertTrue(native_memory.release_free_memory())
+            self.assertTrue(native_memory.release_free_memory())
+        cdll.assert_called_once_with(None)
+        self.assertEqual(trim.call_count, 2)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_preview_store.py

@@ -0,0 +1,457 @@
+"""One resolver for where previews live on disk (#320 PR 1).
+
+``py/api/preview_store.py`` owns every mapping between an image source, a
+media file and its preview on disk. In source mode (the only mode so far)
+it returns exactly the paths the code built by hand before
+(``<source>/thumbnails/<rel>_thumb<ext>``), so nothing changes on disk. The
+structural test keeps new ``thumbnails`` path joins out of the rest of the
+backend.
+"""
+
+import asyncio
+import os
+import re
+import sys
+import tempfile
+import threading
+import unittest
+from pathlib import Path
+import unittest.mock
+from unittest.mock import MagicMock
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+sys.modules.setdefault("server", MagicMock())
+
+from py.api import preview_store, thumbnail_settings  # noqa: E402
+from py.api.thumbnail_encode import (  # noqa: E402
+    PreviewWriteHeld,
+    replace_unless_held,
+)
+
+REPO = Path(__file__).resolve().parent.parent
+WAIT = 10
+
+
+class SourceModeCase(unittest.TestCase):
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self._tmp.cleanup)
+        self.root = Path(self._tmp.name) / "output"
+        self.root.mkdir()
+
+
+class TestSourceModePaths(SourceModeCase):
+    def test_default_location_is_inside_each_source(self):
+        location = preview_store.current()
+        self.assertEqual(location.mode, preview_store.SOURCE_MODE)
+        self.assertTrue(location.available)
+        self.assertIsNone(location.problem)
+
+    def test_store_of_a_source_is_its_thumbnails_folder(self):
+        self.assertEqual(preview_store.store_dir(self.root), self.root / "thumbnails")
+
+    def test_preview_path_matches_the_old_join(self):
+        rel = Path("sub", "deep", "a.png")
+        self.assertEqual(
+            preview_store.preview_path(self.root, rel, ".webp"),
+            self.root / "thumbnails/sub/deep/a_thumb.webp",
+        )
+        self.assertEqual(
+            preview_store.preview_path(self.root, "a.mp4", ".jpg"),
+            self.root / "thumbnails" / "a_thumb.jpg",
+        )
+
+    def test_preview_path_defaults_to_the_written_extension(self):
+        from py.api.thumbnail_encode import writer_settings
+
+        self.assertEqual(
+            preview_store.preview_path(self.root, "a.png").suffix,
+            writer_settings().extension,
+        )
+
+    def test_request_path_maps_url_parts_to_the_old_join(self):
+        parts = ("thumbnails", "sub", "a_thumb.webp")
+        self.assertEqual(
+            preview_store.request_path(self.root, parts), self.root.joinpath(*parts)
+        )
+
+    def test_request_path_is_none_for_anything_but_preview_parts(self):
+        self.assertIsNone(preview_store.request_path(self.root, ("sub", "a.png")))
+        self.assertIsNone(preview_store.request_path(self.root, ("Thumbnails", "a")))
+        self.assertIsNone(preview_store.request_path(self.root, ("thumbnails",)))
+
+    def test_ensure_store_is_the_store(self):
+        self.assertEqual(
+            preview_store.ensure_store(self.root), self.root / "thumbnails"
+        )
+
+    def test_owns_a_preview_inside_the_store(self):
+        inside = self.root / "thumbnails" / "sub" / "a_thumb.webp"
+        self.assertTrue(preview_store.owns(self.root, inside))
+        self.assertFalse(preview_store.owns(self.root, self.root / "a.png"))
+        self.assertFalse(
+            preview_store.owns(self.root, self.root.parent / "elsewhere.webp")
+        )
+
+    @unittest.skipUnless(hasattr(os, "symlink"), "needs symlinks")
+    def test_owns_refuses_a_store_linked_out_of_the_source(self):
+        outside = Path(self._tmp.name) / "outside"
+        outside.mkdir()
+        try:
+            (self.root / "thumbnails").symlink_to(outside, target_is_directory=True)
+        except OSError:
+            self.skipTest("symlinks not permitted")
+        self.assertFalse(
+            preview_store.owns(self.root, self.root / "thumbnails" / "a_thumb.webp")
+        )
+
+    @unittest.skipUnless(hasattr(os, "symlink"), "needs symlinks")
+    def test_owns_refuses_a_preview_folder_linked_out_of_the_store(self):
+        outside = Path(self._tmp.name) / "outside"
+        outside.mkdir()
+        (self.root / "thumbnails").mkdir()
+        try:
+            (self.root / "thumbnails" / "sub").symlink_to(
+                outside, target_is_directory=True
+            )
+        except OSError:
+            self.skipTest("symlinks not permitted")
+        self.assertFalse(
+            preview_store.owns(
+                self.root, self.root / "thumbnails" / "sub" / "a_thumb.webp"
+            )
+        )
+
+    def test_store_container_is_the_source(self):
+        self.assertEqual(preview_store.store_container(self.root), self.root)
+
+    def test_given_store_argument_is_kept_unless_it_is_the_default(self):
+        default = self.root / "thumbnails"
+        other = Path(self._tmp.name) / "custom"
+        self.assertEqual(preview_store.given_store(self.root, default), default)
+        self.assertEqual(preview_store.given_store(self.root, other), other)
+
+
+class TestPreviewText(unittest.TestCase):
+    def test_preview_text_is_an_exact_thumbnails_prefix_and_preview_name(self):
+        for text in (
+            "thumbnails/a_thumb.webp",
+            "thumbnails/sub/a_thumb.jxl",
+            "thumbnails/a_thumb.JPG",
+            "thumbnails/a_thumb.Png",
+        ):
+            with self.subTest(text=text):
+                self.assertTrue(preview_store.is_preview_text(text))
+
+    def test_other_text_is_not_preview_text(self):
+        for text in (
+            "Thumbnails/a_thumb.webp",
+            "thumbnails/a.png",
+            "thumbnails/_thumb.webp",
+            "thumbnails/a_thumb.txt",
+            "sub/thumbnails/a_thumb.webp",
+            "thumbnails/../a_thumb.webp",
+            "thumbnails//a_thumb.webp",
+            "thumbnails/./a_thumb.webp",
+            "thumbnails\\a_thumb.webp",
+            "thumbnails/c:a_thumb.webp",
+            "thumbnails/sub/thumbnails/a_thumb.webp",
+            "thumbnails",
+            "",
+        ):
+            with self.subTest(text=text):
+                self.assertFalse(preview_store.is_preview_text(text))
+
+
+class TestWritesAndTheSwitch(SourceModeCase):
+    def _temp_and_dst(self):
+        folder = self.root / "thumbnails"
+        folder.mkdir()
+        tmp = folder / ".a_thumb.part-1-1.webp"
+        tmp.write_bytes(b"new")
+        return tmp, folder / "a_thumb.webp"
+
+    def test_a_write_from_the_current_generation_lands(self):
+        tmp, dst = self._temp_and_dst()
+        replace_unless_held(tmp, dst, preview_store.write_guard())
+        self.assertEqual(dst.read_bytes(), b"new")
+
+    def test_a_write_whose_generation_changed_is_dropped(self):
+        tmp, dst = self._temp_and_dst()
+        guard = preview_store.write_guard()
+        preview_store.bump_generation()
+        with self.assertRaises(PreviewWriteHeld):
+            replace_unless_held(tmp, dst, guard)
+        self.assertFalse(dst.exists())
+
+    def test_the_hold_check_still_applies(self):
+        tmp, dst = self._temp_and_dst()
+        with self.assertRaises(PreviewWriteHeld):
+            replace_unless_held(tmp, dst, preview_store.write_guard(lambda: False))
+        self.assertFalse(dst.exists())
+
+    def test_the_check_and_the_rename_run_under_the_write_lock(self):
+        """The hold check and the rename hold ``write_lock``, the lock the
+        generation bump takes: a switch cannot fall between them."""
+        tmp, dst = self._temp_and_dst()
+        in_check = threading.Event()
+        release = threading.Event()
+        seen = {}
+
+        def slow_hold():
+            seen["locked"] = preview_store.write_lock.locked()
+            in_check.set()
+            release.wait(WAIT)
+            return True
+
+        guard = preview_store.write_guard(slow_hold)
+        writer = threading.Thread(target=replace_unless_held, args=(tmp, dst, guard))
+        writer.start()
+        try:
+            self.assertTrue(in_check.wait(WAIT))
+            self.assertTrue(seen["locked"])
+            self.assertTrue(preview_store.write_lock.locked())
+        finally:
+            release.set()
+            writer.join(WAIT)
+        self.assertFalse(preview_store.write_lock.locked())
+        self.assertEqual(dst.read_bytes(), b"new")
+        bumped = preview_store.bump_generation()
+        self.assertEqual(preview_store.generation(), bumped)
+
+
+class TestOnlyTheResolverBuildsStorePaths(unittest.TestCase):
+    """No backend module joins ``thumbnails`` onto a path by hand any more."""
+
+    ALLOWED = {
+        Path("py/api/preview_store.py"),
+        Path("py/api/thumbnail_settings.py"),  # URL text
+    }
+    # A path join with a literal "thumbnails" operand.
+    JOIN = re.compile(
+        r"""/\s*["']thumbnails["']"""
+        r"""|["']thumbnails["']\s*/"""
+        r"""|joinpath\([^)]*["']thumbnails["']"""
+        r"""|os\.path\.join\([^)]*["']thumbnails["']"""
+        r"""|["']thumbnails/"""
+        r"""|/\s*PREVIEW_FOLDER|joinpath\([^)]*PREVIEW_FOLDER"""
+        r"""|Path\([^)]*,\s*["']thumbnails["']"""
+    )
+
+    def test_no_thumbnails_join_outside_the_resolver(self):
+        offenders = []
+        files = sorted(REPO.glob("*.py"))
+        for folder in ("py", "utils", "database"):
+            files += sorted((REPO / folder).rglob("*.py"))
+        for path in files:
+            rel = path.relative_to(REPO)
+            if rel in self.ALLOWED:
+                continue
+            for number, line in enumerate(
+                path.read_text(encoding="utf-8").splitlines(), 1
+            ):
+                code = line.split("#", 1)[0]
+                if self.JOIN.search(code):
+                    offenders.append(f"{rel}:{number}: {line.strip()}")
+        self.assertEqual(offenders, [])
+
+
+class TestSettingsLookupsFollowTheStore(SourceModeCase):
+    def test_owned_thumbnails_reads_the_store(self):
+        folder = self.root / "thumbnails" / "sub"
+        folder.mkdir(parents=True)
+        (folder / "a_thumb.webp").write_bytes(b"x")
+        self.assertEqual(
+            thumbnail_settings.owned_thumbnails(self.root, Path("sub", "a.png")),
+            [folder / "a_thumb.webp"],
+        )
+
+
+try:
+    from tests.test_thumbnail_auto import AutoThumbnailCase, _entry
+    from tests.test_api_images import make_png
+except ImportError:  # discovered with tests/ as the top-level directory
+    from test_thumbnail_auto import AutoThumbnailCase, _entry
+    from test_api_images import make_png
+
+
+class RedirectedStoreCase(AutoThumbnailCase):
+    """Every caller follows the resolver: with the store of the output root
+    moved to ``<tmp>/elsewhere/output-key`` (a stand-in for a later mode),
+    previews are read, written, listed and removed there and only there."""
+
+    async def asyncSetUp(self):
+        await super().asyncSetUp()
+        self.container = self.comfy_dir / "elsewhere"
+        self.store = self.container / "output-key"
+        self.store.mkdir(parents=True)
+        self.old_store = self.output_dir / "thumbnails"
+        root = self.output_dir
+        real_store = preview_store.store_dir
+        real_container = preview_store.store_container
+
+        def store_dir(r, location=None):
+            return self.store if Path(r) == root else real_store(r, location)
+
+        def store_container(r, location=None):
+            if Path(r) == root:
+                return self.container
+            return real_container(r, location)
+
+        for name, fake in (
+            ("store_dir", store_dir),
+            ("store_container", store_container),
+        ):
+            patcher = unittest.mock.patch.object(preview_store, name, fake)
+            patcher.start()
+            self.addCleanup(patcher.stop)
+
+    def _preview(self, rel, ext=".webp"):
+        return self.store / (thumbnail_settings.preview_stem(rel) + ext)
+
+
+class TestEveryCallerFollowsTheStore(RedirectedStoreCase):
+    async def test_an_existing_preview_is_served_from_the_store(self):
+        make_png(self.output_dir / "sub" / "a.png", size=(64, 64))
+        preview = self._preview("sub/a.png")
+        make_png(preview, size=(32, 32))
+        url = _entry(self.api, self.output_dir, "sub/a.png")["thumbnail_url"]
+        self.assertIn("v=", url)  # the listing found it in the store
+        resp = await self.client.get(url)
+        self.assertEqual(resp.status, 200)
+        self.assertEqual(await resp.read(), preview.read_bytes())
+
+    async def test_a_legacy_upper_case_preview_is_served_from_the_store(self):
+        make_png(self.output_dir / "k.JPG", size=(64, 64))
+        legacy = self.store / "k_thumb.JPG"
+        legacy.write_bytes(b"legacy preview")
+        resp = await self.client.get(
+            "/prompt_manager/images/serve/thumbnails/k_thumb.JPG?root=0"
+        )
+        self.assertEqual(resp.status, 200)
+        self.assertEqual(await resp.read(), b"legacy preview")
+
+    async def test_a_missing_preview_is_made_in_the_store_on_first_view(self):
+        make_png(self.output_dir / "sub" / "b.png", size=(640, 480))
+        url = _entry(self.api, self.output_dir, "sub/b.png")["thumbnail_url"]
+        resp = await self.client.get(url)
+        self.assertEqual(resp.status, 200)
+        self.assertTrue(self._preview("sub/b.png").is_file())
+        self.assertFalse(self.old_store.exists())
+
+    async def test_a_leftover_in_the_source_folder_is_not_served(self):
+        make_png(self.output_dir / "c.png", size=(640, 480))
+        leftover = self.old_store / "c_thumb.webp"
+        leftover.parent.mkdir()
+        leftover.write_bytes(b"old preview bytes")
+        resp = await self.client.get(
+            "/prompt_manager/images/serve/thumbnails/c_thumb.webp?root=0"
+        )
+        self.assertEqual(resp.status, 200)
+        self.assertNotEqual(await resp.read(), b"old preview bytes")
+        self.assertTrue(self._preview("c.png").is_file())
+
+    async def test_capture_writes_into_the_store(self):
+        path = make_png(self.output_dir / "run" / "d.png", size=(64, 64))
+        self.assertEqual(self.api.queue_capture_thumbnail(str(path)), "queued")
+        await self._idle()
+        self.assertTrue(self._preview("run/d.png").is_file())
+        self.assertFalse(self.old_store.exists())
+
+    async def test_generate_and_clear_follow_the_store(self):
+        make_png(self.output_dir / "e.png", size=(64, 64))
+        payload = await asyncio.get_running_loop().run_in_executor(
+            None, self.api._generate_gallery_thumbnails_sync, [self.output_dir]
+        )
+        self.assertTrue(payload["success"])
+        self.assertTrue(self._preview("e.png").is_file())
+        self.assertFalse(self.old_store.exists())
+        self.assertTrue(payload["thumbnails_path"].endswith("output-key"))
+        cleared, _size = self.api._clear_gallery_thumbnails_sync([self.output_dir])
+        self.assertEqual(cleared, 1)
+        self.assertFalse(self._preview("e.png").exists())
+        self.assertTrue(self.store.is_dir())  # the store itself stays
+
+    async def test_removing_a_file_removes_its_preview_in_the_store(self):
+        from py.api.admin import _remove_thumbnail
+
+        make_png(self.output_dir / "f.png", size=(64, 64))
+        preview = self._preview("f.png")
+        preview.write_bytes(b"x")
+        self.assertTrue(_remove_thumbnail(self.output_dir, Path("f.png")))
+        self.assertFalse(preview.exists())
+
+    async def test_the_temp_sweep_walks_the_store(self):
+        stray = self.store / ".g_thumb.part-1-1.webp"
+        stray.write_bytes(b"x")
+        old = os.path.getmtime(stray) - 10 * 24 * 3600
+        os.utime(stray, (old, old))
+        self.assertEqual(self.api.sweep_thumbnail_temps(), 1)
+        self.assertFalse(stray.exists())
+
+    async def test_history_cover_writes_into_the_store(self):
+        from py.api import history_current
+
+        self.assertTrue(
+            history_current._preview_writable(self.output_dir, self._preview("h.png"))
+        )
+        self.assertFalse(
+            history_current._preview_writable(
+                self.output_dir, self.old_store / "h_thumb.webp"
+            )
+        )
+
+
+class TestCaptureResolvesAtWriteTime(RedirectedStoreCase):
+    def test_a_queued_capture_follows_a_store_that_moved_before_its_write(self):
+        from py.api import thumbnail_auto as auto
+
+        written = []
+        thumbnailer = auto.AutoThumbnailer(
+            lambda src, dst, *a, **k: written.append(dst) or {"action": "generated"},
+            lambda: True,
+            self.api.logger,
+            start_worker=False,
+        )
+        self.addCleanup(thumbnailer.close)
+        src = make_png(self.output_dir / "i.png", size=(32, 32))
+        deferred = auto.DeferredPreview(self.output_dir, Path("i.png"))
+        self.assertTrue(thumbnailer.enqueue(src, deferred, False))
+        moved = self.container / "moved"
+        moved.mkdir()
+        self.store = moved  # the location changes before the worker runs
+        thumbnailer._write_queued(thumbnailer._queue.popleft())
+        self.assertEqual(written, [moved / "i_thumb.webp"])
+
+    def test_the_after_scan_queue_maps_each_target_to_its_own_root(self):
+        from py.api import thumbnail_auto as auto
+
+        nested = self.output_dir / "inner"
+        nested.mkdir()
+        src = nested / "j.png"
+        outer = auto._deferred(
+            [self.output_dir, nested], src, self._preview("inner/j.png")
+        )
+        inner = auto._deferred(
+            [self.output_dir, nested], src, nested / "thumbnails" / "j_thumb.webp"
+        )
+        self.assertEqual(
+            (outer.root, outer.rel), (self.output_dir, Path("inner/j.png"))
+        )
+        self.assertEqual((inner.root, inner.rel), (nested, Path("j.png")))
+        stray = self.comfy_dir / "x_thumb.webp"
+        self.assertEqual(auto._deferred([self.output_dir], src, stray), stray)
+
+    def test_serve_and_history_share_one_flight_key(self):
+        from py.api import thumbnail_auto as auto
+
+        parts = ("thumbnails", "sub", "a_thumb.webp")
+        self.assertEqual(
+            auto._request_key(self.output_dir, parts),
+            auto._dst_key(self._preview("sub/a.png")),
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_thumbnail_auto.py

@@ -291,7 +291,7 @@ class TestThumbnailOnFirstView(AutoThumbnailCase):
         await self._writers(state, auto.ON_DEMAND_LIMIT)
         await self._waiting(len(names))
         self.assertEqual(default_calls, [], "no view waits on a default thread")
-        self.assertTrue(all(name.startswith("pm-preview") for name in state["threads"]))
+        self.assertTrue(all(name.startswith("pm-image") for name in state["threads"]))
         status = await self.client.get("/prompt_manager/images/thumbnails/status")
         self.assertEqual(status.status, 200, "another route answers meanwhile")
         await status.read()
@@ -678,7 +678,7 @@ class TestShutdown(AutoThumbnailCase):
             _entry(self.api, self.output_dir, "viewed.png")["thumbnail_url"]
         )
         await resp.read()
-        pool_threads = list(self.api._preview_pool_threads())
+        pool_threads = list(self.api._image_pool_threads())
         self.assertTrue(pool_threads)
         loop = asyncio.get_running_loop()
         await loop.run_in_executor(
@@ -904,7 +904,8 @@ class TestThumbnailOnCapture(AutoThumbnailCase):
             self.api.queue_capture_thumbnail(str(path))
         release.set()
         await self._idle()
-        self.assertEqual(batches, [(1, 1), (2, 2), (2, 2)])
+        # The job pool's size is the setting; each batch is at most that (#523).
+        self.assertEqual(batches, [(1, 2), (2, 2), (2, 2)])
         self.assertLessEqual(state["peak"], 2)
         for i in range(5):
             self.assertTrue(

tests/test_utils_parallel.py

@@ -1,4 +1,8 @@
-"""Tests for utils/parallel.py: bounded thread-pool map used by long jobs."""
+"""Tests for utils/parallel.py: bounded thread-pool map used by long jobs.
+
+The jobs share one long-lived pool of Worker threads (#523): its threads (and
+their malloc arenas) are reused from job to job instead of a new set per call.
+"""
 
 import os
 import sys
@@ -8,10 +12,22 @@ import unittest
 
 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 
-from utils.parallel import map_parallel  # noqa: E402
+from utils import parallel  # noqa: E402
+from utils.parallel import JOB_THREAD_PREFIX, map_parallel  # noqa: E402
+
+WAIT = 30
+
+
+def _job_threads():
+    return {
+        t for t in threading.enumerate() if t.name.startswith(JOB_THREAD_PREFIX + "_")
+    }
 
 
 class TestMapParallel(unittest.TestCase):
+    def setUp(self):
+        self.assertTrue(parallel.shutdown_job_pool(WAIT))
+        self.addCleanup(parallel.shutdown_job_pool, WAIT)
 
     def test_results_keep_input_order(self):
         def slow_square(n):
@@ -22,12 +38,102 @@ class TestMapParallel(unittest.TestCase):
             map_parallel(slow_square, [1, 2, 3, 4], workers=4), [1, 4, 9, 16]
         )
 
-    def test_one_worker_runs_on_the_calling_thread(self):
+    def test_one_worker_runs_on_the_one_job_thread(self):
         names = []
         map_parallel(
             lambda _: names.append(threading.current_thread().name), [1, 2, 3], 1
         )
-        self.assertEqual(set(names), {threading.current_thread().name})
+        self.assertEqual(len(set(names)), 1)
+        self.assertTrue(names[0].startswith(JOB_THREAD_PREFIX + "_"), names)
+
+    def test_jobs_reuse_the_same_long_lived_threads(self):
+        """Arenas are reused instead of a new set of threads per job."""
+        first, second = set(), set()
+        map_parallel(lambda _: first.add(threading.current_thread()), range(8), 2)
+        map_parallel(lambda _: second.add(threading.current_thread()), range(8), 2)
+        self.assertTrue(second <= first, (first, second))
+        self.assertLessEqual(len(_job_threads()), 2)
+        self.assertTrue(all(t.is_alive() for t in first | second))
+
+    def test_a_new_worker_count_replaces_the_pool(self):
+        map_parallel(lambda n: n, range(4), 2)
+        old = _job_threads()
+        map_parallel(lambda n: n, range(4), 3)
+        for thread in old:
+            thread.join(WAIT)
+            self.assertFalse(thread.is_alive(), thread.name)
+        self.assertLessEqual(len(_job_threads()), 3)
+
+    def test_shutdown_also_joins_a_pool_a_resize_retired(self):
+        entered, release = threading.Event(), threading.Event()
+
+        def hold(_):
+            entered.set()
+            release.wait(WAIT)
+
+        caller = threading.Thread(target=map_parallel, args=(hold, [1], 1))
+        caller.start()
+        self.addCleanup(caller.join, WAIT)
+        self.addCleanup(release.set)
+        self.assertTrue(entered.wait(WAIT))
+        map_parallel(lambda n: n, [1, 2], 2)  # a new size retires the busy pool
+        self.assertFalse(parallel.shutdown_job_pool(0.05))
+        release.set()
+        caller.join(WAIT)
+        self.assertTrue(parallel.shutdown_job_pool(WAIT))
+        self.assertFalse(parallel.job_pool_started())
+
+    def test_an_item_run_from_a_job_thread_runs_inline(self):
+        """A nested map never waits for the threads it is running on."""
+
+        def outer(n):
+            return map_parallel(lambda m: (m, threading.current_thread()), [n], 1)
+
+        results = map_parallel(outer, [1, 2], 1)
+        self.assertEqual([r[0][0] for r in results], [1, 2])
+
+    def test_shutdown_stops_the_threads_and_a_later_job_starts_fresh(self):
+        map_parallel(lambda n: n, range(4), 2)
+        threads = _job_threads()
+        self.assertTrue(threads)
+        self.assertTrue(parallel.job_pool_started())
+        self.assertTrue(parallel.shutdown_job_pool(WAIT))
+        self.assertFalse(parallel.job_pool_started())
+        self.assertFalse(any(t.is_alive() for t in threads))
+        self.assertEqual(map_parallel(lambda n: n + 1, [1], 2), [2])
+
+    def test_shutdown_reports_a_thread_still_running(self):
+        entered, release = threading.Event(), threading.Event()
+
+        def hold(_):
+            entered.set()
+            release.wait(WAIT)
+
+        caller = threading.Thread(target=map_parallel, args=(hold, [1], 1))
+        caller.start()
+        self.assertTrue(entered.wait(WAIT))
+        self.assertFalse(parallel.shutdown_job_pool(0.05))
+        release.set()
+        caller.join(WAIT)
+        self.assertFalse(caller.is_alive())
+
+    def test_a_failing_item_waits_for_the_running_ones(self):
+        """The caller sees the error only after its other items stopped."""
+        running = threading.Event()
+        finished = []
+
+        def item(n):
+            if n == 0:
+                self.assertTrue(running.wait(WAIT))
+                raise ValueError("zero")
+            running.set()
+            time.sleep(0.05)
+            finished.append(n)
+            return n
+
+        with self.assertRaises(ValueError):
+            map_parallel(item, [0, 1], 2)
+        self.assertEqual(finished, [1])
 
     def test_several_workers_use_several_threads(self):
         names = set()

tools/perf/memory_drive.py

@@ -0,0 +1,395 @@
+"""Measure the server's resident memory under a busy Images/Library session (#523).
+
+Manual only, Linux only (it reads ``/proc`` and glibc's ``malloc_info``); never
+run in CI. It starts the production API inside the disposable browser test
+server (``tests/browser/server.py``: its own temporary library, database and
+output folder; never a live ComfyUI) and drives it for ``--minutes``:
+
+- ``--clients`` threads loop over: a workspace grid page of 100 images, 40 of
+  its thumbnails, 6 inspector reads and 6 full-file reads, the Library list
+  with 3 entries' history and images, and the recent prompt and image lists;
+- every minute, ``--batch`` new PNGs (832 to 1536 px, ComfyUI-shaped
+  metadata) are copied into the output folder, then a scan job and a
+  thumbnail job run to completion.
+
+Every 5 s the server samples its RSS, Python heap (tracemalloc), threads and
+glibc arena count. The test server pins glibc's mmap threshold (#516); this
+driver turns that off so the production allocator behaviour is measured,
+unless ``--mallopt`` asks for the pinned comparison run.
+
+    python tools/perf/memory_drive.py --label after --minutes 12
+    python tools/perf/memory_drive.py --repo /path/to/other/checkout --label before
+
+The checkout under ``--repo`` (default: this one) is the code measured; the
+driver itself only needs Pillow. The last line printed is a Markdown table row.
+"""
+
+import argparse
+import ctypes
+import json
+import os
+import random
+import shutil
+import subprocess  # nosec B404 - starts this script's own server child
+import sys
+import tempfile
+import threading
+import time
+import tracemalloc
+import urllib.parse
+import urllib.request
+from concurrent.futures import ThreadPoolExecutor
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+REPO = os.path.dirname(os.path.dirname(HERE))
+PORT_ANNOUNCEMENT = "PM_TEST_SERVER_PORT="
+OUTPUT_ANNOUNCEMENT = "output="
+SAMPLE_SECONDS = 5
+MEDIA_FILES = 120
+MEDIA_SIDES = (832, 1024, 1216, 1536)
+REQUEST_TIMEOUT = 120
+JOB_POLL_LIMIT = 600
+FINAL_STATES = ("finished", "cancelled", "failed")
+M_MMAP_THRESHOLD = -3
+
+
+# ── the server child ────────────────────────────────────────────────────
+
+
+def _arena_count(libc):
+    """glibc's arena count (``<heap nr=`` entries of ``malloc_info``)."""
+    buffer, size = ctypes.c_char_p(), ctypes.c_size_t()
+    stream = libc.open_memstream(ctypes.byref(buffer), ctypes.byref(size))
+    libc.malloc_info(0, stream)
+    libc.fclose(stream)
+    text = ctypes.string_at(buffer, size.value).decode()
+    libc.free(ctypes.cast(buffer, ctypes.c_void_p))
+    return text.count("<heap nr=")
+
+
+def _status():
+    with open("/proc/self/status", encoding="ascii") as status:
+        return {
+            line.split(":")[0]: int(line.split()[1])
+            for line in status
+            if line.startswith(("VmRSS", "RssAnon", "Threads"))
+        }
+
+
+def _sample_forever(libc, samples_path):
+    start = time.monotonic()
+    announced = False
+    with open(samples_path, "a", buffering=1, encoding="utf-8") as out:
+        while True:
+            folder_paths = sys.modules.get("folder_paths")
+            if not announced and hasattr(folder_paths, "get_output_directory"):
+                out.write(OUTPUT_ANNOUNCEMENT + folder_paths.get_output_directory())
+                out.write("\n")
+                announced = True
+            status = _status()
+            out.write(
+                json.dumps(
+                    {
+                        "t": round(time.monotonic() - start),
+                        "rss": status["VmRSS"] >> 10,
+                        "anon": status["RssAnon"] >> 10,
+                        "heap": tracemalloc.get_traced_memory()[0] >> 20,
+                        "threads": status["Threads"],
+                        "arenas": _arena_count(libc),
+                    }
+                )
+                + "\n"
+            )
+            time.sleep(SAMPLE_SECONDS if announced else 0.5)
+
+
+def serve(repo, samples_path, mallopt):
+    """Run the browser test server of *repo* in this process, sampled."""
+    sys.path.insert(0, repo)
+    os.chdir(repo)
+    tracemalloc.start(1)
+    libc = ctypes.CDLL(None)
+    libc.open_memstream.restype = ctypes.c_void_p
+    libc.open_memstream.argtypes = [
+        ctypes.POINTER(ctypes.c_char_p),
+        ctypes.POINTER(ctypes.c_size_t),
+    ]
+    libc.fclose.argtypes = [ctypes.c_void_p]
+    libc.malloc_info.argtypes = [ctypes.c_int, ctypes.c_void_p]
+    libc.free.argtypes = [ctypes.c_void_p]
+    if mallopt:
+        libc.mallopt(M_MMAP_THRESHOLD, 128 * 1024)
+    threading.Thread(
+        target=_sample_forever, args=(libc, samples_path), daemon=True
+    ).start()
+    from tests.browser import server
+
+    if not mallopt:
+        server.pin_mmap_threshold = lambda: False  # production allocator
+    server.main(["--port", "0", "--exit-on-stdin-eof"])
+
+
+# ── the driver ──────────────────────────────────────────────────────────
+
+
+def make_media(folder, count=MEDIA_FILES, seed=523):
+    """Synthetic ComfyUI-shaped PNGs (smooth noise, prompt and workflow)."""
+    from PIL import Image, PngImagePlugin
+
+    rng = random.Random(seed)
+    os.makedirs(folder, exist_ok=True)
+    for index in range(count):
+        side = MEDIA_SIDES[index % len(MEDIA_SIDES)]
+        small = Image.frombytes(
+            "RGB", (64, 64), bytes(rng.randrange(256) for _ in range(64 * 64 * 3))
+        )
+        image = small.resize((side, side), Image.BICUBIC)
+        text = f"synthetic scene {index}, soft light, detailed"
+        prompt = {
+            "3": {
+                "class_type": "KSampler",
+                "inputs": {
+                    "seed": index,
+                    "steps": 20,
+                    "cfg": 7.0,
+                    "sampler_name": "euler",
+                    "scheduler": "normal",
+                    "denoise": 1.0,
+                    "model": ["4", 0],
+                    "positive": ["6", 0],
+                    "negative": ["7", 0],
+                    "latent_image": ["5", 0],
+                },
+            },
+            "4": {
+                "class_type": "CheckpointLoaderSimple",
+                "inputs": {"ckpt_name": "model.safetensors"},
+            },
+            "5": {
+                "class_type": "EmptyLatentImage",
+                "inputs": {"width": side, "height": side, "batch_size": 1},
+            },
+            "6": {
+                "class_type": "CLIPTextEncode",
+                "inputs": {"text": text, "clip": ["4", 1]},
+            },
+            "7": {
+                "class_type": "CLIPTextEncode",
+                "inputs": {"text": "blurry", "clip": ["4", 1]},
+            },
+            "9": {
+                "class_type": "SaveImage",
+                "inputs": {"filename_prefix": "img", "images": ["8", 0]},
+            },
+        }
+        workflow = {
+            "nodes": [
+                {"id": int(k), "type": v["class_type"]} for k, v in prompt.items()
+            ]
+        }
+        info = PngImagePlugin.PngInfo()
+        info.add_text("prompt", json.dumps(prompt))
+        info.add_text("workflow", json.dumps(workflow))
+        image.save(
+            os.path.join(folder, f"m{index:03d}.png"), pnginfo=info, compress_level=4
+        )
+
+
+class Driver:
+    def __init__(self, base, output, media, batch):
+        self.base, self.output, self.media, self.batch = base, output, media, batch
+        self.errors, self.batches = [], 0
+        self.names = sorted(os.listdir(media))
+
+    def call(self, method, path, body=None, raw=False):
+        data = None if body is None else json.dumps(body).encode()
+        request = urllib.request.Request(  # nosec B310 - loopback test server
+            self.base + path,
+            data=data,
+            method=method,
+            headers={"Content-Type": "application/json"},
+        )
+        try:
+            with urllib.request.urlopen(  # nosec B310
+                request, timeout=REQUEST_TIMEOUT
+            ) as response:
+                payload = response.read()
+        except Exception as error:  # keep driving; errors are counted
+            self.errors.append(f"{method} {path.split('?')[0]}: {error}")
+            return None
+        return payload if raw else json.loads(payload or b"{}")
+
+    def add_batch(self):
+        self.batches += 1
+        folder = os.path.join(self.output, f"batch{self.batches:03d}")
+        os.makedirs(folder)
+        for index in range(self.batch):
+            name = self.names[(self.batches * self.batch + index) % len(self.names)]
+            shutil.copyfile(
+                os.path.join(self.media, name),
+                os.path.join(folder, f"img_{self.batches:03d}_{index:03d}.png"),
+            )
+
+    def wait_job(self, response):
+        job = ((response or {}).get("job") or {}).get("id")
+        for _ in range(JOB_POLL_LIMIT if job else 0):
+            state = self.call("GET", f"/prompt_manager/jobs/{job}")
+            if not state or (state.get("job") or {}).get("state") in FINAL_STATES:
+                return
+            time.sleep(1)
+
+    def scan_and_thumbnails(self):
+        self.add_batch()
+        self.wait_job(self.call("POST", "/prompt_manager/jobs/scan", {}))
+        self.wait_job(self.call("POST", "/prompt_manager/jobs/thumbnails", {}))
+
+    def browse(self, _index=None):
+        offset = random.randint(0, 3) * 100  # nosec B311 - load shape only
+        page = self.call(
+            "GET",
+            f"/prompt_manager/images/output?mode=workspace&limit=100&offset={offset}",
+        )
+        images = [i for i in (page or {}).get("images") or [] if isinstance(i, dict)]
+        urls = [i.get("thumbnail_url") or i.get("url") for i in images]
+        urls = [u for u in urls if u]
+        for url in random.sample(urls, min(40, len(urls))):
+            self.call("GET", url, raw=True)
+        for image in random.sample(images, min(6, len(images))):
+            self.inspect(image)
+            if image.get("url"):
+                self.call("GET", image["url"], raw=True)
+        library = self.call("GET", "/prompt_manager/library?limit=100") or {}
+        entries = [e for e in library.get("items") or [] if isinstance(e, dict)]
+        for entry in random.sample(entries, min(3, len(entries))):
+            prompt_id = entry.get("id") or entry.get("positive_id")
+            if prompt_id is not None:
+                self.call(
+                    "GET", f"/prompt_manager/library/{prompt_id}/history", raw=True
+                )
+                self.call(
+                    "GET", f"/prompt_manager/library/{prompt_id}/images", raw=True
+                )
+        self.call("GET", "/prompt_manager/recent?limit=50", raw=True)
+        self.call("GET", "/prompt_manager/images/recent?limit=50", raw=True)
+
+    def inspect(self, image):
+        relative, index = image.get("path"), image.get("root_index")
+        if not relative or not isinstance(index, int):
+            return
+        query = {"root": str(index)}
+        if image.get("root_scope"):
+            query["root_scope"] = image["root_scope"]
+        self.call(
+            "GET",
+            "/prompt_manager/images/inspect/"
+            + urllib.parse.quote(relative)
+            + "?"
+            + urllib.parse.urlencode(query),
+            raw=True,
+        )
+
+
+def _wait_for_output(samples_path, process):
+    while process.poll() is None:
+        with open(samples_path, encoding="utf-8") as samples:
+            for line in samples:
+                if line.startswith(OUTPUT_ANNOUNCEMENT):
+                    return line[len(OUTPUT_ANNOUNCEMENT) :].strip()
+        time.sleep(0.2)
+    raise RuntimeError("the server exited before it started")
+
+
+def summarize(samples_path, label, batches, errors):
+    rows = []
+    with open(samples_path, encoding="utf-8") as samples:
+        for line in samples:
+            if line.startswith("{"):
+                rows.append(json.loads(line))
+    if not rows:
+        raise RuntimeError("no samples were taken")
+    rss = [row["rss"] for row in rows]
+    late = [row for row in rows if row["t"] >= 120] or rows
+    end = rows[-1]
+    return (
+        f"| {label} | {rss[0]} → {end['rss']} | {min(r['rss'] for r in late)}–"
+        f"{max(r['rss'] for r in late)} | {max(rss)} | {rows[0]['heap']} → "
+        f"{end['heap']} | {max(r['threads'] for r in rows)} | "
+        f"{max(r['arenas'] for r in rows)} | {batches} | {len(errors)} |"
+    )
+
+
+def drive(args):
+    work = args.out or tempfile.mkdtemp(prefix="pm-memory-")
+    media = os.path.join(work, "media")
+    if not os.path.isdir(media):
+        print(f"Writing {MEDIA_FILES} PNGs to {media}", flush=True)
+        make_media(media)
+    samples_path = os.path.join(work, f"{args.label}.samples")
+    open(samples_path, "w", encoding="utf-8").close()
+    command = [sys.executable, os.path.abspath(__file__), "--serve", samples_path]
+    command += ["--repo", os.path.abspath(args.repo)]
+    command += ["--mallopt"] if args.mallopt else []
+    with open(os.path.join(work, f"{args.label}.stderr"), "w") as stderr:
+        process = subprocess.Popen(  # nosec B603
+            command, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=stderr
+        )
+        try:
+            port = None
+            for line in process.stdout:
+                text = line.decode(errors="replace")
+                if text.startswith(PORT_ANNOUNCEMENT):
+                    port = int(text[len(PORT_ANNOUNCEMENT) :])
+                    break
+            if port is None:
+                raise RuntimeError("the server announced no port")
+            output = _wait_for_output(samples_path, process)
+            driver = Driver(f"http://127.0.0.1:{port}", output, media, args.batch)
+            run(driver, args)
+        finally:
+            process.stdin.close()
+            process.wait(60)
+    print(f"Samples: {samples_path}")
+    print(
+        "| run | RSS start → end (MB) | after 2 min | max | heap (MB) | threads "
+        "| arenas | batches | errors |"
+    )
+    print(summarize(samples_path, args.label, driver.batches, driver.errors))
+
+
+def run(driver, args):
+    driver.call("POST", "/__test__/reset", {})
+    driver.scan_and_thumbnails()
+    deadline = time.monotonic() + args.minutes * 60
+    next_batch = time.monotonic() + 60
+    with ThreadPoolExecutor(args.clients) as pool:
+        while time.monotonic() < deadline:
+            list(pool.map(driver.browse, range(args.clients)))
+            if time.monotonic() > next_batch:
+                driver.scan_and_thumbnails()
+                next_batch = time.monotonic() + 60
+    time.sleep(SAMPLE_SECONDS + 1)  # one more sample after the load stops
+
+
+def main(argv=None):
+    parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
+    parser.add_argument("--repo", default=REPO, help="checkout to measure")
+    parser.add_argument("--label", default="run")
+    parser.add_argument("--minutes", type=float, default=12)
+    parser.add_argument("--clients", type=int, default=6)
+    parser.add_argument("--batch", type=int, default=60, help="new PNGs a minute")
+    parser.add_argument("--out", help="work folder (media, samples); default temp")
+    parser.add_argument(
+        "--mallopt", action="store_true", help="pin the mmap threshold (comparison)"
+    )
+    parser.add_argument("--serve", metavar="SAMPLES", help=argparse.SUPPRESS)
+    args = parser.parse_args(argv)
+    if not sys.platform.startswith("linux"):
+        parser.error("Linux only: it reads /proc and glibc's malloc_info")
+    if args.serve:
+        serve(os.path.abspath(args.repo), args.serve, args.mallopt)
+    else:
+        drive(args)
+
+
+if __name__ == "__main__":
+    main()

utils/native_memory.py

@@ -0,0 +1,65 @@
+"""Give glibc's free heap pages back to the OS after a long image job (#523).
+
+Image work (decode, resize, encode) leaves image-sized free blocks in the
+malloc arenas of the threads that did it; glibc keeps those pages resident.
+``release_free_memory`` calls ``malloc_trim(0)`` once, which returns the free
+pages of every arena and changes no allocator setting (no ``mallopt``, no
+``MALLOC_ARENA_MAX``: ComfyUI owns this process's allocator).
+
+glibc only: on Windows, macOS and musl (Alpine) it does nothing. It never
+raises; a missing or failing ``malloc_trim`` is a quiet ``False``.
+"""
+
+import ctypes
+import os
+import sys
+import threading
+
+_lock = threading.Lock()
+_trim = None
+_looked_up = False
+
+
+def _glibc() -> bool:
+    """Whether this process runs on glibc (musl has no CS_GNU_LIBC_VERSION)."""
+    if not sys.platform.startswith("linux"):
+        return False
+    try:
+        version = os.confstr("CS_GNU_LIBC_VERSION")
+    except (AttributeError, ValueError, OSError):
+        return False
+    return bool(version) and version.startswith("glibc")
+
+
+def _malloc_trim():
+    """glibc's ``malloc_trim``, looked up once; None where there is none."""
+    global _trim, _looked_up
+    with _lock:
+        if not _looked_up:
+            _looked_up = True
+            _trim = _lookup()
+        return _trim
+
+
+def _lookup():
+    if not _glibc():
+        return None
+    try:
+        trim = ctypes.CDLL(None).malloc_trim
+    except (OSError, AttributeError):
+        return None
+    trim.argtypes = [ctypes.c_size_t]
+    trim.restype = ctypes.c_int
+    return trim
+
+
+def release_free_memory() -> bool:
+    """Return free heap pages to the OS once (glibc); True when it ran."""
+    trim = _malloc_trim()
+    if trim is None:
+        return False
+    try:
+        trim(0)
+    except Exception:  # a foreign libc symbol: never fail the caller's job
+        return False
+    return True

utils/parallel.py

@@ -15,18 +15,36 @@ create new thread at interpreter shutdown") only once the interpreter is
 marked finalizing, after the ``atexit`` callbacks. ``refused_at_shutdown``
 recognises both; ``note_refusal`` logs one DEBUG line per place, with no
 traceback.
+
+One long-lived pool (#523): every ``map_parallel`` call shares one pool of
+``workers`` threads (``JOB_THREAD_PREFIX``), kept between jobs, instead of a
+new pool per call. Image work then reuses the same few threads, and their
+malloc arenas, rather than leaving each call's image-sized buffers in a new
+set of arenas. A different ``workers`` count (the Worker threads setting
+changed) replaces the pool; the old one finishes what it holds and exits.
+``shutdown_job_pool`` stops it (the package teardown); the next call starts
+a fresh one.
 """
 
 import logging
 import threading
-from concurrent.futures import ThreadPoolExecutor
+import time
+from concurrent.futures import ThreadPoolExecutor, wait
 
 from .exit_signal import checked
 
 SHUTDOWN_REFUSAL = "interpreter shutdown"
+# ``submit`` on a pool already shut down (this module's teardown or a resize).
+SHUT_POOL_REFUSAL = "after shutdown"
+JOB_THREAD_PREFIX = "pm-worker"
+# The package teardown's bound for the job threads to finish their item.
+JOB_POOL_JOIN_SECONDS = 5
 _logger = logging.getLogger("prompt_manager.parallel")
 _noted = set()
 _noted_lock = threading.Lock()
+_pool_lock = threading.Lock()
+_pool = None  # (workers, ThreadPoolExecutor)
+_retired = []  # pools replaced by a new size, finishing what they hold
 
 
 def refused_at_shutdown(error):
@@ -45,39 +63,113 @@ def note_refusal(logger, what):
     logger.debug(f"{what}: refused, the interpreter is exiting")
 
 
-def map_parallel(func, items, workers):
-    """Apply ``func`` to every item on up to ``workers`` threads, keeping order.
+def job_pool_started():
+    """Whether a job pool exists (the teardown stops it)."""
+    return _pool is not None or bool(_retired)
 
-    ``workers`` of 1 (or fewer) runs on the calling thread with no pool. An
-    exception in ``func`` propagates to the caller, so callers that must
-    survive bad input wrap ``func`` themselves. A pool refused at interpreter
-    exit runs the items it did not start on the calling thread (#375). An item
+
+def _job_pool(workers):
+    """The shared pool of ``workers`` threads, replacing one of another size."""
+    global _pool
+    with _pool_lock:
+        if _pool is not None and _pool[0] == workers:
+            return _pool[1]
+        retired = _pool
+        _pool = (
+            workers,
+            ThreadPoolExecutor(workers, thread_name_prefix=JOB_THREAD_PREFIX),
+        )
+        current = _pool[1]
+    if retired is not None:
+        retired[1].shutdown(wait=False)  # it still runs what it holds
+        with _pool_lock:
+            _retired[:] = [p for p in _retired if _alive(p)] + [retired[1]]
+    return current
+
+
+def shutdown_job_pool(timeout=None):
+    """Stop the job pool (and any pool a resize retired): queued items are
+    cancelled (their callers run them on their own thread), running ones are
+    joined up to ``timeout`` seconds. True when no job thread is left."""
+    global _pool
+    with _pool_lock:
+        current, _pool = _pool, None
+        executors = _retired[:] + ([] if current is None else [current[1]])
+        _retired.clear()
+    threads = []
+    for executor in executors:
+        # After shutdown: it takes the lock ``submit`` holds while it starts
+        # and records a thread, so every thread is listed and none starts.
+        executor.shutdown(wait=False, cancel_futures=True)
+        threads.extend(getattr(executor, "_threads", ()))
+    deadline = None if timeout is None else time.monotonic() + timeout
+    for thread in threads:
+        thread.join(None if deadline is None else max(0.0, deadline - time.monotonic()))
+    return not any(thread.is_alive() for thread in threads)
+
+
+def _alive(executor):
+    return any(t.is_alive() for t in getattr(executor, "_threads", ()))
+
+
+def _forget(executor):
+    """Drop *executor* as the shared pool (a later call starts a new one)."""
+    global _pool
+    with _pool_lock:
+        if _pool is not None and _pool[1] is executor:
+            _pool = None
+
+
+def _on_job_thread():
+    return threading.current_thread().name.startswith(JOB_THREAD_PREFIX + "_")
+
+
+def map_parallel(func, items, workers):
+    """Apply ``func`` to every item on the shared pool of ``workers``
+    threads, keeping order.
+
+    ``workers`` below 1 counts as 1. A call made from a job thread runs on
+    that thread, so nested work never waits for its own pool. An exception
+    in ``func`` propagates to the caller once the call's other items have
+    stopped (those not started are dropped), so callers that must survive
+    bad input wrap ``func`` themselves. A pool refused at interpreter exit
+    runs the items it did not start on the calling thread (#375). An item
     that has not started when the interpreter starts exiting raises
     ExitRequested instead (#376).
     """
     func = checked(func)
     items = list(items)
-    workers = max(1, int(workers or 1))
-    if workers == 1 or len(items) <= 1:
+    if not items or _on_job_thread():
         return [func(item) for item in items]
-    max_workers = min(workers, len(items))
-    with ThreadPoolExecutor(max_workers, thread_name_prefix="pm-worker") as pool:
-        futures = []
-        try:
-            for item in items:
-                futures.append(pool.submit(func, item))
-        except RuntimeError as error:
-            if not refused_at_shutdown(error):
-                raise
+    pool = _job_pool(max(1, int(workers or 1)))
+    futures = []
+    try:
+        for item in items:
+            futures.append(pool.submit(func, item))
+    except RuntimeError as error:
+        if refused_at_shutdown(error):
             note_refusal(_logger, "A worker pool")
             # Drop what was queued but not started; it runs here instead.
             # (A refused Thread.start leaves its item queued without a future
             # here: it is cancelled too, unless a busy pool thread took it in
             # between, the one case where an item can run twice.)
-            pool.shutdown(wait=True, cancel_futures=True)
-        else:
-            return [future.result() for future in futures]
-    return _run_rest(func, items, futures)
+            _forget(pool)
+            pool.shutdown(wait=False, cancel_futures=True)
+        elif SHUT_POOL_REFUSAL not in str(error):
+            raise
+    return _collect(func, items, futures)
+
+
+def _collect(func, items, futures):
+    """``_run_rest``; on an error, first drop the call's items not started
+    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)
+        raise
 
 
 def _run_rest(func, items, futures):

web/js/workspace/prompt-sender.mjs

@@ -11,6 +11,13 @@ const linkedTo = (target,roles) => Object.keys(roles).filter(role=>Array.isArray
 const nodeLabel = target => `#${target.id} ${target.label}`;
 // A lost reply leaves the outcome unknown; anything else the host said is a refusal, and nothing was written.
 const LOST_REPLY = /timed out|aborted|disposed/i;
+const outcomeUnknown = (posted,error) => posted && (LOST_REPLY.test(error.message) || /unexpected acknowledgement/.test(error.message));
+const UNKNOWN_ADVICE = 'Check ComfyUI, then preview again.';
+/** The Destination node value that sends to every eligible node of the window (#542, D191). Node ids never take it. */
+export const ALL_NODES = '*';
+// "All" sends in node id order: 2 before 10, and a subgraph id "3:1" after "3".
+const byNodeId = (a,b) => a.id.localeCompare(b.id,'en',{numeric:true});
+const allMessage = (count,where) => `Sending to all ${count} eligible nodes in ${where}.`;
 
 /** Why nothing in the discovered windows can receive these roles, most specific first (D173). */
 function unavailableMessage(found,roles) {
@@ -40,23 +47,90 @@ function distinctLabels(items,suffix) {
 
 export function createSendSession({client, roles:rawRoles, onChange}) {
     const roles = Object.freeze(validateRoles(rawRoles));
-    let state = {phase:'idle',hosts:[],hostId:'',targetId:'',preview:null,outcomeUnknown:false,message:''};
+    // `previews` holds the reviewed nodes of "All", in send order; `preview` the one node otherwise.
+    let state = {phase:'idle',hosts:[],hostId:'',targetId:'',preview:null,previews:[],outcomeUnknown:false,message:''};
     let operation, generation = 0, disposed = false;
     const emit = patch => { if (!disposed) { state = Object.freeze({...state,...patch}); onChange(state); } };
     function invalidate() { generation++; operation?.abort(); operation = null; }
-    function begin(phase) { invalidate(); operation = new AbortController(); emit({phase,preview:null,outcomeUnknown:false,message:''}); return {signal:operation.signal,version:generation}; }
+    function begin(phase) { invalidate(); operation = new AbortController(); emit({phase,preview:null,previews:[],outcomeUnknown:false,message:''}); return {signal:operation.signal,version:generation}; }
     const current = request => !disposed && generation === request.version;
-    const selected = () => state.hosts.find(host=>host.hostId===state.hostId)?.targets.find(target=>target.id===state.targetId);
-    async function fetchPreview(request) {
-        const preview=await client.preview(state.hostId,state.targetId,roles,{signal:request.signal});
+    const chosenHost = () => state.hosts.find(host=>host.hostId===state.hostId);
+    /** The nodes the current choice sends to: one, every eligible node of the window for "All", or none. */
+    function chosen() {
+        const host=chosenHost();
+        if (!host) return [];
+        if (state.targetId===ALL_NODES) return host.targets.length>1 ? host.targets.toSorted(byNodeId) : [];
+        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) {
+        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!==state.targetId || !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.');
+        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=selected()?.revision;
-        if(found!==undefined && preview.revision!==found) throw Error('The workflow changed since this window was found. Nothing was sent. Find windows again.');
+        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.');
         return preview;
     }
+    function checkAcknowledgement(acknowledgement,targetId) {
+        if(acknowledgement?.targetId!==targetId || !acknowledgement.values || Object.keys(acknowledgement.values).length!==Object.keys(roles).length || Object.keys(roles).some(role=>!Object.hasOwn(acknowledgement.values,role) || acknowledgement.values[role]!==roles[role])) throw Error('ComfyUI returned an unexpected acknowledgement.');
+    }
+    const reviewedText = change => JSON.stringify([change.fields.map(field=>[field.role,field.before]).sort(),change.composition ?? null]);
+    const sameChanges = (reviewed,fresh) => reviewedText(reviewed)===reviewedText(fresh);
+    /** Q542-2 C: every node of "All" previews before anything is written; the first refusal names its node. */
+    async function previewEach(request,hostId,targets) {
+        const reviewed=[];
+        for (const target of targets) {
+            try {
+                const preview=await fetchPreview(request,hostId,target);
+                if (!preview) return null;
+                reviewed.push({target,label:target.label,preview});
+            } catch(error) {
+                if (!current(request)) return null;
+                throw Error(`${target.label}: ${error.message}${/Nothing was sent/.test(error.message) ? '' : ' Nothing was sent.'}`);
+            }
+        }
+        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.
+     */
+    async function applyEach(request,hostId,reviewed,where,held) {
+        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;
+                }
+                posted=written=true;
+                const acknowledgement=await client.apply(hostId,ticket,{signal:request.signal});
+                if (!current(request)) return;
+                checkAcknowledgement(acknowledgement,target.id);
+                sent.push(target);
+            } catch(error) {
+                if (!current(request)) return;
+                const summary=[`Sent to ${sent.length} of ${reviewed.length} nodes in ${where}.`,...refused];
+                if (outcomeUnknown(posted,error)) {
+                    const unsent=reviewed.slice(index+1).map(item=>item.label);
+                    emit({phase:'error',outcomeUnknown:true,message:[...summary,`No confirmation received from ${target.label}; the update may have applied.`,...(unsent.length ? [`Not sent: ${unsent.join(', ')}.`] : []),UNKNOWN_ADVICE,error.message].join(' ')});
+                    return;
+                }
+                refused.push(`${target.label}: ${error.message}`);
+            }
+        }
+        const message=[`Sent to ${sent.length} of ${reviewed.length} nodes in ${where}.`,...refused].join(' ');
+        emit(refused.length ? {phase:'error',message} : {phase:'success',message});
+    }
     function hostChoice(host) {
         if (host.targets.length>1) return {hostId:host.hostId,targetId:'',message:`Choose a destination node in ${host.label}.`};
         const [target] = host.targets;
@@ -79,38 +153,61 @@ export function createSendSession({client, roles:rawRoles, onChange}) {
                 else { const choice=hostChoice(hosts[0]); emit({phase:'select',hosts,...choice,message:choice.message+note}); }
             } catch (error) { if (current(request)) emit({phase:'error',message:error.message}); }
         },
-        selectHost(hostId) { if (disposed) return; invalidate(); const host=state.hosts.find(host=>host.hostId===hostId); emit({phase:'select',preview:null,outcomeUnknown:false,...(host ? hostChoice(host) : {hostId:'',targetId:'',message:'Choose a ComfyUI window.'})}); },
-        selectTarget(targetId) { if (disposed) return; invalidate(); const host=state.hosts.find(host=>host.hostId===state.hostId); const chosen=host?.targets.some(target=>target.id===targetId); emit({phase:'select',outcomeUnknown:false,targetId:chosen?targetId:'',preview:null,message:chosen?'Apply to send it, or preview the changes first.':'Choose a destination node, then preview changes.'}); },
+        selectHost(hostId) { if (disposed) return; invalidate(); const host=state.hosts.find(host=>host.hostId===hostId); emit({phase:'select',preview:null,previews:[],outcomeUnknown:false,...(host ? hostChoice(host) : {hostId:'',targetId:'',message:'Choose a ComfyUI window.'})}); },
+        selectTarget(targetId) {
+            if (disposed) return; invalidate(); const host=chosenHost();
+            // D173: "All" is a choice only where there are two or more nodes, and it is never made for you.
+            const all=targetId===ALL_NODES && host?.targets.length>1;
+            const chosenOne=all || Boolean(host?.targets.some(target=>target.id===targetId));
+            const message=all ? `${allMessage(host.targets.length,host.label)} Apply to send it, or preview the changes first.` : chosenOne ? 'Apply to send it, or preview the changes first.' : 'Choose a destination node, then preview changes.';
+            emit({phase:'select',outcomeUnknown:false,targetId:chosenOne?targetId:'',preview:null,previews:[],message});
+        },
         async preview() {
-            if (disposed || !selected() || ['previewing','applying'].includes(state.phase)) return;
+            const targets=chosen();
+            if (disposed || !targets.length || ['previewing','applying'].includes(state.phase)) return;
+            const all=state.targetId===ALL_NODES; const hostId=state.hostId; const windowLabel=chosenHost()?.label;
             const request=begin('previewing');
             try {
-                const preview=await fetchPreview(request);
+                if (all) {
+                    const previews=await previewEach(request,hostId,targets);
+                    if (previews) emit({phase:'ready',previews,message:`${allMessage(targets.length,windowLabel)} Review the complete changes, then apply.`});
+                    return;
+                }
+                const preview=await fetchPreview(request,hostId,targets[0]);
                 if (!preview) return;
-                const windowLabel=state.hosts.find(host=>host.hostId===state.hostId)?.label;
                 emit({phase:'ready',preview,message:`Sending to ${nodeLabel({id:preview.target.id,label:preview.target.label ?? ''})} in ${windowLabel}. Review the complete changes, then apply.`});
-            } catch(error) { if(current(request)) emit({phase:'error',preview:null,message:error.message}); }
+            } catch(error) { if(current(request)) emit({phase:'error',preview:null,previews:[],message:error.message}); }
         },
         async apply() {
             // Success stays closed: the second click of a double click must not send again.
-            if(disposed || !selected() || !['select','ready','error'].includes(state.phase) || state.outcomeUnknown) return;
-            const held=state.phase==='ready' ? state.preview : null;
-            const targetId=state.targetId; const hostId=state.hostId;
+            const targets=chosen();
+            if(disposed || !targets.length || !['select','ready','error'].includes(state.phase) || state.outcomeUnknown) return;
+            const all=state.targetId===ALL_NODES;
+            const held=state.phase==='ready' ? (all ? state.previews : state.preview) : null;
+            const hostId=state.hostId; const windowLabel=chosenHost()?.label;
             const request=begin('applying');
             emit({message:'Waiting for ComfyUI confirmation. Closing this dialog does not undo a sent update.'});
+            if (all) {
+                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));
+                return;
+            }
+            const [target]=targets;
             let posted=false;
             try {
-                const preview=held ?? await fetchPreview(request);
+                const preview=held ?? await fetchPreview(request,hostId,target);
                 if(!preview || !current(request)) return;
                 posted=true;
                 const acknowledgement=await client.apply(hostId,preview.ticket,{signal:request.signal});
                 if(!current(request)) return;
-                if(acknowledgement?.targetId!==targetId || !acknowledgement.values || Object.keys(acknowledgement.values).length!==Object.keys(roles).length || Object.keys(roles).some(role=>!Object.hasOwn(acknowledgement.values,role) || acknowledgement.values[role]!==roles[role])) throw Error('ComfyUI returned an unexpected acknowledgement.');
+                checkAcknowledgement(acknowledgement,target.id);
                 emit({phase:'success',message:'Sent to ComfyUI.'});
             } catch(error) {
                 if(!current(request)) return;
-                const unknown=posted && (LOST_REPLY.test(error.message) || /unexpected acknowledgement/.test(error.message));
-                emit({phase:'error',preview:null,outcomeUnknown:unknown,message:unknown ? `No confirmation received. The update may have applied. Check ComfyUI, then preview again. ${error.message}` : error.message});
+                const unknown=outcomeUnknown(posted,error);
+                emit({phase:'error',preview:null,outcomeUnknown:unknown,message:unknown ? `No confirmation received. The update may have applied. ${UNKNOWN_ADVICE} ${error.message}` : error.message});
             }
         },
         dispose() { if(disposed) return; disposed=true; invalidate(); client.dispose(); },
@@ -159,7 +256,7 @@ export function createPromptSender({scope,baseUrl=window.ApiClient.resolveUrl('/
         inspect.disabled=true;apply.disabled=true;
         const actions=element('footer',undefined,'prompt-sender-actions');actions.append(find,inspect,apply);
         dialog.append(header,description,controls,status,preview,actions);document.body.append(dialog);
-        let session,closed=false;
+        let session,closed=false,headingIds=0;
         function dispose() {
             if(closed) return;closed=true;local.dispose();session?.dispose();
             if(dialog.open) dialog.close();dialog.remove();
@@ -176,8 +273,8 @@ export function createPromptSender({scope,baseUrl=window.ApiClient.resolveUrl('/
         function options(select,placeholder,entries,value) {
             select.replaceChildren(...[["",placeholder],...entries].map(([value,label])=>{const option=element('option',label);option.value=value;return option;}));select.value=value;
         }
-        function textBlock(parent,label,value) {
-            parent.append(element('h4',label));const text=element('pre');text.tabIndex=0;text.append(element('code',value === '' ? '(empty)' : value));parent.append(text);
+        function textBlock(parent,label,value,level=4) {
+            parent.append(element(`h${Math.min(level,6)}`,label));const text=element('pre');text.tabIndex=0;text.append(element('code',value === '' ? '(empty)' : value));parent.append(text);
         }
         function render(state) {
             if(closed) return;
@@ -187,28 +284,43 @@ export function createPromptSender({scope,baseUrl=window.ApiClient.resolveUrl('/
             dialog.setAttribute('aria-busy',String(busy));
             options(host,'Choose a ComfyUI window',state.hosts.map(item=>[item.hostId,item.label]),state.hostId);
             const selected=state.hosts.find(item=>item.hostId===state.hostId);
-            options(target,'Choose a destination node',(selected?.targets || []).map(item=>[item.id,item.label]),state.targetId);
+            const nodes=(selected?.targets || []).map(item=>[item.id,item.label]);
+            // Q542-4 A: the last option sends to every eligible node of this window.
+            if(nodes.length>1) nodes.push([ALL_NODES,`All ${nodes.length} eligible nodes`]);
+            options(target,'Choose a destination node',nodes,state.targetId);
             host.disabled=busy;target.disabled=busy || !selected;
             find.disabled=busy;inspect.disabled=busy || !state.targetId;apply.disabled=busy || !state.targetId || state.phase==='success' || state.outcomeUnknown;
             preview.replaceChildren();
             if(state.preview) {
                 preview.append(element('h3','Review changes'));
-                for(const field of state.preview.fields || []) {
-                    const section=element('section',undefined,`prompt-sender-field prompt-role-${field.role}`);
-                    section.append(element('h3',field.role==='positive'?'Positive prompt':'Negative prompt'));
-                    textBlock(section,'Before',field.before);textBlock(section,'After',field.after);preview.append(section);
-                }
-                const composition=state.preview.composition;
-                if(composition) {
-                    const section=element('section',undefined,'prompt-sender-composition');section.append(element('h3','Positive composition'));
-                    textBlock(section,'Prepend (preserved)',composition.prepend || '');textBlock(section,'Append (preserved)',composition.append || '');
-                    textBlock(section,'Combined before',composition.before ?? 'Resolved at execution from connected inputs.');
-                    textBlock(section,'Combined after',composition.after ?? 'Resolved at execution from connected inputs.');
-                    if(composition.prependConnected || composition.appendConnected) section.append(element('p','Connected prepend and append inputs are preserved and resolved when the workflow runs.'));
-                    preview.append(section);
+                changes(preview,state.preview,3);
+            }
+            if(state.previews?.length) {
+                preview.append(element('h3','Review changes'));
+                // One section per node of "All", in the order they will be sent, named by its heading.
+                for(const item of state.previews) {
+                    const node=element('section',undefined,'prompt-sender-node');const heading=element('h4',item.label);
+                    heading.id=`prompt-sender-node-${++headingIds}`;node.setAttribute('aria-labelledby',heading.id);
+                    node.append(heading);changes(node,item.preview,5);preview.append(node);
                 }
             }
         }
+        function changes(parent,change,level) {
+            for(const field of change.fields || []) {
+                const section=element('section',undefined,`prompt-sender-field prompt-role-${field.role}`);
+                section.append(element(`h${level}`,field.role==='positive'?'Positive prompt':'Negative prompt'));
+                textBlock(section,'Before',field.before,level+1);textBlock(section,'After',field.after,level+1);parent.append(section);
+            }
+            const composition=change.composition;
+            if(composition) {
+                const section=element('section',undefined,'prompt-sender-composition');section.append(element(`h${level}`,'Positive composition'));
+                textBlock(section,'Prepend (preserved)',composition.prepend || '',level+1);textBlock(section,'Append (preserved)',composition.append || '',level+1);
+                textBlock(section,'Combined before',composition.before ?? 'Resolved at execution from connected inputs.',level+1);
+                textBlock(section,'Combined after',composition.after ?? 'Resolved at execution from connected inputs.',level+1);
+                if(composition.prependConnected || composition.appendConnected) section.append(element('p','Connected prepend and append inputs are preserved and resolved when the workflow runs.'));
+                parent.append(section);
+            }
+        }
         dialog.showModal();
         try {
             const url=new URL(baseUrl,window.location.href);

web/js/workspace/system-backups.mjs

@@ -179,13 +179,27 @@ export function reviewRefusal(error) {
     return REVIEW_FAILED;
 }
 
+/**
+ * #583: the review uploads not answered yet. WebKit can read a FormData's file part after the fetch call returned (it
+ * did whenever a request interceptor, here Playwright's, was armed); a garbage collection in between dropped a form
+ * nothing referenced any more, and the server got an empty file. Each form is held here until its upload is answered.
+ * Exported for the unit tests only.
+ */
+export const uploadsInFlight = new Set();
+
 /** Stage `file` for review; the server's answer, or an ApiError naming its refusal. */
 export async function uploadRestore(file, { fetchUrl = window.ApiClient.fetchUrl, ApiError = window.ApiClient.ApiError,
     signal } = {}) {
     const form = new FormData();
     form.append('database_file', file, file.name);
-    const response = await fetchUrl(PREVIEW_URL, { method: 'POST', body: form, cache: 'no-store', credentials: 'same-origin',
-        signal });
+    uploadsInFlight.add(form);
+    let response;
+    try {
+        response = await fetchUrl(PREVIEW_URL, { method: 'POST', body: form, cache: 'no-store', credentials: 'same-origin',
+            signal });
+    } finally {
+        uploadsInFlight.delete(form);
+    }
     let data = null;
     try {
         data = await response.json();

web/js/workspace/workspace.css

@@ -416,6 +416,9 @@ pre.prompt-text { font: inherit; font-size: 12px; line-height: 1.75; margin-bloc
 .prompt-sender-preview { margin-block: 20px; min-inline-size: 0; /* D48 */ }
 .prompt-sender-field { border: 1px solid var(--ca-line); border-radius: 7px; padding: 14px; margin-block: 12px; min-inline-size: 0; /* D48 */ }
 .prompt-sender-field :where(pre, textarea), .prompt-sender-composition { white-space: pre-wrap; overflow-wrap: anywhere; font: inherit; font-size: 12px; line-height: 1.65; max-inline-size: 100%; /* D48 */ }
+/* D191 (#542): each node of "All eligible nodes" reviews under its own heading. */
+.prompt-sender-node { border-block-start: 1px solid var(--ca-line); padding-block-start: 14px; margin-block: 18px; min-inline-size: 0; /* D191 */ }
+.prompt-sender-node > h4 { font-size: 15px; /* D191 */ }
 .prompt-sender-field textarea { inline-size: 100%; min-block-size: 100px; resize: vertical; /* D48 */ }
 .prompt-sender-composition { padding: 12px; background: var(--ca-field); border-radius: 6px; /* D48 */ }
 .prompt-sender-actions { justify-content: flex-end; margin-block-start: 18px; /* D48 */ }