Merge remote-tracking branch 'origin/main' into perf/523-image-memory-2

# Conflicts: # TESTING.md # py/api/__init__.py # py/api/admin.py
perf(images): read PNG text without decoding pixels (#523 ) (#577 )
2026-10-31 17:46:05 -07:00 · 2026-11-01 00:04:27 +00:00 · 2026-10-31 23:17:17 +00:00 · 2026-10-31 16:09:18 -07:00 · 2026-10-31 15:58:25 -07:00 · 2026-10-31 15:36:07 -07:00
43 changed files with 2504 additions and 288 deletions
@@ -1,9 +1,9 @@
-name: Windows and macOS
+name: Windows

 permissions:
  contents: read

-# Non-blocking portability runs (#420): never on pull requests or in the merge queue, and no context
+# Non-blocking Windows runs (#420; macOS dropped from CI in #571): never on pull requests or in the merge queue, and no context
 # here is a required check. A red run is fixed at its cause before the workspace cutover (#334).
 on:
  push:
@@ -25,10 +25,7 @@ env:
 # run counts (tests/vendor/ci-workflow.test.mjs recomputes it, #493).
 # Measured 2026-10-29 on main runs 40935337099 (dd247e3), 40938407252 (dd247e3), 40941240624 (dd247e3):
 #   slowest leg: 7m24s (py3.12 windows) / 7m31s (py3.13 windows) / 7m32s (py3.14 windows) -> 15 min
-#     (windows 4m56s-7m32s, macos 4m39s-6m14s)
-#   native-trash (#499) has no green main runs yet: its probe, run 40949370943 (2026-10-29), took
-#     about 1 minute per macOS leg, so ceil(1.5 x 2) + 3 gives it 6 minutes (step 4). Recalibrate
-#     after three green runs.
+#     (windows 4m56s-7m32s)
 #   native-trash-linux (#543) has no green main runs yet: it installs the dev requirements and runs
 #     one test module, about 1 minute by the same estimate, so ceil(1.5 x 2) + 3 gives it 6 minutes.
 #     Recalibrate after three green runs.
@@ -40,7 +37,7 @@ jobs:
    strategy:
      fail-fast: false
      matrix:
-        os: [windows-latest, macos-latest]
+        os: [windows-latest]
        python-version: ["3.12", "3.13", "3.14"]

    steps:
@@ -60,7 +57,7 @@ jobs:

      # The runner's TEMP is on C:, the OS disk, where a flush took about 6 ms
      # and a test database's create-to-delete about 111 ms; on the D: work
-      # disk the same took 0.5 ms and 23 ms (macOS: 19 ms). Tests write their
+      # disk the same took 0.5 ms and 23 ms. Tests write their
      # databases under TEMP, so it moves to a folder on D: (measured in #487,
      # run 40915866444). The runner's TEMP was an 8.3 short name
      # (C:\Users\RUNNER~1\...), so the new one is spelled by its short name too
@@ -83,31 +80,6 @@ jobs:
          PM_TEST_SLOWEST: "40"
        run: python -m tests.run_suite -v

-  # The real Trash on both macOS CPUs (#499 design §10.2). It runs apart from the suite, so a red
-  # suite never skips the native proof.
-  native-trash:
-    name: native trash on ${{ matrix.os }}
-    runs-on: ${{ matrix.os }}
-    timeout-minutes: 6
-    strategy:
-      fail-fast: false
-      matrix:
-        os: [macos-latest, macos-15-intel]
-
-    steps:
-      - uses: actions/checkout@v7
-
-      - name: Set up Python
-        uses: actions/setup-python@v7
-        with:
-          python-version: "3.12"
-
-      - name: Native trash proof
-        timeout-minutes: 4
-        env:
-          PM_NATIVE_TRASH: macos
-        run: python -m unittest tests.test_trash_macos_native tests.test_trash_macos -v
-
  # The same tests as the fake mount table, on a real ext4 mount used as the output root (#543). A
  # runner can mount with sudo; the tests run as the normal user, so owner and mode checks are real.
  native-trash-linux:
@@ -28,7 +28,7 @@ env:
 #   py matrix, slowest py3.10 on windows-latest: 12m43s / 15m54s / 14m53s -> 27 min
 #     (ubuntu 5m27s-6m27s, macos 6m14s-8m30s, windows 11m42s-15m54s)
 #     3.10 and 3.11 left the matrix and 3.14 joined it in #356; re-measure after three green main runs.
-#     Windows and macOS moved to portability.yml (#420), which has its own measured limit (#493);
+#     Windows moved to portability.yml (#420; macOS left CI in #571), which has its own measured limit (#493);
 #     this Linux-only matrix keeps 27 minutes until it is re-measured.
 #   real ComfyUI compatibility: 13m32s / 8m55s / 11m46s -> 24 min
 #   coverage gate (85%): 7m26s / 7m22s / 7m54s -> 15 min
@@ -39,7 +39,7 @@ env:
 # Pull requests run a fast subset (#420): Python 3.12 on ubuntu-latest, coverage, lint and frontend
 # modules. The merge queue (merge_group), pushes to main and manual runs add Python 3.13 and 3.14 on
 # Linux and the real ComfyUI job. "all tests" is the required gate that holds the queue to that set.
-# Windows and macOS run in portability.yml (push to main, nightly, on demand) and never gate a merge.
+# Windows runs in portability.yml (push to main, nightly, on demand) and never gate a merge.
 jobs:
  test:
    name: py${{ matrix.python-version }} on ${{ matrix.os }}
@@ -167,6 +167,8 @@ Searches your library by text, category, tags and minimum rating, then outputs t

 ![Prompt Search List node producing three images](assets/readme/node-search.webp)

+To get the images of a Search List batch linked to their prompts, wire the Search List output into a Prompt Manager Text node, not straight into CLIP Text Encode. Wired straight in, the images stay unlinked: the search text is never saved as a prompt.
+
 ## Install

 **Requirements:** ComfyUI and Python 3.12 or newer. `watchdog` and `Pillow` are the only dependencies of their own; the rest comes with ComfyUI.
@@ -236,12 +238,12 @@ The documentation for those pages is in [README-legacy.md](README-legacy.md).

 | Layer | What runs |
 | --- | --- |
-| Python | Over 200 test files on Linux, Windows and macOS with Python 3.12 to 3.14, with an 85% coverage gate |
+| Python | Over 200 test files on Linux and Windows with Python 3.12 to 3.14, with an 85% coverage gate |
 | Frontend | About 100 unit test files, with coverage floors per file |
 | Browser | Over 60 Playwright spec files in Chromium, Firefox, WebKit and touch emulation |
 | Real ComfyUI | A pinned CPU ComfyUI, started in Classic and Nodes 2.0, running real workflows |

-Linux runs on every pull request and in the merge queue. Windows and macOS run on every push to main and nightly. To run the tests yourself, see [TESTING.md](TESTING.md).
+Linux runs on every pull request and in the merge queue. Windows runs on every push to main and nightly. macOS is not tested; it should work, but support is best-effort. To run the tests yourself, see [TESTING.md](TESTING.md).

 ## License

@@ -261,7 +261,7 @@ the process exits non-zero.
  job fails anyway and the dump is what matters.
 - `PM_TEST_SLOWEST=N` (off by default) lists the N slowest tests after the
  summary, each timed from its `start:` line to the next one like the hang
-  limit. The Windows and macOS jobs (`portability.yml`) set it to 40, so every
+  limit. The Windows jobs (`portability.yml`) set it to 40, so every
  run shows where its time went (#487). The list prints only when the run
  ends: a job GitHub cancels or a watchdog exit prints none, so read the
  timestamps between `start:` lines instead.
@@ -327,7 +327,7 @@ Python 3.13/3.14, real ComfyUI and Firefox/WebKit failures first show up in the
 merge queue, on the merge-group commit rather than the
 pushed one. Find that run with `gh run list --event merge_group --workflow test.yml`
 (or `viewerjs-check.yml`) and match its head branch
-(`gh-readonly-queue/main/pr-<N>-...`). Windows and macOS failures show up only
+(`gh-readonly-queue/main/pr-<N>-...`). Windows failures show up only
 after the merge, in `gh run list --workflow portability.yml`.

 Start fixing as soon as a failed log shows the cause, but push the fix only after
@@ -345,11 +345,16 @@ and the time left. The steps, in order:
   monitor's callbacks (Needs review recorder, preview queue, Images listing
   listener).
 2. `ImageMonitor.stop_monitoring` works against one 5 s deadline. Closing the
-   image handler joins the image worker; watchdog's `observer.stop()` (not
-   bounded here) and the observer join get what is left, which can be 0 s.
-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).
+   image handler joins the image worker; watchdog's `observer.stop()` and the
+   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),
+   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:
@@ -361,6 +366,15 @@ stored, with the parts not reached False, and the `atexit` callback is
 removed: the later steps are skipped, no summary line is logged, and a later
 call returns that result and runs nothing.

+The observer is started from a short-lived daemon thread, so every thread it
+creates is a daemon (a thread inherits that flag from the thread that creates
+it). watchdog 7 on Windows starts a plain `threading.Thread` per watch for its
+`ReadDirectoryChangesW` reader; created from ComfyUI's main thread it would not
+be a daemon, and the interpreter waits for every such thread before it runs any
+`atexit` callback, so the teardown would never start (#561).
+`tests/test_package_exit.py` runs a child whose observer does both (a plain
+reader thread, a `stop()` that never returns) and requires its exit.
+
 ComfyUI never runs aiohttp's `on_cleanup` hooks, because it never calls
 `runner.cleanup()`. So `atexit` is the live host's path. It runs on Ctrl+C
 (`main.py` catches `KeyboardInterrupt` and returns) and on a normal exit.
@@ -445,12 +459,44 @@ A hot SQLite read that returns many rows aggregates them into one step
 (`json_group_array`) instead of stepping once per row, so it does not wait for
 the GIL once per row while ComfyUI generates (#354, `get_image_link_matches`).

+A metadata read never decodes pixels (#523). Pillow's `PngImageFile.text` loads
+every pixel to reach text chunks after IDAT; the scan (`_extract_comfyui_metadata`)
+and the image monitor (`ComfyUIMetadataExtractor.extract_metadata`) read PNG text
+through `utils/png_text.py`, which runs Pillow's own chunk handlers after the
+first IDAT and seeks over pixel data. The legacy `.text` read is the floor:
+`tests/test_png_text.py` compares both readers on every image under
+`tests/fixtures`, a curated corpus (text after IDAT, zTXt, iTXt, split IDAT, bad
+checksums, truncations, oversized text, APNG) and a seeded fuzz, and asserts that
+neither reader decodes pixels by patching `ImageFile.load`. The only reads that
+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
 requests, because shared-runner timing must not block a merge. Dispatch it on a
 branch to measure before merging.
-Shared Windows and macOS runners do not run them; the structural tests cover
+Shared Windows runners do not run them; the structural tests cover
 portability. Run it locally with
 `PROMPTMANAGER_PERF_BUDGET=1 python -m unittest tests.test_api_workspace_images.TestWorkspaceImages.test_filter_budget`
 and quote the printed p95 in the PR body.
@@ -646,10 +692,10 @@ Before changing paths, SQLite, streams, restore or cleanup:
  ordinary busy timeout; cross-thread handoffs wait on events with generous
  ceilings, never a short wall clock (#242).
 - Run native Windows CI early for these changes. Pull requests and the merge
-  queue run only Linux, so start Windows and macOS on the branch with
+  queue run only Linux, so start Windows on the branch with
  `gh workflow run portability.yml --ref <branch>`.
-  Include a Windows impact and evidence note in the PR. All six Windows/macOS
-  combinations in that run, every Linux leg and every applicable
+  Include a Windows impact and evidence note in the PR. All three Windows
+  legs in that run, every Linux leg and every applicable
  exact-head check must pass before merge. Portable `ntpath`/`PureWindowsPath`
  controls do not establish native filesystem, locking or cleanup success.

@@ -668,7 +714,7 @@ what runs:
 | `page workflows (firefox\|webkit, shard N/2)` | skipped | run |
 | `all tests`, `viewerjs` (gates) | run | run |
 | `perf budget` | skipped | push and dispatch only (not in the queue) |
-| `py3.12-3.14 on windows-latest/macos-latest` (`portability.yml`) | not run | push to `main`, nightly, `workflow_dispatch` only; never required |
+| `py3.12-3.14 on windows-latest` (`portability.yml`) | not run | push to `main`, nightly, `workflow_dispatch` only; never required |

 GitHub has one required set for pull requests and the queue. A job skipped by
 its `if:` reports Success. A dropped matrix leg never reports, so it would stay
@@ -681,8 +727,9 @@ guarantee:
 So a merge group cannot turn green unless all three Linux Python legs, real
 ComfyUI and the four browser shards passed.

-Windows and macOS (Python 3.12-3.14) are out of the blocking path for now
-(maintainer, 2026-10-28). They run in `portability.yml` on every push to `main`,
+Windows (Python 3.12-3.14) is out of the blocking path for now
+(maintainer, 2026-10-28). macOS is not tested in CI (maintainer, 2026-10-31, #571);
+its code stays and support is best-effort. They run in `portability.yml` on every push to `main`,
 nightly and on demand, and no context there is required. A red portability run
 is still a failure: fix it at its cause, not by rerunning, and have every one
 green before the workspace cutover (#334). `tests/vendor/ci-workflow.test.mjs` pins the
@@ -1407,6 +1454,8 @@ is shorter than the row it reserves.
 Each Overview card is the count of the page it names: the shell's Library, Images and Tags sidebar counts, one source,
 so a card and its badge cannot differ. `/prompt_manager/stats` is for the legacy page and counts raw rows. A visit re-reads
 the three unless a read is in flight. An unread or failed card shows a dash; Retry re-reads only the failed counts.
+System › Overview's Images metric reads the same shell count (#555), so it, the Overview card and the Images badge show
+one number; its Statistics dialog labels the legacy row counts as "Database records".
 Tests: `tests/js/workspace-state.test.js`, `tests/browser/workspace.spec.mjs`, `tests/comfyui/workspace.spec.mjs`.

 ### Images page controls and recorded deviations (#249 PR C)
@@ -2549,8 +2598,8 @@ files to the system trash. **No code path unlinks or overwrites a media file.**
  ("Moving files to the Recycle Bin isn't available on Windows: files larger than the bin's
  capacity would be deleted permanently."). Library-only removal works everywhere. Tests use `tests/fake_trash.py`; the
  browser test server installs it, so no test touches a real trash. The native proof,
-  `tests/test_trash_macos_native.py`, runs only with `PM_NATIVE_TRASH=macos` in
-  portability.yml's `native-trash` job (`macos-latest` and `macos-15-intel`): the startup
+  `tests/test_trash_macos_native.py`, runs only with `PM_NATIVE_TRASH=macos` on a Mac, by hand
+  (CI dropped its `native-trash` macOS job in #571; it last passed on `macos-latest` and `macos-15-intel`): the startup
  volume, a name clash, fresh APFS and FAT32 disk images, an image used as an output root,
  a read-only image and an unwritable folder (design and probe:
  `docs/egm/499-macos-trash/`, local).
@@ -3844,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
@@ -3878,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,
@@ -4006,7 +4055,11 @@ the per-section hand-off panel, was retired by S12 (#329): every section is nati
  "Save preferences" is `aria-disabled` (opacity .55, `not-allowed`) while clean or
  saving; a setting the last save could not save gets a red outline. CSS under D92.
 - **D93** (Overview): real values. The metrics are the Library's eligible positives
-  and active saved negatives and the indexed `generated_images` rows (Q2). "Database
+  and active saved negatives and the Images count (#555, amending Q2's indexed
+  `generated_images` rows): the shell's Images sidebar count, the Images page's own
+  total, which also lists files no row links, as the concept's `images.length` is
+  both. A visit re-reads it unless a read is in flight; "–" until read or after a
+  failed read. `images.indexed` stays in the API and is not shown. "Database
  integrity" is "Not checked" until a check ran this session, then "N issues" (checks
  that found issues, even beside a check that could not run), "Passed" or "Incomplete".
  The review row is "N ambiguous image links" (the Needs review queue's pending count,
@@ -4044,10 +4097,11 @@ the per-section hand-off panel, was retired by S12 (#329): every section is nati
 - **D96** (Statistics; amended by D153): "Usage statistics" without the sample note: the
  concept's "Runs by category" table without its Avg. rating column (Category / Runs)
  with the real runs of eligible parents per category ("Uncategorized"), then a second
-  table "Library totals" (Total / Count: Prompts, Tags, Images from the legacy
-  statistics; no Rated images). The API still reports `average_rating` and
+  table "Database records" (Total / Count: Prompt records, Tags on prompts, Image
+  records: the raw row counts of the legacy statistics, labelled as rows so they never
+  read as the Library or Images page counts, #555; no Rated images). The API still reports `average_rating` and
  `rated_images`; the workspace does not show them.
-  The Overview row says "Runs by category, with library totals." because
+  The Overview row says "Runs by category, with database record counts." because
  that is what the dialog shows (the concept's own dialog shows only "Runs by
  category"); per-prompt run counts, prompt tags and dated runs are recorded, but no
  table of them exists yet. A failed read toasts "Statistics could not
@@ -5413,7 +5467,7 @@ list stay unscoped: any row's control proves them.
 | Worker threads | `admin.html:437–445` | S4 Sources | covered | `sources` · slider · "Worker threads" | worker range max is cpu_count (workspace-system-sources.spec.mjs) |
 | LoraManager badge, enable, path, trigger words, CivitAI key, clear key | `admin.html:448–490`; `admin.js:1052–1156` | S10 Integrations | covered | `integrations` · switch · "LoraManager integration" | LoRA rows show detected / enabled / cache-loaded separately (workspace-system-integrations.spec.mjs) |
 | LoRA data import with progress | `admin.html:492–498,517–541`; `admin.js:1157` | S10 Integrations | covered | `integrations` · button · "Review import" | Review import shows replacement, blocked items and network effects with the backup box; Start opens the job (D107) (workspace-system-integrations.spec.mjs) |
-| Statistics | `admin.html:159`; `admin.js:1627` | S2 Overview | covered | `overview` · button · "Explore" | statistics dialog lists runs by category and library totals (workspace-system.spec.mjs) |
+| Statistics | `admin.html:159`; `admin.js:1627` | S2 Overview | covered | `overview` · button · "Explore" | statistics dialog lists runs by category and database record counts (workspace-system.spec.mjs) |
 | Export components JSON `/export` | `admin.html:163`; `admin.js:1607`; `py/api/prompts.py:238` | S9 Backup & export | covered | `backups` · button · "Export all prompts JSON" | Export JSON downloads /library/export; secondary buttons download all prompts (/export) and /pairs/export (D103) (workspace-system-backups.spec.mjs) |
 | Saved pairs export | `py/api/pairs.py:84` | S9 Backup & export | covered | `backups` · button · "Export saved pairs JSON" | Export JSON downloads /library/export; secondary buttons download all prompts (/export) and /pairs/export (D103) (workspace-system-backups.spec.mjs) |
 | Library export | `py/api/library.py:223` | S9 Backup & export | covered | `backups` · button · "Export JSON" | Export JSON downloads /library/export; secondary buttons download all prompts (/export) and /pairs/export (D103) (workspace-system-backups.spec.mjs) |
@@ -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():
@@ -435,12 +448,9 @@ except Exception as e:

 __all__ = ["NODE_CLASS_MAPPINGS", "NODE_DISPLAY_NAME_MAPPINGS", "WEB_DIRECTORY"]

-# Print startup message with loaded tools
-print()
-print(f"\033[94m[ComfyUI-PromptManager] Version:\033[0m {get_version()}")
-for node_key, display_name in NODE_DISPLAY_NAME_MAPPINGS.items():
-    # ASCII only: a Windows console using a legacy code page cannot print emoji,
-    # and a UnicodeEncodeError here would abort loading the whole node.
-    print(f"  \033[94mLoaded:\033[0m {display_name}")
-print(f"\033[94mTotal: {len(NODE_CLASS_MAPPINGS)} tools loaded\033[0m")
-print()
+# One startup line. ASCII only: a Windows console using a legacy code page
+# cannot print emoji, and a UnicodeEncodeError here would abort loading the node.
+print(
+    f"\033[94m[ComfyUI-PromptManager] Version:\033[0m {get_version()}"
+    f" - {len(NODE_CLASS_MAPPINGS)} tools loaded"
+)
@@ -85,6 +85,8 @@ 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:
    import sys
@@ -95,6 +97,8 @@ 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


@@ -526,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())
@@ -1048,6 +1060,8 @@ class PromptManagerAPI(
    def _extract_comfyui_metadata(self, image_path):
        """Embedded ComfyUI metadata of an image (PNG text chunks) or a video.

+        PNG text is read without decoding the pixels (``utils.png_text``, #523).
+
        Videos go through ffprobe (see utils/video_metadata). Unreadable files
        are a warning, not an error: a scan over a large output tree meets
        truncated and foreign files, and it carries on past them.
@@ -1056,7 +1070,7 @@ class PromptManagerAPI(
            return read_video_metadata(image_path)
        try:
            with Image.open(image_path) as img:
-                return dict(getattr(img, "text", None) or {})
+                return png_text(img)
        except Exception as e:
            self.logger.warning(
                f"Unreadable image {os.path.basename(str(image_path))}: {e}"
@@ -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,9 +1669,10 @@ 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:
-        Pillow decodes the whole image to reach trailing text chunks and
-        releases the GIL while doing so.
+        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.
        """
        from ..config import PromptManagerConfig

@@ -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:
@@ -2975,6 +2975,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:
@@ -3997,7 +3998,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 +4212,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 +4236,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 +4266,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
@@ -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,7 +62,7 @@ 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
@@ -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()

@@ -294,7 +299,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.
    """

@@ -491,12 +496,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
@@ -567,7 +575,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 +630,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 +663,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 +674,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 +718,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 +793,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,13 +837,13 @@ 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(
@@ -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)
@@ -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.
@@ -4,7 +4,7 @@ description = "A powerful ComfyUI custom node that extends the standard text enc
 version = "3.4.1"
 license = {file = "LICENSE"}
 requires-python = ">=3.12"
-dependencies = ["watchdog>=2.1.0,<7", "Pillow>=8.0.0"]
+dependencies = ["watchdog>=2.1.0", "Pillow>=8.0.0"]

 [build-system]
 requires = ["setuptools>=84.2.1"]
@@ -1,5 +1,5 @@
 # Runtime dependencies
-watchdog>=6.0.0,<7
+watchdog>=6.0.0
 Pillow>=12.3.0
 aiohttp>=3.14.5

@@ -1,4 +1,4 @@
 # Runtime dependencies for PromptManager.
 # aiohttp, numpy and torch come from ComfyUI itself and are not listed here.
-watchdog>=2.1.0,<7            # Output directory monitoring
+watchdog>=2.1.0               # Output directory monitoring
 Pillow>=8.0.0                # Image metadata extraction and thumbnails
@@ -162,13 +162,15 @@ async function serverJson(request, url) {
 }

 const number = value => value.toLocaleString('en-US');
+// The sidebar's Images count (state.mjs), which System's Images metric shows (#555).
+const IMAGES_COUNT_URL = '/prompt_manager/images/output?mode=workspace&limit=1';

 /** The concept's shell and Overview with the recorded deviations applied, given the server's real overview. */
-function mapOverviewConcept(reference, overview) {
+function mapOverviewConcept(reference, overview, images) {
    // D165: the discreet legacy-pages line follows the settings layout.
    reference = { ...reference, tree: [...reference.tree, 'p.legacy-pages.small'] };
    const relabel = { 'Database integrity': 'Not checked', 'Positive prompts': number(overview.library.positive_prompts),
-        'Negative variants': number(overview.library.negative_variants), Images: number(overview.images.indexed) };
+        'Negative variants': number(overview.library.negative_variants), Images: number(images) };
    const rows = {
        // D93: the real pending count; Review metadata opens Needs review (the concept's go-review).
        '3 ambiguous image links': row => ({ ...row,
@@ -179,8 +181,9 @@ function mapOverviewConcept(reference, overview) {
        'One display configuration': row => ({ ...row, description: 'Grid and density settings are shared across views.' }),
        // D93: the real sources, not samples.
        '2 sample output sources': row => ({ ...row, title: `${number(overview.sources.effective)} output sources` }),
-        // D96: the dialog shows runs by category and library totals, not prompt, tag or date usage (D153: no ratings).
-        Statistics: row => ({ ...row, description: 'Runs by category, with library totals.' }),
+        // D96: the dialog shows runs by category and database record counts, not prompt, tag or date usage (D153: no
+        // ratings; #555: the counts are labelled as records, not as the Library or Images page counts).
+        Statistics: row => ({ ...row, description: 'Runs by category, with database record counts.' }),
    };
    const surfaces = reference.columns.surfaces.map(surface => ({ ...surface,
        rows: surface.rows.map(row => (rows[row.title] ?? (value => value))(row)) }));
@@ -218,6 +221,8 @@ async function conceptOverview(page, theme) {
 async function productionOverview(page, theme) {
    await openProduction(page, 'system/overview', theme);
    await expect(page.locator('#workspace-main .summary-metrics')).toBeVisible();
+    // #555: the Images metric is the shell's Images count: compare once it was read.
+    await expect(page.locator('#workspace-main .summary-metrics > .metric:nth-child(3) strong')).toHaveText(/^[\d,]+$/);
    // The thumbnail row is its own read (I2): compare once it answered.
    await expect(page.locator('#workspace-main .operation h3').filter({ hasText: 'thumbnails' })).toHaveText('No missing thumbnails');
    // So is the backup row (S9).
@@ -231,6 +236,7 @@ for (const theme of THEMES) for (const [width, height] of SIZES) {
    test(`system shell and overview match the concept ${theme} ${width}x${height}`, async ({ page, request }, testInfo) => {
        await page.setViewportSize({ width, height });
        const overview = await serverJson(request, '/prompt_manager/system/overview');
+        const images = (await serverJson(request, IMAGES_COUNT_URL)).total;
        // The fixture's thumbnails all exist, so the concept's "No missing thumbnails" is also the real title.
        expect((await serverJson(request, '/prompt_manager/images/thumbnails/status')).pending).toBe(0);
        const concept = await conceptOverview(page, theme);
@@ -239,7 +245,7 @@ for (const theme of THEMES) for (const [width, height] of SIZES) {
        await page.screenshot({ path: testInfo.outputPath(`concept-system-${theme}-${width}.png`), fullPage: true });
        const production = await productionOverview(page, theme);
        await page.screenshot({ path: testInfo.outputPath(`production-system-${theme}-${width}.png`), fullPage: true });
-        expect(diffInventory(mapOverviewConcept(concept, overview), production)).toEqual([]);
+        expect(diffInventory(mapOverviewConcept(concept, overview, images), production)).toEqual([]);
        expect(production.pageScroll).toBe(false);
    });
 }
@@ -265,8 +271,9 @@ const UNMAPPED_PATHS = [
 test('system comparison detects mutations', async ({ page, request }) => {
    await page.setViewportSize({ width: 1600, height: 1100 });
    const overview = await serverJson(request, '/prompt_manager/system/overview');
+    const images = (await serverJson(request, IMAGES_COUNT_URL)).total;
    const concept = await conceptOverview(page, 'dark');
-    const expected = mapOverviewConcept(concept, overview);
+    const expected = mapOverviewConcept(concept, overview, images);
    const production = await productionOverview(page, 'dark');
    expect(diffInventory(expected, production)).toEqual([]);
    // Unmapped, the concept differs exactly where a deviation is recorded, and nowhere else.
@@ -290,7 +297,7 @@ test('system comparison detects mutations', async ({ page, request }) => {
 });

 /**
- * D96: the concept's table, its rows the real ones; then "Library totals" in the same table markup. D153 (#447): the
+ * D96: the concept's table, its rows the real ones; then "Database records" (#555) in the same table markup. D153 (#447): the
 * concept's "Avg. rating" column and a "Rated images" total are not drawn.
 */
 function mapStatisticsConcept(reference, statistics) {
@@ -302,9 +309,10 @@ function mapStatisticsConcept(reference, statistics) {
        tables: [
            { ...runs, head: runs.head.filter(([, , text]) => text !== 'Avg. rating'),
                rows: statistics.by_category.map(row => [row.category ?? 'Uncategorized', number(row.runs)]) },
-            { caption: 'Library totals', head: [['th', 'col', 'Total'], ['th', 'col', 'Count']],
-                rows: [['Prompts', number(statistics.totals.total_prompts)], ['Tags', number(statistics.totals.total_tags)],
-                    ['Images', number(statistics.totals.total_images)]] },
+            { caption: 'Database records', head: [['th', 'col', 'Total'], ['th', 'col', 'Count']],
+                rows: [['Prompt records', number(statistics.totals.total_prompts)],
+                    ['Tags on prompts', number(statistics.totals.total_tags)],
+                    ['Image records', number(statistics.totals.total_images)]] },
        ],
    };
 }
@@ -420,7 +428,7 @@ for (const theme of THEMES) test(`statistics and health dialogs match the concep
    expect(diffInventory(mapHealthConcept(concept.health), await page.evaluate(measureDialog)).map(item => item.path))
        .toEqual(['operations.rows.0.title']);
    await page.evaluate(() => { document.querySelector('#modal .operation h3').textContent = 'Integrity check'; });
-    // And a removed "Library totals" table and a changed caption in the statistics dialog.
+    // And a removed "Database records" table and a changed caption in the statistics dialog.
    await page.keyboard.press('Escape');
    await page.locator('#workspace-main .section-nav').getByRole('button', { name: 'Overview', exact: true }).click();
    await page.locator('#workspace-main').getByRole('button', { name: 'Explore', exact: true }).click();
@@ -8,6 +8,8 @@ import { readCoverageTable, parseControl, CONTROL_EXEMPTIONS } from '../../tools
 const workspace = '/prompt_manager/workspace';
 const OVERVIEW_URL = '/prompt_manager/system/overview';
 const THUMBNAILS_URL = '/prompt_manager/images/thumbnails/status';
+// The sidebar's Images count (state.mjs NAV_COUNT_SOURCES): the Images page's own total.
+const IMAGES_COUNT_URL = '/prompt_manager/images/output?mode=workspace&limit=1';
 const number = value => value.toLocaleString('en-US');

 test.beforeEach(async ({ page, request, browserName }) => {
@@ -55,6 +57,7 @@ test('overview shows real counts and Not checked', async ({ page, request }) =>
    page.on('request', item => { if (!['GET', 'HEAD', 'OPTIONS'].includes(item.method())) writes.push(item.url()); });
    const overview = await serverJson(request, OVERVIEW_URL);
    const thumbnails = await serverJson(request, THUMBNAILS_URL);
+    const images = (await serverJson(request, IMAGES_COUNT_URL)).total;
    // The concept fixture: six positives, their saved negatives and 24 linked images, all with thumbnails.
    expect([overview.library.positive_prompts, overview.images.indexed, thumbnails.pending]).toEqual([6, 24, 0]);
    // I2: the Overview never walks the sources for thumbnails; the row reads the status route itself.
@@ -66,7 +69,8 @@ test('overview shows real counts and Not checked', async ({ page, request }) =>
        [item.querySelector('strong').textContent, item.querySelector('span').textContent]))).toEqual([
        [number(overview.library.positive_prompts), 'Positive prompts'],
        [number(overview.library.negative_variants), 'Negative variants'],
-        [number(overview.images.indexed), 'Images'],
+        // #555: the Images page's own count, the same number as its sidebar badge.
+        [number(images), 'Images'],
        ['Not checked', 'Database integrity'],
    ]);
    expect(await rowsOf(layout, 'Library health')).toEqual([
@@ -79,11 +83,31 @@ test('overview shows real counts and Not checked', async ({ page, request }) =>
    expect(await rowsOf(layout, 'Workspace preferences')).toEqual([
        ['One display configuration', 'Grid and density settings are shared across views.', 'Appearance'],
        ['2 output sources', 'Monitoring and enrichment share one source list.', 'Manage sources'],
-        ['Statistics', 'Runs by category, with library totals.', 'Explore'],
+        ['Statistics', 'Runs by category, with database record counts.', 'Explore'],
    ]);
    expect(writes).toEqual([]);
 });

+test('#555: the Images metric is the sidebar Images count, not the indexed image rows', async ({ page, request }) => {
+    const overview = await serverJson(request, OVERVIEW_URL);
+    // The Images page lists files no row links (the demo library: 122 files, 110 rows); the listing says so.
+    const listed = overview.images.indexed + 12;
+    await page.route(`**${IMAGES_COUNT_URL}`, async route => {
+        const response = await route.fetch();
+        await route.fulfill({ response, json: { ...(await response.json()), total: listed } });
+    });
+    const layout = await openOverview(page);
+    const imagesMetric = layout.locator('.summary-metrics > .metric').filter({ has: page.locator('span', { hasText: /^Images$/ }) });
+    await expect(imagesMetric.locator('strong')).toHaveText(number(listed));
+    await expect(page.locator('#primary-nav a[data-route="images"] .nav-count')).toHaveText(number(listed));
+    // A failed count reads as a dash, never the row count.
+    await page.unroute(`**${IMAGES_COUNT_URL}`);
+    await page.route(`**${IMAGES_COUNT_URL}`, route => route.fulfill({ status: 503, json: { success: false, error: 'Unavailable' } }));
+    await page.goto(`${workspace}#/system/appearance`);
+    await page.locator('#workspace-main .section-nav').getByRole('button', { name: 'Overview', exact: true }).click();
+    await expect(imagesMetric.locator('strong')).toHaveText('–');
+});
+
 // D129 (#249 S6b): where the library lives. The hook makes the fixture's database read as a node-folder library; its
 // Move library is a stand-in that touches no file (workspace_system_fixture.py).
 const banner = page => page.locator('#library-banner');
@@ -364,7 +388,7 @@ test('View backups / Appearance / Manage sources switch sections', async ({ page
    }
 });

-test('statistics dialog lists runs by category and library totals', async ({ page, request }) => {
+test('statistics dialog lists runs by category and database record counts', async ({ page, request }) => {
    const statistics = await serverJson(request, '/prompt_manager/system/statistics');
    expect(statistics.by_category.length).toBeGreaterThan(1);
    const layout = await openOverview(page);
@@ -373,13 +397,14 @@ test('statistics dialog lists runs by category and library totals', async ({ pag
    const dialog = page.locator('dialog#modal[open]');
    await expect(dialog.locator('#modal-title')).toHaveText('Usage statistics');
    const tables = dialog.locator('#modal-body > table');
-    await expect(tables.locator('caption')).toHaveText(['Runs by category', 'Library totals']);
+    await expect(tables.locator('caption')).toHaveText(['Runs by category', 'Database records']);
    // D153 (#447): no rating column and no rated-images total.
    await expect(tables.first().locator('thead th')).toHaveText(['Category', 'Runs']);
    const cells = table => table.locator('tbody tr').evaluateAll(rows => rows.map(row => [...row.cells].map(cell => cell.textContent)));
    expect(await cells(tables.first())).toEqual(statistics.by_category.map(row => [row.category ?? 'Uncategorized', number(row.runs)]));
-    expect(await cells(tables.nth(1))).toEqual([['Prompts', number(statistics.totals.total_prompts)],
-        ['Tags', number(statistics.totals.total_tags)], ['Images', number(statistics.totals.total_images)]]);
+    // #555: raw rows of the legacy /stats, labelled as rows so they never read as the Library or Images page counts.
+    expect(await cells(tables.nth(1))).toEqual([['Prompt records', number(statistics.totals.total_prompts)],
+        ['Tags on prompts', number(statistics.totals.total_tags)], ['Image records', number(statistics.totals.total_images)]]);
    await page.keyboard.press('Escape');
    await expect(explore).toBeFocused();
    // A statistics read that fails says so; the Overview stays.
@@ -145,9 +145,10 @@ test('long vocabularies append and Load more is keyboard reachable',async({page,
    await expect(loadMore(page)).toHaveCount(0);
    await expect(rows(page).nth(200).getByRole('button',{name:'Browse',exact:true})).toBeFocused();
    await expect(rows(page).locator('td:nth-child(1)')).toHaveText(manifest.tags);
-    // Reload with the window at its top: Firefox restores a scrolled window, whose end then appends a page before the check.
-    await main(page).locator('.two-col .table-wrap').evaluate(node=>new Promise(resolve=>{node.scrollTop=0;requestAnimationFrame(()=>requestAnimationFrame(resolve));}));
-    await page.reload();
+    // Open the screen as a new history entry, not a reload: Firefox restores a reloaded page's scrolled tag window, and
+    // its end then appends a page before the check (#567). Scrolling the window to its top first does not always stick.
+    await page.goto('about:blank');
+    await page.goto(route);
    await expect(rows(page)).toHaveCount(100);
    // D183: the end of the list arrives inside the tag window.
    const scroll=top=>main(page).locator('.two-col .table-wrap').evaluate((node,y)=>new Promise(resolve=>{node.scrollTop=y ?? node.scrollHeight;requestAnimationFrame(()=>requestAnimationFrame(resolve));}),top);
@@ -16,12 +16,22 @@ const OVERVIEW = Object.freeze({

 test('metrics are the server counts and Not checked before any health check', async () => {
    const { metricValues } = await overviewModule;
-    assert.deepEqual(metricValues(OVERVIEW), [
+    assert.deepEqual(metricValues(OVERVIEW, 24), [
        [(1234).toLocaleString(), 'Positive prompts'], ['7', 'Negative variants'], ['24', 'Images'],
        ['Not checked', 'Database integrity'],
    ]);
 });

+test('#555: the Images metric is the Images page count (the sidebar badge), not the indexed rows', async () => {
+    const { metricValues } = await overviewModule;
+    // The demo library: 110 indexed rows, 122 files on the Images page (unlinked ones included).
+    const overview = { ...OVERVIEW, images: { indexed: 110, rated: 0 } };
+    assert.deepEqual(metricValues(overview, 122)[2], [(122).toLocaleString(), 'Images']);
+    // Not read yet (undefined) or failed (null): a dash, as on the workspace Overview, never the row count.
+    assert.deepEqual(metricValues(overview, undefined)[2], ['–', 'Images']);
+    assert.deepEqual(metricValues(overview, null)[2], ['–', 'Images']);
+});
+
 test('metric 4 is the last health result', async () => {
    const { integrityText } = await overviewModule;
    assert.equal(integrityText(null), 'Not checked');
@@ -150,7 +160,7 @@ test('statistics rows are the real runs, then the library totals, without rating
        rated_images: 20,
    };
    assert.deepEqual(categoryRows(statistics), [['Landscapes', (1200).toLocaleString()], ['Uncategorized', '3']]);
-    assert.deepEqual(totalsRows(statistics), [['Prompts', '12'], ['Tags', '15'], ['Images', '24']]);
+    assert.deepEqual(totalsRows(statistics), [['Prompt records', '12'], ['Tags on prompts', '15'], ['Image records', '24']]);
 });

 /** A fake api answering each `get`/`post` from a queue per URL, recording the calls. */
@@ -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,
@@ -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):
@@ -564,6 +564,66 @@ class TestUndecidedGraph(unittest.TestCase):
        self.assertEqual(decision.review, "untraced")


+def _search_list(text="fox", side="positive"):
+    """f22: a PromptSearchList batch (no PromptManager node) feeds CLIPTextEncode."""
+    graph = parity.build("f22")["prompt"]
+    graph["10"]["inputs"]["text"] = text
+    if side == "negative":
+        # The search feeds the negative; the positive is typed text.
+        graph["6"]["inputs"]["text"] = parity.POS
+        graph["7"]["inputs"]["text"] = ["10", 0]
+    return graph
+
+
+class TestSearchTextIsNeverAPrompt(unittest.TestCase):
+    """#548: the PromptSearchList query is a search, never the prompt."""
+
+    def test_the_search_text_is_never_linked_or_stored(self):
+        for policy in Policy:
+            with self.subTest(policy=policy):
+                decision = _decide(
+                    _graph(_search_list()), policy=policy, lookup=_Lookup()
+                )
+                self.assertEqual(decision.kind, "never")
+                self.assertIsNone(decision.text)
+                self.assertEqual(decision.unlinked_status, "unresolved")
+                self.assertTrue(decision.try_recorded_run)
+
+    def test_a_search_text_from_a_primitive_is_never_linked(self):
+        graph = _search_list()
+        graph["10"]["inputs"]["text"] = ["30", 0]
+        graph["30"] = {"class_type": "PrimitiveString", "inputs": {"value": "fox"}}
+        for policy in Policy:
+            with self.subTest(policy=policy):
+                decision = _decide(_graph(graph), policy=policy, lookup=_Lookup())
+                self.assertEqual(decision.kind, "never")
+                self.assertIsNone(decision.text)
+
+    def test_an_empty_search_text_is_unchanged(self):
+        """Regression guard (passes on the base): nothing to exclude."""
+        for policy in Policy:
+            with self.subTest(policy=policy):
+                decision = _decide(
+                    _graph(_search_list("")), policy=policy, lookup=_Lookup()
+                )
+                self.assertEqual((decision.kind, decision.text), ("never", None))
+
+    def test_without_a_sampler_any_search_list_query_is_excluded(self):
+        graph = _search_list()
+        del graph["3"]
+        self.assertTrue(file_prompt._search_text(graph, " fox "))
+        self.assertFalse(file_prompt._search_text(graph, "a fox"))
+        self.assertFalse(file_prompt._search_text(None, "fox"))
+
+    def test_a_search_on_the_negative_keeps_the_typed_positive(self):
+        """Regression guard (passes on the base): only the positive's source
+        is excluded, so the legacy answer from typed text still links."""
+        graph = _search_list(side="negative")
+        decision = _decide(_graph(graph), lookup=_Lookup())
+        self.assertEqual((decision.kind, decision.text), ("positive", parity.POS))
+        self.assertEqual(decision.source, "legacy-parser")
+
+
 class TestStepsWithoutAnUndecidedGraph(unittest.TestCase):
    def test_step_9_ambiguous_parameters_only(self):
        meta = {
@@ -392,7 +392,10 @@ class TestStartStop(unittest.TestCase):
            self.assertLogs("prompt_manager.image_monitor", "WARNING") as logs,
        ):
            self.assertFalse(self.monitor.stop_monitoring(timeout=5.0))
-        self.assertEqual(fake.calls[-1], ("join", 0.0))
+        # The stop runs on its own thread and gets 0 s too: it may not have
+        # returned yet, and then nothing is joined at all.
+        joins = [call for call in fake.calls if call[0] == "join"]
+        self.assertTrue(all(call == ("join", 0.0) for call in joins), fake.calls)
        self.assertIn(
            "watchdog observer still running after 0.0 s", "\n".join(logs.output)
        )
@@ -401,6 +404,80 @@ class TestStartStop(unittest.TestCase):
    def test_stop_monitoring_without_an_observer_returns_true(self):
        self.assertTrue(self.monitor.stop_monitoring())

+    def test_threads_the_observer_starts_are_daemons(self):
+        # watchdog 7 on Windows starts a plain threading.Thread per watch in
+        # observer.start(); a non-daemon one holds the interpreter's exit
+        # before any atexit callback runs (#561).
+        fake = ReaderObserver()
+        self.addCleanup(fake.release.set)
+        self._start_with(fake)
+        self.assertIsNotNone(fake.reader)
+        self.assertTrue(fake.reader.daemon)
+        self.assertFalse(threading.current_thread().daemon)
+
+    def test_an_observer_that_fails_to_start_on_its_thread_is_reported(self):
+        fake = FakeObserver()
+        fake.start = MagicMock(side_effect=OSError("handle refused"))
+        with self.assertLogs("prompt_manager.image_monitor", "ERROR") as logs:
+            with patch.object(im_mod, "Observer", return_value=fake):
+                self.monitor.start_monitoring([self.tmp.name])
+        self.assertIn("handle refused", "\n".join(logs.output))
+        self.assertFalse(self.monitor.is_monitoring)
+
+    def test_an_observer_stop_that_never_returns_is_bounded(self):
+        # watchdog's observer.stop() joins its emitters (and, in 7.x on
+        # Windows, their reader threads) without a limit (#561).
+        fake = StuckStopObserver()
+        self.addCleanup(fake.release.set)
+        self._start_with(fake)
+        with self.assertLogs("prompt_manager.image_monitor", "WARNING") as logs:
+            self.assertFalse(self.monitor.stop_monitoring(timeout=0.05))
+        self.assertIn("watchdog observer still running", "\n".join(logs.output))
+        self.assertTrue(fake.entered.wait(30))
+        self.assertFalse(self.monitor.is_monitoring)
+        # A second stop waits for the stop already running; it never calls
+        # observer.stop() a second time.
+        fake.release.set()
+        self.assertTrue(self.monitor.stop_monitoring())
+        self.assertEqual(fake.stop_calls, 1)
+        self.assertFalse(self.monitor._unfinished)
+
+
+class ReaderObserver(FakeObserver):
+    """A FakeObserver whose start() starts a plain thread, as watchdog 7's
+    Windows emitter does for its ReadDirectoryChangesW reader."""
+
+    def __init__(self):
+        super().__init__()
+        self.release = threading.Event()
+        self.reader = None
+
+    def start(self):
+        self.reader = threading.Thread(target=self.release.wait)
+        self.reader.start()
+        super().start()
+
+    def stop(self):
+        self.release.set()
+        self.reader.join(30)
+        super().stop()
+
+
+class StuckStopObserver(FakeObserver):
+    """A FakeObserver whose stop() does not return until ``release``."""
+
+    def __init__(self):
+        super().__init__()
+        self.entered = threading.Event()
+        self.release = threading.Event()
+        self.stop_calls = 0
+
+    def stop(self):
+        self.stop_calls += 1
+        self.entered.set()
+        self.release.wait(30)
+        super().stop()
+

 class FakeEvent:
    def __init__(self, path, is_directory=False):
@@ -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)
@@ -136,6 +136,11 @@ def cleanup_package(package, require_stopped=True):
    if registration is not None:
        assert registration.dispose(3), "package history worker leaked"
    close_every_database()
+    # The package's exit actions outlive it otherwise: at interpreter exit
+    # they would import from a package no longer in sys.modules (#580).
+    exit_signal = getattr(package, "_exit_signal", None)
+    if exit_signal is not None:
+        exit_signal.reset()
    sys.modules.pop(package.__name__, None)


@@ -390,7 +395,39 @@ class TestConsoleBanner(unittest.TestCase):

        stream.flush()
        banner = stream.buffer.getvalue().decode("cp1252")
-        self.assertIn("Loaded:", banner)
+        lines = [
+            line
+            for line in banner.splitlines()
+            if "PromptManager" in line or "loaded" in line
+        ]
+        self.assertEqual(len(lines), 1, banner)
+        self.assertRegex(
+            lines[0], r"\[ComfyUI-PromptManager\] Version:.* \S+ - \d+ tools loaded$"
+        )
+        self.assertNotIn("Loaded:", banner)
+
+
+class TestFixtureExitActions(unittest.TestCase):
+    """A disposed fixture package runs nothing at interpreter exit (#580)."""
+
+    def test_a_disposed_package_leaves_no_exit_action(self):
+        tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(tmp.cleanup)
+        self.addCleanup(close_every_database)
+        output_dir = os.path.join(tmp.name, "output")
+        os.makedirs(output_dir)
+        name = "prompt_manager_pkg_exit_actions"
+        self.addCleanup(sys.modules.pop, name, None)
+
+        with contextlib.redirect_stdout(io.StringIO()):
+            with load_package(name, os.path.join(tmp.name, "p.db"), output_dir) as (
+                package,
+                _,
+            ):
+                signal = package._exit_signal
+                self.assertTrue(signal._on_exit, "the package installs exit actions")
+
+        self.assertEqual(signal._on_exit, [])


 class PriorFileHandler(logging.FileHandler):
@@ -553,7 +590,7 @@ class TestPackageLoggerIsolation(unittest.TestCase):
        original_print = print

        def fail_banner(*args, **kwargs):
-            if any("Total:" in str(arg) for arg in args):
+            if any("tools loaded" in str(arg) for arg in args):
                captured["package"] = sys.modules[name]
                captured["handlers"] = [
                    h
@@ -587,13 +624,16 @@ PACKAGE_THREAD_NAMES = ("PromptManagerPromptCleanup", "PromptManagerImageWorker"

 def package_threads():
    """Live threads this design owns: the tracker, the image worker, and the
-    watchdog observer and emitters (which have no name of their own)."""
+    watchdog observer and emitters (which have no name of their own).
+    watchdog 7's InotifyFD, one daemon reader per process that no observer
+    stop ends, is not the package's."""
    from watchdog.utils import BaseThread

    return {
        thread
        for thread in threading.enumerate()
-        if thread.name in PACKAGE_THREAD_NAMES or isinstance(thread, BaseThread)
+        if thread.name in PACKAGE_THREAD_NAMES
+        or (isinstance(thread, BaseThread) and type(thread).__name__ != "InotifyFD")
    }


@@ -759,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)
@@ -108,6 +108,15 @@ LEGACY_WRONG = {
        "a cozy library with tall shelves",
        "legacy shows branch 1; the file name names branch 4's saver",
    ),
+    # #548, the one shape where nothing is read on purpose: legacy shows a
+    # PromptSearchList's search text ("fox") as the positive, but that is the
+    # library query, not the prompt the batch ran. The scan and the monitor
+    # never store or link it either (utils.file_prompt._search_text).
+    ("inspection_parity", "f22"): (
+        None,
+        "legacy shows the PromptSearchList search text; each item is decided "
+        "at run time, so no positive is read (#548)",
+    ),
 }


@@ -149,6 +158,24 @@ class LegacyParityGateTest(unittest.TestCase):
                self.assertEqual(text, expected)
                self.assertNotEqual(text, legacy)

+    def test_the_search_list_query_is_never_stored_either(self):
+        """#548: the scan and the monitor depart from legacy for f22 too."""
+        from utils.file_prompt import Policy, decide_file_prompt
+
+        graph, workflow = corpora()["inspection_parity"]["f22"]
+        self.assertEqual(legacy_positive(graph, workflow), "fox")
+        for policy in Policy:
+            with self.subTest(policy=policy):
+                decision = decide_file_prompt(
+                    {"prompt": graph, "workflow": workflow},
+                    file_name="f22.png",
+                    file_size=None,
+                    in_output=True,
+                    lookup=lambda: None,
+                    policy=policy,
+                )
+                self.assertEqual((decision.kind, decision.text), ("never", None))
+
    def test_no_graph_leaves_the_legacy_agreement(self):
        recorded = json.loads(AGREEMENT_PATH.read_text(encoding="utf-8"))
        current = {name: agreeing(graphs) for name, graphs in corpora().items()}
@@ -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()
@@ -31,22 +31,65 @@ CHILD = textwrap.dedent("""
    import threading

    root, tmp = sys.argv[1], sys.argv[2]
+    stuck = sys.argv[3:] == ["stuck-observer"]
    sys.path.insert(0, root)
    state = {}

+    # watchdog 7's Windows observer at its worst (#561): start() starts a
+    # plain reader thread, and stop() waits for it, which never ends.
+    class StuckObserver:
+
+        def __init__(self, *args, **kwargs):
+            self._never = threading.Event()
+            self._reader = None
+
+        def schedule(self, *args, **kwargs):
+            pass
+
+        def start(self):
+            self._reader = threading.Thread(
+                target=self._never.wait, name="StuckReader"
+            )
+            self._reader.start()
+
+        def stop(self):
+            self._reader.join()
+
+        def join(self, timeout=None):
+            self._reader.join(timeout)
+
+        def is_alive(self):
+            return self._reader.is_alive()
+
+    if stuck:
+        import watchdog.observers
+
+        watchdog.observers.Observer = StuckObserver
+
    def report():
        from watchdog.utils import BaseThread

        package = state.get("package")
        result = getattr(package, "_background_work_result", None)
        # The image worker, the tracker, and watchdog's observer and emitters.
+        # watchdog 7's InotifyFD is one daemon reader per process that no
+        # observer stop ends, by design; it is not the package's.
        alive = sorted(
            t.name
            for t in threading.enumerate()
            if t.name in ("PromptManagerImageWorker", "PromptManagerPromptCleanup")
-            or isinstance(t, BaseThread)
+            or (isinstance(t, BaseThread) and type(t).__name__ != "InotifyFD")
+        )
+        # Threads that would hold the exit before any atexit callback (#561).
+        holding = sorted(
+            t.name
+            for t in threading.enumerate()
+            if not t.daemon and t is not threading.main_thread()
+        )
+        print(
+            "PM_EXIT_REPORT "
+            + json.dumps({"result": result, "alive": alive, "holding": holding})
        )
-        print("PM_EXIT_REPORT " + json.dumps({"result": result, "alive": alive}))
        sys.stdout.flush()

    # First registered, so it runs last: after the package's teardown and
@@ -64,6 +107,10 @@ CHILD = textwrap.dedent("""
            fake_server_module(queue_task_done=True),
        )
    state["package"] = package
+    if stuck:
+        # The stop gives up after this long instead of the full 5 s.
+        monitor_module = sys.modules[package.__name__ + ".utils.image_monitor"]
+        monitor_module.OBSERVER_JOIN_SECONDS = 0.5
    monitor = package._global_image_monitor
    handler = monitor.handler if monitor is not None else None
    if handler is None:
@@ -78,47 +125,52 @@ CHILD = textwrap.dedent("""
    """)


+def run_child(testcase, *args):
+    """Run CHILD with ``args``; returns (completed, report, detail)."""
+    with tempfile.TemporaryDirectory() as tmp:
+        os.makedirs(os.path.join(tmp, "output"))
+        script = os.path.join(tmp, "child.py")
+        with open(script, "w", encoding="utf-8") as handle:
+            handle.write(CHILD)
+        env = dict(os.environ)
+        env.pop("PROMPT_MANAGER_CONFIG_PATH", None)
+        # The child's library and log file live in its own folder, never
+        # in the checkout's logs/ (which this runner may hold open) or a
+        # data folder the caller set.
+        env["PROMPT_MANAGER_DATA_DIR"] = tmp
+        env["PYTHONIOENCODING"] = "utf-8"
+        try:
+            completed = subprocess.run(
+                [sys.executable, script, ROOT, tmp, *args],
+                cwd=ROOT,
+                env=env,
+                capture_output=True,
+                timeout=CHILD_CEILING_SECONDS,
+                encoding="utf-8",
+                errors="replace",
+            )
+        except subprocess.TimeoutExpired as hung:
+            testcase.fail(
+                f"the child did not exit within {CHILD_CEILING_SECONDS} s\n"
+                f"stdout:\n{hung.stdout}\nstderr:\n{hung.stderr}"
+            )
+        # Leaving the block removes the folder: on Windows that fails
+        # if the child left a handle open, which it cannot once it exited,
+        # so the check here is that the child exited at all.
+    detail = f"stdout:\n{completed.stdout}\nstderr:\n{completed.stderr}"
+    testcase.assertEqual(completed.returncode, 0, detail)
+    reports = [
+        line[len(REPORT_PREFIX) :]
+        for line in completed.stdout.splitlines()
+        if line.startswith(REPORT_PREFIX)
+    ]
+    testcase.assertEqual(len(reports), 1, detail)
+    return completed, json.loads(reports[0]), detail
+
+
 class TestPackageExit(unittest.TestCase):
    def test_teardown_runs_at_a_normal_exit_with_the_worker_alive(self):
-        with tempfile.TemporaryDirectory() as tmp:
-            os.makedirs(os.path.join(tmp, "output"))
-            script = os.path.join(tmp, "child.py")
-            with open(script, "w", encoding="utf-8") as handle:
-                handle.write(CHILD)
-            env = dict(os.environ)
-            env.pop("PROMPT_MANAGER_CONFIG_PATH", None)
-            # The child's library and log file live in its own folder, never
-            # in the checkout's logs/ (which this runner may hold open) or a
-            # data folder the caller set.
-            env["PROMPT_MANAGER_DATA_DIR"] = tmp
-            env["PYTHONIOENCODING"] = "utf-8"
-            try:
-                completed = subprocess.run(
-                    [sys.executable, script, ROOT, tmp],
-                    cwd=ROOT,
-                    env=env,
-                    capture_output=True,
-                    timeout=CHILD_CEILING_SECONDS,
-                    encoding="utf-8",
-                    errors="replace",
-                )
-            except subprocess.TimeoutExpired as hung:
-                self.fail(
-                    f"the child did not exit within {CHILD_CEILING_SECONDS} s\n"
-                    f"stdout:\n{hung.stdout}\nstderr:\n{hung.stderr}"
-                )
-            # Leaving the block removes the folder: on Windows that fails
-            # if the child left a handle open, which it cannot once it exited,
-            # so the check here is that the child exited at all.
-        detail = f"stdout:\n{completed.stdout}\nstderr:\n{completed.stderr}"
-        self.assertEqual(completed.returncode, 0, detail)
-        reports = [
-            line[len(REPORT_PREFIX) :]
-            for line in completed.stdout.splitlines()
-            if line.startswith(REPORT_PREFIX)
-        ]
-        self.assertEqual(len(reports), 1, detail)
-        report = json.loads(reports[0])
+        completed, report, detail = run_child(self)
        # "thumbnails" is True here because no preview worker started; its
        # close result is tested in test_thumbnail_auto and test_init_package.
        self.assertEqual(
@@ -127,6 +179,7 @@ class TestPackageExit(unittest.TestCase):
            detail,
        )
        self.assertEqual(report["alive"], [], detail)
+        self.assertEqual(report["holding"], [], detail)
        # The teardown's own warnings. Import-time warnings (the API's
        # output-folder search finds no ComfyUI here) are not about exit.
        for text in (
@@ -137,6 +190,16 @@ class TestPackageExit(unittest.TestCase):
        ):
            self.assertNotIn(text, completed.stderr, detail)

-
-if __name__ == "__main__":
-    unittest.main()
+    def test_an_observer_that_never_stops_does_not_hold_the_exit(self):
+        # Before #561 the reader thread kept the interpreter waiting before
+        # any atexit callback ran, so the teardown never started.
+        completed, report, detail = run_child(self, "stuck-observer")
+        self.assertEqual(
+            report["result"],
+            {"monitor": False, "thumbnails": True, "tracker": True},
+            detail,
+        )
+        # The stuck reader is still alive, but as a daemon it holds nothing.
+        self.assertEqual(report["holding"], [], detail)
+        self.assertIn("watchdog observer still running", completed.stderr, detail)
+        self.assertIn("Background work not stopped in time", completed.stderr, detail)
@@ -90,6 +90,18 @@ class TestPairScan(AdminAPITestCase):
        self.assertEqual(self.api.db.list_prompt_pairs()["total"], 0)
        self.assertEqual(self._texts(), [])

+    def test_a_prompt_search_list_query_is_never_stored_as_a_prompt(self):
+        # #548: the search text was stored and every batch image linked to it.
+        g = sampled_graph()
+        g["10"] = {
+            "class_type": "PromptSearchList",
+            "inputs": {"text": "fox", "category": "", "tags": "", "limit": 50},
+        }
+        g["1"]["inputs"]["text"] = ["10", 0]
+        self.assertEqual(self._scanned("search.png", g), "unresolved")
+        self.assertEqual(self._texts(), [])
+        self.assertEqual(self.api.db.list_prompt_pairs()["total"], 0)
+
    def test_numeric_widgets_are_never_stored_as_a_prompt(self):
        # The pre-redesign scan stored "2048" from a resolution widget here.
        g = {
@@ -0,0 +1,455 @@
+"""PNG text without a pixel decode (#523), against the legacy reader floor.
+
+Pillow's ``PngImageFile.text`` loads (decodes) every pixel to reach text
+chunks written after IDAT. ``utils.png_text.png_text`` walks the remaining
+chunks instead. The legacy reader (``Image.open(path).text``, every failure
+read as ``{}``) is the floor: a file it read must read the same, and the
+curated corpus below must read identically. The documented difference is
+pixel data, which the walk never decodes: undecodable or cut-short pixel data
+made the legacy reader fail, the walk still reads the text before it.
+"""
+
+import io
+import random
+import struct
+import tempfile
+import unittest
+import zlib
+from pathlib import Path
+from unittest.mock import patch
+
+from PIL import Image, ImageFile, PngImagePlugin
+
+from utils.metadata_extractor import ComfyUIMetadataExtractor
+from utils.png_text import png_text
+
+FIXTURES = Path(__file__).resolve().parent / "fixtures"
+IMAGE_SUFFIXES = {".png", ".jpg", ".jpeg", ".webp", ".gif", ".bmp"}
+SIGNATURE = b"\x89PNG\r\n\x1a\n"
+FUZZ_SEED = 523
+FUZZ_CASES = 400
+
+
+def legacy_text(path):
+    """The legacy scan reader: Pillow's ``.text`` (a full decode)."""
+    try:
+        with Image.open(path) as image:
+            return dict(getattr(image, "text", None) or {})
+    except Exception:
+        return {}
+
+
+def walked_text(path):
+    """The new reader under the same failure rule."""
+    try:
+        with Image.open(path) as image:
+            return png_text(image)
+    except Exception:
+        return {}
+
+
+def chunk(kind, data):
+    return (
+        struct.pack(">I", len(data))
+        + kind
+        + data
+        + struct.pack(">I", zlib.crc32(kind + data) & 0xFFFFFFFF)
+    )
+
+
+def chunks_of(raw):
+    """``[(kind, data)]`` of a well-formed PNG."""
+    found, offset = [], 8
+    while offset < len(raw):
+        length, kind = struct.unpack(">I4s", raw[offset : offset + 8])
+        found.append((kind, raw[offset + 8 : offset + 8 + length]))
+        offset += 12 + length
+    return found
+
+
+def assemble(parts):
+    return SIGNATURE + b"".join(
+        part if isinstance(part, bytes) else chunk(*part) for part in parts
+    )
+
+
+def base_png(mode="RGB", size=(64, 64), text=None, noise=False):
+    """A PNG from Pillow (text before IDAT), as ``[(kind, data)]``."""
+    image = Image.new(mode, size)
+    if noise:
+        rng = random.Random(7)
+        image = Image.frombytes(
+            "RGB", size, bytes(rng.randrange(256) for _ in range(size[0] * size[1] * 3))
+        )
+    info = PngImagePlugin.PngInfo()
+    for key, value in (text or {}).items():
+        info.add_text(key, value)
+    buffer = io.BytesIO()
+    image.save(buffer, format="PNG", pnginfo=info)
+    return chunks_of(buffer.getvalue())
+
+
+def split_idat(parts, pieces=3):
+    """Spread the IDAT payload over ``pieces`` consecutive IDAT chunks."""
+    data = b"".join(d for k, d in parts if k == b"IDAT")
+    step = max(1, len(data) // pieces)
+    idats = [(b"IDAT", data[i : i + step]) for i in range(0, len(data), step)]
+    first = next(i for i, (k, _) in enumerate(parts) if k == b"IDAT")
+    rest = [p for p in parts if p[0] != b"IDAT"]
+    return rest[:first] + idats + rest[first:]
+
+
+def before_iend(parts, *extra):
+    return parts[:-1] + list(extra) + parts[-1:]
+
+
+def text_chunk(key, value):
+    return (b"tEXt", key.encode("latin-1") + b"\0" + value.encode("latin-1"))
+
+
+def ztxt_chunk(key, value):
+    return (b"zTXt", key.encode() + b"\0\0" + zlib.compress(value.encode("latin-1")))
+
+
+def itxt_chunk(key, value, compressed=False):
+    body = value.encode("utf-8")
+    if compressed:
+        body = zlib.compress(body)
+    flag = b"\1\0" if compressed else b"\0\0"
+    return (b"iTXt", key.encode() + b"\0" + flag + b"en\0" + b"Title\0" + body)
+
+
+def bad_crc(kind, data):
+    good = chunk(kind, data)
+    return good[:-4] + bytes(b ^ 0xFF for b in good[-4:])
+
+
+PROMPT = '{"3": {"class_type": "KSampler", "inputs": {"seed": 1}}}'
+WORKFLOW = '{"nodes": [], "links": []}'
+
+
+def curated():
+    """Name to PNG bytes; each must read the same with both readers."""
+    plain = base_png(text={"prompt": PROMPT})
+    trailing = before_iend(
+        base_png(), text_chunk("prompt", PROMPT), text_chunk("workflow", WORKFLOW)
+    )
+    cases = {
+        "no-text": assemble(base_png()),
+        "text-before-pixels": assemble(plain),
+        "text-after-pixels": assemble(trailing),
+        "ztxt-after-pixels": assemble(
+            before_iend(base_png(), ztxt_chunk("prompt", PROMPT))
+        ),
+        "itxt-plain-and-compressed": assemble(
+            before_iend(
+                base_png(),
+                itxt_chunk("prompt", "café " + PROMPT),
+                itxt_chunk("workflow", WORKFLOW, compressed=True),
+            )
+        ),
+        "xmp-after-pixels": assemble(
+            before_iend(
+                base_png(),
+                itxt_chunk("XML:com.adobe.xmp", "<x:xmpmeta/>"),
+                text_chunk("prompt", PROMPT),
+            )
+        ),
+        "itxt-bad-utf8": assemble(
+            before_iend(
+                base_png(),
+                (b"iTXt", b"prompt\0\0\0\0\0\xff\xfe"),
+                text_chunk("workflow", WORKFLOW),
+            )
+        ),
+        "later-copy-wins": assemble(
+            before_iend(plain, text_chunk("prompt", '{"later": 1}'))
+        ),
+        "empty-key-and-no-separator": assemble(
+            before_iend(base_png(), (b"tEXt", b"\0value"), (b"tEXt", b"lonely"))
+        ),
+        "split-pixels-text-between": assemble(
+            split_idat(before_iend(base_png(noise=True), text_chunk("prompt", PROMPT)))
+        ),
+        "palette": assemble(
+            before_iend(base_png(mode="P"), text_chunk("prompt", PROMPT))
+        ),
+        "sixteen-bit": assemble(
+            before_iend(base_png(mode="I;16"), text_chunk("prompt", PROMPT))
+        ),
+        "bad-crc-before-pixels": assemble(
+            [base_png()[0], bad_crc(*text_chunk("prompt", PROMPT))] + base_png()[1:]
+        ),
+        "bad-crc-after-pixels": assemble(
+            base_png()[:-1] + [bad_crc(*text_chunk("prompt", PROMPT))] + base_png()[-1:]
+        ),
+        "unknown-chunk-after-pixels": assemble(
+            before_iend(base_png(), (b"prVt", b"x" * 40), text_chunk("prompt", PROMPT))
+        ),
+        "garbage-chunk-id-after-pixels": assemble(
+            base_png()[:-1]
+            + [b"\0\0\0\4\x00\x01\x02\x03abcd\0\0\0\0"]
+            + [chunk(*text_chunk("prompt", PROMPT))]
+            + base_png()[-1:]
+        ),
+        "oversized-ztxt-after-pixels": assemble(
+            before_iend(
+                base_png(),
+                (
+                    b"zTXt",
+                    b"prompt\0\0"
+                    + zlib.compress(b"a" * (PngImagePlugin.MAX_TEXT_CHUNK + 10)),
+                ),
+            )
+        ),
+        "pixels-length-past-end": assemble(
+            [p for p in base_png() if p[0] != b"IDAT" and p[0] != b"IEND"]
+        )
+        + b"\xff\xff\x00\x00IDAT"
+        + b"\0" * 64,
+    }
+    full = assemble(trailing)
+    idat_at = full.index(b"IDAT")
+    cases["truncated-in-pixels"] = full[: idat_at + 20]
+    cases["no-end-after-pixels"] = full[: full.rindex(b"IEND") - 4]
+    cases["truncated-in-trailing-text"] = full[: full.rindex(b"workflow") + 4]
+    cases["unknown-chunk-past-end"] = (
+        assemble(before_iend(base_png(), ztxt_chunk("prompt", PROMPT))[:-1])
+        + chunk(b"prVt", b"x" * 40)[:-20]
+    )
+    cases["not-a-png"] = b"nope, not an image at all"
+    cases["signature-only"] = SIGNATURE
+    return cases
+
+
+def corrupt_pixels():
+    """Complete IDAT chunks with valid CRCs over data zlib cannot inflate."""
+    parts = before_iend(base_png(), text_chunk("prompt", PROMPT))
+    return assemble(
+        [(k, b"\x78\x9c" + b"\xff" * 30) if k == b"IDAT" else (k, d) for k, d in parts]
+    )
+
+
+def apng():
+    frames = [Image.new("RGB", (16, 16), colour) for colour in ("red", "blue")]
+    info = PngImagePlugin.PngInfo()
+    info.add_text("prompt", PROMPT)
+    buffer = io.BytesIO()
+    frames[0].save(
+        buffer, format="PNG", save_all=True, append_images=frames[1:], pnginfo=info
+    )
+    return assemble(
+        before_iend(chunks_of(buffer.getvalue()), text_chunk("workflow", WORKFLOW))
+    )
+
+
+def fixture_images():
+    return sorted(p for p in FIXTURES.rglob("*") if p.suffix.lower() in IMAGE_SUFFIXES)
+
+
+class _TempFiles(unittest.TestCase):
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self._tmp.cleanup)
+
+    def write(self, name, raw):
+        path = Path(self._tmp.name) / name
+        path.write_bytes(raw)
+        return path
+
+
+class PngTextParityTests(_TempFiles):
+    def test_every_fixture_image_reads_like_the_legacy_reader(self):
+        images = fixture_images()
+        self.assertGreaterEqual(len(images), 25)
+        for path in images:
+            with self.subTest(path=path.relative_to(FIXTURES)):
+                self.assertEqual(walked_text(path), legacy_text(path))
+
+    def test_curated_corpus_reads_like_the_legacy_reader(self):
+        for name, raw in curated().items():
+            with self.subTest(name):
+                path = self.write(name + ".png", raw)
+                self.assertEqual(walked_text(path), legacy_text(path))
+
+    def test_curated_corpus_exercises_trailing_text(self):
+        """The corpus is not vacuous: trailing text exists and is read."""
+        cases = curated()
+        path = self.write("t.png", cases["text-after-pixels"])
+        self.assertEqual(walked_text(path), {"prompt": PROMPT, "workflow": WORKFLOW})
+        with Image.open(path) as image:
+            self.assertNotIn("prompt", image.info)
+        self.assertEqual(
+            walked_text(self.write("x.png", cases["truncated-in-pixels"])), {}
+        )
+
+    def test_itxt_values_keep_their_language_and_type(self):
+        path = self.write("i.png", curated()["itxt-plain-and-compressed"])
+        new, old = walked_text(path), legacy_text(path)
+        self.assertIsInstance(new["prompt"], PngImagePlugin.iTXt)
+        self.assertEqual(new["prompt"].lang, old["prompt"].lang)
+        self.assertEqual(new["prompt"].tkey, old["prompt"].tkey)
+
+    def test_animated_png_falls_back_to_the_legacy_read(self):
+        path = self.write("a.png", apng())
+        self.assertEqual(walked_text(path), legacy_text(path))
+        self.assertEqual(walked_text(path)["workflow"], WORKFLOW)
+
+    def test_truncated_pixels_read_alike_when_truncated_images_load(self):
+        """With LOAD_TRUNCATED_IMAGES set (a node pack may set it) both readers
+        accept a short file alike."""
+        path = self.write("t.png", curated()["truncated-in-pixels"])
+        with patch.object(ImageFile, "LOAD_TRUNCATED_IMAGES", True):
+            self.assertEqual(walked_text(path), legacy_text(path))
+
+    def test_a_text_decode_error_ends_the_walk_like_legacy(self):
+        """Pillow's ``load_end`` stops at a chunk whose handler raises
+        ``UnicodeDecodeError`` and keeps the text read so far."""
+        path = self.write(
+            "u.png",
+            assemble(
+                before_iend(
+                    base_png(),
+                    ztxt_chunk("prompt", PROMPT),
+                    text_chunk("workflow", WORKFLOW),
+                    ztxt_chunk("parameters", "steps 20"),
+                )
+            ),
+        )
+
+        def undecodable(stream, position, length):
+            raise UnicodeDecodeError("latin-1", b"", 0, 1, "test")
+
+        with patch.object(PngImagePlugin.PngStream, "chunk_tEXt", undecodable):
+            old = legacy_text(path)
+            self.assertEqual(walked_text(path), old)
+        self.assertEqual(old, {"prompt": PROMPT})
+
+    def test_the_text_memory_budget_counts_text_before_the_pixels(self):
+        """Pillow's MAX_TEXT_MEMORY spans the whole file: text before and
+        after the pixels, each under it, fail together in both readers."""
+        path = self.write(
+            "m.png",
+            assemble(
+                before_iend(
+                    base_png(text={"prompt": PROMPT}), text_chunk("workflow", WORKFLOW)
+                )
+            ),
+        )
+        budget = len(PROMPT) + len(WORKFLOW) - 1
+        with patch.object(PngImagePlugin, "MAX_TEXT_MEMORY", budget):
+            self.assertEqual(legacy_text(path), {})
+            self.assertEqual(walked_text(path), {})
+        with patch.object(PngImagePlugin, "MAX_TEXT_MEMORY", budget + 1):
+            self.assertEqual(walked_text(path), legacy_text(path))
+            self.assertEqual(len(walked_text(path)), 2)
+
+    def test_complete_but_undecodable_pixels_read_more_than_legacy(self):
+        path = self.write("c.png", corrupt_pixels())
+        self.assertEqual(legacy_text(path), {})
+        self.assertEqual(walked_text(path), {"prompt": PROMPT})
+
+    def test_cut_short_pixels_read_the_text_before_them(self):
+        full = assemble(base_png(text={"prompt": PROMPT}))
+        path = self.write("s.png", full[: full.index(b"IDAT") + 12])
+        self.assertEqual(legacy_text(path), {})
+        self.assertEqual(walked_text(path), {"prompt": PROMPT})
+
+    def test_mutated_files_never_read_less_than_legacy(self):
+        """Seeded fuzz: truncations and byte flips of a file with text before,
+        between and after its pixel chunks. Wherever the legacy reader found
+        text, the walk finds exactly the same."""
+        base = assemble(
+            split_idat(
+                before_iend(
+                    base_png(text={"workflow": WORKFLOW}),
+                    ztxt_chunk("prompt", PROMPT),
+                    itxt_chunk("parameters", "steps 20"),
+                )
+            )
+        )
+        rng = random.Random(FUZZ_SEED)
+        compared = 0
+        for case in range(FUZZ_CASES):
+            raw = bytearray(base)
+            if case % 2:
+                del raw[rng.randrange(8, len(raw)) :]
+            else:
+                for _ in range(rng.randrange(1, 4)):
+                    raw[rng.randrange(8, len(raw))] = rng.randrange(256)
+            path = self.write(f"fuzz{case}.png", bytes(raw))
+            old = legacy_text(path)
+            if old:
+                compared += 1
+                with self.subTest(case=case):
+                    self.assertEqual(walked_text(path), old)
+        self.assertGreater(compared, FUZZ_CASES // 5)
+
+    def test_non_png_images_have_no_text(self):
+        for name in ("tiny.jpg", "tiny.webp"):
+            with Image.open(FIXTURES / "local-inspect" / name) as image:
+                self.assertEqual(png_text(image), {})
+
+
+class NoPixelDecodeTests(_TempFiles):
+    """Metadata reads never decode pixels (the #523 memory finding)."""
+
+    def png_fixtures(self):
+        return sorted((FIXTURES / "inspection_parity").glob("*.png"))
+
+    def test_png_text_does_not_load_pixels_and_leaves_the_image_usable(self):
+        path = self.write("t.png", curated()["text-after-pixels"])
+        with Image.open(path) as image:
+            with patch.object(ImageFile.ImageFile, "load", autospec=True) as load:
+                self.assertEqual(png_text(image)["workflow"], WORKFLOW)
+            load.assert_not_called()
+            image.load()
+            self.assertEqual(image.size, (64, 64))
+            self.assertEqual(image.text["workflow"], WORKFLOW)
+
+    def test_scan_reader_does_not_decode_pixels(self):
+        from py.api import PromptManagerAPI
+
+        api = PromptManagerAPI.__new__(PromptManagerAPI)
+        fixtures = self.png_fixtures() + [
+            self.write("t.png", curated()["text-after-pixels"])
+        ]
+        with patch.object(ImageFile.ImageFile, "load", autospec=True) as load:
+            found = [api._extract_comfyui_metadata(str(p)) for p in fixtures]
+        load.assert_not_called()
+        self.assertEqual(found, [legacy_text(p) for p in fixtures])
+        self.assertTrue(all(found))
+
+    def test_monitor_extractor_does_not_decode_pixels(self):
+        extractor = ComfyUIMetadataExtractor()
+        fixtures = self.png_fixtures()
+        with patch.object(ImageFile.ImageFile, "load", autospec=True) as load:
+            found = [extractor.extract_metadata(str(p)) for p in fixtures]
+        load.assert_not_called()
+        self.assertTrue(all(metadata and len(metadata) > 1 for metadata in found))
+
+
+class MonitorExtractorParityTests(_TempFiles):
+    """``extract_metadata`` answers exactly as with the legacy ``.text``."""
+
+    def legacy_extract(self, path):
+        def old_text(image):
+            return dict(getattr(image, "text", None) or {})
+
+        with patch("utils.metadata_extractor.png_text", old_text):
+            return ComfyUIMetadataExtractor().extract_metadata(str(path))
+
+    def test_every_fixture_and_curated_file_extracts_like_legacy(self):
+        paths = fixture_images() + [
+            self.write(name + ".png", raw) for name, raw in curated().items()
+        ]
+        extractor = ComfyUIMetadataExtractor()
+        for path in paths:
+            with self.subTest(path=path.name):
+                self.assertEqual(
+                    extractor.extract_metadata(str(path)), self.legacy_extract(path)
+                )
+
+
+if __name__ == "__main__":
+    unittest.main()
@@ -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(
@@ -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()
@@ -331,7 +331,7 @@ const testJobs = () => jobs(readWorkflow('test.yml'));

 test('the Python matrix is Linux 3.12 on a pull request and Linux 3.12-3.14 on every other event', () => {
    const body = testJobs().get('test');
-    assert.deepEqual(matrixList(body, 'os'), ['ubuntu-latest'], 'Windows and macOS never gate a merge');
+    assert.deepEqual(matrixList(body, 'os'), ['ubuntu-latest'], 'Windows never gates a merge');
    const match = /^ {8}python-version: \$\{\{ github\.event_name == 'pull_request' && fromJSON\('(\[[^\]]*\])'\) \|\| fromJSON\('(\[[^\]]*\])'\) \}\}$/m.exec(body);
    assert.ok(match, 'python-version is chosen by event');
    assert.deepEqual({ pr: JSON.parse(match[1]), other: JSON.parse(match[2]) }, { pr: ['3.12'], other: ['3.12', '3.13', '3.14'] });
@@ -339,34 +339,21 @@ test('the Python matrix is Linux 3.12 on a pull request and Linux 3.12-3.14 on e
    assert.match(body, /^ {4}name: py\$\{\{ matrix\.python-version \}\} on \$\{\{ matrix\.os \}\}$/m);
 });

-test('Windows and macOS run 3.12-3.14 on main, nightly and on demand, never on a PR or in the queue', () => {
+test('Windows runs 3.12-3.14 on main, nightly and on demand, never on a PR or in the queue', () => {
    const text = readWorkflow('portability.yml');
    const triggers = text.slice(text.indexOf('\non:'), text.indexOf('\nconcurrency:'));
    for (const event of ['push', 'schedule', 'workflow_dispatch']) assert.match(triggers, new RegExp(`^ {2}${event}:`, 'm'));
    for (const event of ['pull_request', 'merge_group']) assert.doesNotMatch(triggers, new RegExp(`^ {2}${event}`, 'm'));
    assert.match(triggers, /^ {4}branches: \[main\]$/m, 'push runs only for main');
    const body = jobs(text).get('test');
-    assert.deepEqual(matrixList(body, 'os'), ['windows-latest', 'macos-latest']);
+    assert.deepEqual(matrixList(body, 'os'), ['windows-latest'], 'macOS is not tested in CI (#571)');
+    assert.doesNotMatch(text, /macos-/, 'no macOS runner is used');
    assert.deepEqual(matrixList(body, 'python-version'), ['"3.12"', '"3.13"', '"3.14"']);
    assert.match(body, /^ {6}fail-fast: false$/m);
    assert.match(body, /^ {8}run: python -m tests\.run_suite -v$/m);
    assert.doesNotMatch(text, /cancel-in-progress: true|cancel-in-progress: \$/, 'no portability run is cancelled');
 });

-// #499: the native trash proof runs on both macOS CPUs, apart from the suite so a red suite never skips it.
-test('portability runs the native trash proof on arm64 and Intel macOS', () => {
-    const body = jobs(readWorkflow('portability.yml')).get('native-trash');
-    assert.ok(body, 'portability.yml has a native-trash job');
-    assert.deepEqual(matrixList(body, 'os'), ['macos-latest', 'macos-15-intel']);
-    assert.match(body, /^ {6}fail-fast: false$/m);
-    assert.doesNotMatch(body, /^ {4}needs:/m, 'it does not wait for the suite job');
-    assert.match(body, /^ {10}PM_NATIVE_TRASH: macos$/m);
-    assert.match(body, /^ {8}run: python -m unittest tests\.test_trash_macos_native tests\.test_trash_macos -v$/m);
-    const jobMinutes = Number(/^ {4}timeout-minutes: (\d+)$/m.exec(body)[1]);
-    const stepMinutes = Number(/^ {8}timeout-minutes: (\d+)$/m.exec(body)?.[1]);
-    assert.ok(stepMinutes > 0 && jobMinutes - stepMinutes >= 2, 'the proof step stops before the job limit');
-});
-
 // #493: the portability limit follows the rule in its comment: ceil(1.5 x the slowest of three
 // green main runs) + 3 minutes, never under 5, with those runs and times written above jobs:.
 // The times are whole-job wall times. That the runs were green and on main is the author's word.
@@ -427,5 +414,5 @@ test('every required check reports on a pull request and the gates carry the ful
    ].sort());
    assert.equal(contexts.filter(context => /^py/.test(context)).length, 1,
        'a dropped matrix leg never reports on a pull request, so it cannot be required');
-    assert.ok(!contexts.some(context => /windows|macos/.test(context)), 'Windows and macOS never gate a merge');
+    assert.ok(!contexts.some(context => /windows|macos/.test(context)), 'Windows never gates a merge');
 });
@@ -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()
@@ -230,6 +230,9 @@ def _undecided_graph(found: _File) -> Decision:
    if switched is not None:
        # The legacy answer is a switched-off node's own text (#90)
        return Decision(switched, canonical=found.canonical)
+    if text is not None and _search_text(found.canonical["prompt"], text):
+        # The query of a Search List batch is a search, never a prompt (#548)
+        return _never(found, _unlinked(found), try_recorded_run=True)
    if text is not None and _joined_text(found.canonical["prompt"], text):
        # The legacy parser joins prepend + text + append: the text a
        # PromptManager node stores cannot be told from it (#343).
@@ -311,6 +314,8 @@ _NOT_TEXT_INPUTS = frozenset(
    }
 )
 _MAX_SWITCH_VISITS = 5000
+# The batch node whose ``text`` input is a library search query (#548).
+SEARCH_LIST_TYPE = "PromptSearchList"


 def _negative_name(key: Any) -> bool:
@@ -328,10 +333,28 @@ def _switched_off_upstream(graph: Any) -> Optional[str]:
    """
    if not isinstance(graph, dict):
        return None
-    roots = _positive_roots(graph)
-    if not roots:
+    upstream = _positive_upstream(graph)
+    if upstream is None:
        ids = _prompt_manager_ids(graph)
        return _switch_kind(save_setting(graph, node_id) for node_id in ids)
+    found = [
+        node_id
+        for node_id in upstream
+        if graph[node_id].get("class_type") in PROMPT_MANAGER_TYPES
+    ]
+    return _switch_kind(save_setting(graph, node_id) for node_id in found)
+
+
+def _positive_upstream(graph: dict) -> Optional[set]:
+    """The ids of the nodes upstream of a sampler's positive, or None when
+    the graph has no recognisable sampler.
+
+    Follows every input except negative and non-text ones (#90's walk follows
+    only prompt-like names).
+    """
+    roots = _positive_roots(graph)
+    if not roots:
+        return None
    seen, found = set(), set()
    stack = [str(link[0]) for link in roots]
    while stack and len(seen) < _MAX_SWITCH_VISITS:
@@ -342,8 +365,7 @@ def _switched_off_upstream(graph: Any) -> Optional[str]:
        node = graph.get(node_id)
        if not isinstance(node, dict):
            continue
-        if node.get("class_type") in PROMPT_MANAGER_TYPES:
-            found.add(node_id)
+        found.add(node_id)
        inputs = node.get("inputs")
        for key, value in (inputs.items() if isinstance(inputs, dict) else ()):
            if key in _NOT_TEXT_INPUTS or _negative_name(key):
@@ -351,7 +373,28 @@ def _switched_off_upstream(graph: Any) -> Optional[str]:
            if isinstance(value, (list, tuple)) and len(value) == 2:
                if isinstance(value[0], (str, int)) and isinstance(value[1], int):
                    stack.append(str(value[0]))
-    return _switch_kind(save_setting(graph, node_id) for node_id in found)
+    return found
+
+
+def _search_text(graph: Any, text: str) -> bool:
+    """Whether ``text`` is the search query of a PromptSearchList node that
+    feeds a positive (any such node when no sampler is recognised).
+
+    The legacy parser traces a positive wired from a Search List to the
+    node's ``text`` input, which is the database search, never the prompt
+    the batch ran (#548). Each item is counted by the node at run time.
+    """
+    if not isinstance(graph, dict):
+        return False
+    upstream = _positive_upstream(graph)
+    ids = graph if upstream is None else upstream
+    wanted = text.strip()
+    return any(
+        isinstance(graph[node_id], dict)
+        and graph[node_id].get("class_type") == SEARCH_LIST_TYPE
+        and resolve_text(graph, node_id) == wanted
+        for node_id in ids
+    )


 def _off_text(graph: Any, text: str) -> Optional[str]:
@@ -96,6 +96,65 @@ WORKER_JOIN_SECONDS = 5.0
 OBSERVER_JOIN_SECONDS = 5.0


+def _start_daemon_observer(observer) -> None:
+    """``observer.start()`` on a short-lived daemon thread (#561).
+
+    A thread is a daemon when the thread that creates it is one. watchdog
+    7 on Windows starts a plain ``threading.Thread`` per watch inside
+    ``start()``; created from ComfyUI's main thread it is not a daemon, and
+    the interpreter waits for every such thread before it runs any
+    ``atexit`` callback, so the teardown that would stop it never runs.
+    Started from here, every thread the observer creates is a daemon.
+    Waits for the start without a limit (start-up, not exit); re-raises
+    what ``start()`` raised.
+    """
+    raised: List[BaseException] = []
+
+    def start() -> None:
+        try:
+            observer.start()
+        except BaseException as e:  # re-raised on the caller's thread
+            raised.append(e)
+
+    starter = threading.Thread(
+        target=start, name="PromptManagerObserverStart", daemon=True
+    )
+    starter.start()
+    starter.join()
+    if raised:
+        raise raised[0]
+
+
+class _ObserverStop:
+    """One ``observer.stop()`` running on a daemon thread (#561).
+
+    watchdog's ``stop()`` joins its emitters (and, in 7.x on Windows, their
+    reader threads) without a limit. Run here, a stop that never returns
+    holds neither the caller past its deadline nor the interpreter's exit.
+    """
+
+    def __init__(self, observer) -> None:
+        self.error: Optional[Exception] = None
+        self._thread = threading.Thread(
+            target=self._run,
+            args=(observer,),
+            name="PromptManagerObserverStop",
+            daemon=True,
+        )
+        self._thread.start()
+
+    def _run(self, observer) -> None:
+        try:
+            observer.stop()
+        except Exception as e:  # reported by the stop that waits for it
+            self.error = e
+
+    def wait(self, timeout: float) -> bool:
+        """True once ``stop()`` has returned or raised."""
+        self._thread.join(timeout)
+        return not self._thread.is_alive()
+
+
 class ImageGenerationHandler(FileSystemEventHandler):
    """Filesystem event handler for detecting new image generation.

@@ -1170,6 +1229,9 @@ class ImageMonitor:
        # (observer, handler) pairs a stop left running: the next stop waits
        # for them again instead of reporting True.
        self._unfinished: List[Tuple[Any, Any]] = []
+        # Observer stops still running on their thread, by observer: a later
+        # stop waits for the same call instead of starting another (#561).
+        self._observer_stops: Dict[Any, _ObserverStop] = {}
        self.logger = get_logger("prompt_manager.image_monitor")
        self.review_recorder: Optional[Callable[[str], str]] = None
        self.thumbnail_queue: Optional[Callable[[str], str]] = None
@@ -1273,7 +1335,7 @@ class ImageMonitor:
            self.logger.warning("No directory could be watched")
            return
        try:
-            observer.start()
+            _start_daemon_observer(observer)
        except Exception as e:
            self.logger.error(f"Image monitoring could not start: {e}")
            return
@@ -1303,12 +1365,12 @@ class ImageMonitor:

        Closes the handler first (its worker is joined and its dispatch
        returns at once), then stops the watchdog observer and joins it. The
-        worker and the observer share the one deadline: the observer's join
-        gets what the worker left, possibly 0 s.
-        watchdog's ``observer.stop()`` joins its emitters without a limit;
-        it counts against the deadline but is not bounded here. A part an
-        earlier call left running is waited for again. Safe to call when
-        monitoring never started and more than once.
+        worker and the observer share the one deadline: the observer's stop
+        and join get what the worker left, possibly 0 s. watchdog's
+        ``observer.stop()`` joins its emitters without a limit, so it runs on
+        a daemon thread and is waited for only until the deadline (#561). A
+        part an earlier call left running is waited for again. Safe to call
+        when monitoring never started and more than once.

        Returns:
            True when every worker and observer it stopped has ended.
@@ -1349,13 +1411,27 @@ class ImageMonitor:
            return False

    def _stop_observer(self, observer, timeout: float) -> bool:
-        """``observer.stop()`` (safe to repeat), then a join of ``timeout``
-        seconds; a join that times out is logged with the time it got."""
+        """``observer.stop()`` on its own daemon thread, then a join, both
+        within ``timeout`` seconds; a stop or join still running then is
+        logged with the time it got. A stop still running from an earlier
+        call is waited for, never started again."""
        if observer is None:
            return True
+        deadline = time.monotonic() + timeout
+        stop = self._observer_stops.get(observer)
+        if stop is None:
+            stop = self._observer_stops[observer] = _ObserverStop(observer)
+        if not stop.wait(timeout):
+            self.logger.warning(
+                f"watchdog observer still running after {timeout:.1f} s"
+            )
+            return False
+        del self._observer_stops[observer]
+        if stop.error is not None:
+            self.logger.warning(f"Error while stopping image monitoring: {stop.error}")
+            return False
        try:
-            observer.stop()
-            observer.join(timeout)
+            observer.join(max(0.0, deadline - time.monotonic()))
        except Exception as e:
            self.logger.warning(f"Error while stopping image monitoring: {e}")
            return False
@@ -37,6 +37,7 @@ from typing import Optional, Dict, Any
 from PIL import Image

 from .logging_config import get_logger
+from .png_text import png_text


 class ComfyUIMetadataExtractor:
@@ -88,12 +89,14 @@ class ComfyUIMetadataExtractor:
                # Add basic file information
                metadata["file_info"] = self.get_file_info(image_path, image)

-                # Extract ComfyUI-specific metadata from PNG text chunks
-                if hasattr(image, "text") and image.text:
+                # Extract ComfyUI-specific metadata from PNG text chunks,
+                # read without decoding the pixels (#523)
+                text = png_text(image)
+                if text:
                    # Look for ComfyUI workflow data
-                    if "workflow" in image.text:
+                    if "workflow" in text:
                        try:
-                            workflow_data = json.loads(image.text["workflow"])
+                            workflow_data = json.loads(text["workflow"])
                            metadata["workflow"] = workflow_data

                            # Extract text encoder nodes from workflow
@@ -107,9 +110,9 @@ class ComfyUIMetadataExtractor:
                            self.logger.warning(f"Failed to parse workflow JSON: {e}")

                    # Look for prompt data
-                    if "prompt" in image.text:
+                    if "prompt" in text:
                        try:
-                            prompt_data = json.loads(image.text["prompt"])
+                            prompt_data = json.loads(text["prompt"])
                            metadata["prompt"] = prompt_data
                        except json.JSONDecodeError as e:
                            self.logger.warning(f"Failed to parse prompt JSON: {e}")
@@ -129,13 +132,13 @@ class ComfyUIMetadataExtractor:
                    ]

                    for field in metadata_fields:
-                        if field in image.text:
+                        if field in text:
                            try:
                                # Try to parse as JSON first
-                                metadata[field] = json.loads(image.text[field])
+                                metadata[field] = json.loads(text[field])
                            except (json.JSONDecodeError, TypeError):
                                # Store as string if not valid JSON
-                                metadata[field] = image.text[field]
+                                metadata[field] = text[field]

                return (
                    metadata
@@ -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
@@ -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):
@@ -0,0 +1,116 @@
+"""PNG text chunks without decoding pixels (#523).
+
+Pillow's ``PngImageFile.text`` calls ``load()``, which decodes every pixel, to
+reach text chunks written after the image data. On a scan or a busy image
+monitor that is a full-size buffer per file for a few kilobytes of text.
+
+``png_text`` returns the same dictionary as ``image.text`` for an opened,
+single-frame PNG, by walking the chunks after the first IDAT with Pillow's own
+chunk handlers (same decoding, limits and duplicate rule) and seeking over the
+pixel data. It follows Pillow's ``load_end`` loop: chunk checksums after the
+pixels are not checked, an unreadable chunk header ends the walk, and another
+chunk that runs past the end of the file fails like ``ImageFile._safe_read``
+(unless ``ImageFile.LOAD_TRUNCATED_IMAGES``).
+
+The one difference is pixel data, which is never decoded: pixel data that is
+undecodable, or cut short, no longer hides the text read before it. Pillow
+itself accepts a file cut inside its last pixel chunk once the decoder has
+all rows, and only a decode can tell those apart, so the walk never reads
+less than ``.text`` did (the PR #170 "legacy parity floor").
+
+Anything else (other formats, animated PNGs, an image already loaded) is
+answered by Pillow itself.
+"""
+
+import os
+import struct
+
+from PIL import ImageFile
+
+_PIXEL_CHUNKS = (b"IDAT", b"fdAT")
+# Signature, IHDR chunk and IDAT chunk header: the earliest pixel data offset.
+_FIRST_DATA_OFFSET = 8 + 25 + 8
+
+
+def png_text(image) -> dict:
+    """``dict(image.text)`` for an opened image, without a pixel decode.
+
+    The image stays usable: its file position and Pillow state are restored.
+    Non-PNG images answer ``{}`` (only PNG has text chunks in Pillow).
+    """
+    if not _walkable(image):
+        return dict(getattr(image, "text", None) or {})
+    fp = image.fp
+    resume = fp.tell()
+    try:
+        return _walk_after_pixels(image.png, fp, image.tile[0][2] - 8)
+    finally:
+        fp.seek(resume)
+
+
+def _walkable(image) -> bool:
+    """A plain single-frame PNG whose header pass stopped at its first IDAT."""
+    tile = getattr(image, "tile", None)
+    return (
+        image.format == "PNG"
+        and getattr(image, "png", None) is not None
+        and getattr(image, "fp", None) is not None
+        and not getattr(image, "is_animated", False)
+        and isinstance(tile, list)
+        and len(tile) == 1
+        and tile[0][0] == "zip"
+        # The first IDAT's data offset; anything else is not a layout we know.
+        and isinstance(tile[0][2], int)
+        and tile[0][2] >= _FIRST_DATA_OFFSET
+    )
+
+
+def _walk_after_pixels(stream, fp, offset: int) -> dict:
+    """Pillow's ``load_end`` chunk loop from the first IDAT header, skipping pixels.
+
+    ``stream`` is the image's own ``PngStream``: its text so far and its text
+    memory budget carry on exactly as a ``load()`` would carry them. The text
+    and info dictionaries are copied first, so ``image.text`` and
+    ``image.info`` are unchanged. Like ``load_end``, a handler for a chunk
+    misplaced after the pixels (a second IHDR or PLTE) may still update the
+    stream's other header fields; Pillow ignores them once the image is open.
+    """
+    end = fp.seek(0, os.SEEK_END)
+    saved = stream.im_text, stream.im_info, stream.text_memory
+    stream.im_text, stream.im_info = dict(saved[0]), dict(saved[1])
+    try:
+        fp.seek(offset)
+        while True:
+            try:
+                kind, position, length = stream.read()
+            except (struct.error, SyntaxError):
+                break
+            if kind == b"IEND":
+                break
+            try:
+                handled = kind not in _PIXEL_CHUNKS and _handle(
+                    stream, kind, position, length
+                )
+            except UnicodeDecodeError:
+                break
+            if not handled:
+                if position + length > end:
+                    if kind in _PIXEL_CHUNKS or ImageFile.LOAD_TRUNCATED_IMAGES:
+                        break
+                    raise OSError("Truncated File Read")
+                fp.seek(position + length + 4)
+                continue
+            fp.read(4)  # checksum: not verified after the pixels, as in Pillow
+        return stream.im_text
+    finally:
+        stream.im_text, stream.im_info, stream.text_memory = saved
+
+
+def _handle(stream, kind: bytes, position: int, length: int) -> bool:
+    """Run Pillow's handler; ``False`` when the chunk's bytes must be skipped."""
+    try:
+        stream.call(kind, position, length)
+    except (EOFError, AttributeError):
+        # A chunk Pillow reads past (EOFError) or has no handler for.
+        return False
+    return True
@@ -176,6 +176,12 @@ function mountWorkspace() {
    function notifyCounts() { for (const listener of countListeners) listener(); }
    /** Re-read the given counts unless a read is already in flight. */
    function freshenCounts(ids) { refreshCounts(ids.filter(id => !countRequests.has(id))); }
+    /** #555: the counts as System › Overview reads them, so its Images metric and the Images badge cannot differ. */
+    const sharedCounts = Object.freeze({
+        get: id => knownCounts.get(id),
+        subscribe(listener) { countListeners.add(listener); return () => countListeners.delete(listener); },
+        freshen: freshenCounts,
+    });
    /** A count a page has just read itself: it replaces the badge, and an older read in flight no longer lands. */
    function setCount(id, count) {
        countRequests.get(id)?.abort();
@@ -488,7 +494,7 @@ function mountWorkspace() {
        if (route.id === 'system') {
            // The concept's System (app.js renderSystem): its heading description, section nav and Save row.
            description.textContent = route.description;
-            system = mountSystem(main, api, view, { section: route.section, activity });
+            system = mountSystem(main, api, view, { section: route.section, activity, counts: sharedCounts });
            return;
        }
        if (route.id === 'inspect') {
@@ -11,8 +11,10 @@
 *   D99: its Review thumbnails opens the same "Thumbnail maintenance" dialog as Sources (thumbnail-review.mjs).
 *   #317: it reads "N missing · M outdated · K failed"; while previews are outdated (another format than the
 *   configured one) a second button, "Convert outdated", opens that dialog on its Convert outdated scope (D131).
- * - D96: the Statistics dialog lists the real runs per category, then "Library totals" (D153: no rating column or
- *   rated-images total).
+ * - #555: metric 3, "Images", is the shell's Images count (the Images page's total, its sidebar badge), not the
+ *   overview's indexed rows, which leave out files no row links: a dash until it is read or when its read failed.
+ * - D96: the Statistics dialog lists the real runs per category, then "Database records" (D153: no rating column or
+ *   rated-images total; #555: the legacy row counts, labelled as records).
 * - D95: the "Health report" lists every check the server ran, Missing files per image source, with a chip per real
 *   result and when it was checked. The check runs from Maintenance's notice ("Run health checks", D123) as a
 *   read job of the registry, tracked in the Activity store (its event stream); the report opens when it ends.
@@ -61,12 +63,20 @@ export function integrityText(health) {
    return health.status === 'passed' ? 'Passed' : 'Incomplete';
 }

-/** The concept's four metrics `[value, label]`, in its order. */
-export function metricValues(overview) {
+/** #555: the Images metric's value from the shell's Images count; a dash while unread (undefined) or failed (null). */
+export function imagesText(images) {
+    return Number.isSafeInteger(images) && images >= 0 ? number(images) : '–';
+}
+
+/**
+ * The concept's four metrics `[value, label]`, in its order. `images` is the shell's Images count (#555): the concept's
+ * `images.length`, the same number as the Images sidebar badge.
+ */
+export function metricValues(overview, images) {
    return [
        [number(overview.library.positive_prompts), 'Positive prompts'],
        [number(overview.library.negative_variants), 'Negative variants'],
-        [number(overview.images.indexed), 'Images'],
+        [imagesText(images), 'Images'],
        [integrityText(overview.health), 'Database integrity'],
    ];
 }
@@ -174,10 +184,15 @@ export function categoryRows(statistics) {
    return statistics.by_category.map(row => [row.category ?? 'Uncategorized', number(row.runs)]);
 }

-/** D96: the added "Library totals" rows `[label, count]` (the legacy statistics; D153 drops the rated images). */
+/**
+ * D96: the added "Database records" rows `[label, count]` (the legacy statistics; D153 drops the rated images). #555:
+ * they count raw rows (every prompt row, tags on a prompt, every image row), so they say so and never read as the
+ * Library or Images page counts.
+ */
 export function totalsRows(statistics) {
    const { totals } = statistics;
-    return [['Prompts', totals.total_prompts], ['Tags', totals.total_tags], ['Images', totals.total_images]]
+    return [['Prompt records', totals.total_prompts], ['Tags on prompts', totals.total_tags],
+        ['Image records', totals.total_images]]
        .map(([label, value]) => [label, number(value)]);
 }

@@ -208,11 +223,11 @@ function table(caption, head, rows, empty) {
    return node;
 }

-/** The Statistics dialog: "Runs by category" (the concept's table), then "Library totals" (D96). */
+/** The Statistics dialog: "Runs by category" (the concept's table), then "Database records" (D96, #555). */
 export function statisticsNodes(statistics) {
    return [
        table('Runs by category', ['Category', 'Runs'], categoryRows(statistics), 'No prompts in the library yet.'),
-        table('Library totals', ['Total', 'Count'], totalsRows(statistics), ''),
+        table('Database records', ['Total', 'Count'], totalsRows(statistics), ''),
    ];
 }

@@ -332,9 +347,9 @@ export function healthNotice({ api, activity, scope, navigate = navigateTo, noti
    return notice;
 }

-function metricsNode(overview) {
+function metricsNode(overview, images) {
    const metrics = el('div', undefined, 'summary-metrics');
-    for (const [value, label] of metricValues(overview)) {
+    for (const [value, label] of metricValues(overview, images)) {
        const metric = el('div', undefined, 'metric');
        metric.append(el('strong', value), el('span', label));
        metrics.append(metric);
@@ -349,7 +364,7 @@ function surface(heading, rows) {
 }

 /** The Overview's nodes, and its thumbnail row (painted by its own read). */
-function overviewNodes(overview, { api, activity, notify, scope, navigate, openStatistics, openThumbnails }) {
+function overviewNodes(overview, { api, activity, notify, scope, navigate, openStatistics, openThumbnails, images }) {
    const go = section => () => navigate(`#/system/${section}`);
    const review = operationRow({ scope, title: ambiguousTitle(overview.needs_review),
        description: Number.isSafeInteger(overview.needs_review?.pending)
@@ -374,26 +389,36 @@ function overviewNodes(overview, { api, activity, notify, scope, navigate, openS
        operationRow({ scope, title: 'One display configuration', description: 'Grid and density settings are shared across views.',
            label: 'Appearance', onClick: go('appearance') }),
        operationRow({ scope, title: sourcesTitle, description: sourcesDescription, label: 'Manage sources', onClick: go('sources') }),
-        operationRow({ scope, title: 'Statistics', description: 'Runs by category, with library totals.', label: 'Explore',
+        operationRow({ scope, title: 'Statistics', description: 'Runs by category, with database record counts.', label: 'Explore',
            onClick: openStatistics }),
    ]);
    const columns = el('div', undefined, 'two-col');
    columns.append(health, preferences);
-    return { nodes: [metricsNode(overview), columns], thumbnails, backup: backupRow };
+    const metrics = metricsNode(overview, images);
+    return { nodes: [metrics, columns], thumbnails, backup: backupRow, metrics };
 }

 /**
 * The Overview section's mount (`mountSystem`'s `sections.overview`): returns its loading line, which the server's
 * answer replaces in place (the concept's metrics and columns sit directly in `.settings-layout`). A read that a newer
 * one superseded, or one answering after the section left, paints nothing.
+ *
+ * `counts` is the shell's sidebar counts (`{get(id), subscribe(listener), freshen(ids)}`, index.mjs): the Images metric
+ * shows its `images` count and follows it, and a visit re-reads it as the workspace Overview does (#555).
 */
-export function mountOverview(container, { api, activity, scope, navigate = navigateTo, notify = showWorkspaceToast,
-    latestOnly = window.ApiClient.latestOnly, STALE = window.ApiClient.STALE }) {
+export function mountOverview(container, { api, activity, scope, counts = null, navigate = navigateTo,
+    notify = showWorkspaceToast, latestOnly = window.ApiClient.latestOnly, STALE = window.ApiClient.STALE }) {
    const overviewRead = latestOnly({ lifetime: scope.signal });
    const statisticsRead = latestOnly({ lifetime: scope.signal });
    const thumbnailsRead = latestOnly({ lifetime: scope.signal });
    const backupRead = latestOnly({ lifetime: scope.signal });
    let shown = [];
+    // #555: the Images metric's value node, repainted whenever the shell's Images count changes.
+    let imagesValue = null;
+    const imagesCount = () => counts?.get('images');
+    const paintImages = () => {
+        if (imagesValue) imagesValue.textContent = imagesText(imagesCount());
+    };
    // The backup status (undefined while read, null when it failed), painted into its row once both are there (S9).
    let backupStatus;
    let backupRow = null;
@@ -439,8 +464,10 @@ export function mountOverview(container, { api, activity, scope, navigate = navi
        try {
            const overview = await overviewRead(signal => api.get(OVERVIEW_URL, { signal, cache: 'no-store' }));
            if (overview === STALE || scope.signal.aborted) return;
-            const { nodes, thumbnails, backup } = overviewNodes(overview, { api, activity, notify, scope, navigate, openStatistics,
+            const { nodes, thumbnails, backup, metrics } = overviewNodes(overview, { api, activity, notify, scope, navigate,
+                openStatistics, images: imagesCount(),
                openThumbnails: trigger => openThumbnailReview({ api, store: activity, scope, trigger, navigate }) });
+            imagesValue = metrics.children[2].querySelector('strong');
            thumbnailRow = thumbnails;
            backupRow = backup;
            sources = overview.sources;
@@ -488,6 +515,10 @@ export function mountOverview(container, { api, activity, scope, navigate = navi
    const first = swap(loading());
    // D129: a finished Move library changes where the library lives.
    scope.cleanup(onMoveEnded(activity, load));
+    if (counts) {
+        scope.cleanup(counts.subscribe(paintImages));
+        counts.freshen(['images']);
+    }
    load();
    loadThumbnails();
    loadBackup();
@@ -135,7 +135,7 @@ export const DELIVERED_SECTIONS = Object.freeze({
 /**
 * Mount System under the heading `renderHeading` wrote in `main`. Returns `{show(section)}`, which shows another
 * section in place. `scope` is the page's view scope; `sections` maps a section id to a delivered section's mount
- * (`(container, {api, activity, scope, drafts}) => nodes`) for every section. `activity` is the page's Activity store, which follows the jobs a section starts.
+ * (`(container, {api, activity, scope, drafts, counts}) => nodes`; `counts` is the shell's sidebar counts, #555) for every section. `activity` is the page's Activity store, which follows the jobs a section starts.
 *
 * The host values are read on mount (and by the store again after a conflict), so a save carries the current revision;
 * until a read returned one, a host save fails without a request (system-state.mjs). The browser values are read on
@@ -159,7 +159,8 @@ function legacyPagesLine() {
    return line;
 }

-export function mountSystem(main, api, scope, { section, activity, sections = DELIVERED_SECTIONS, navigate = navigateTo }) {
+export function mountSystem(main, api, scope, { section, activity, counts = null, sections = DELIVERED_SECTIONS,
+    navigate = navigateTo }) {
    const store = systemDrafts(api);
    jobs = activity;
    store.reloadHost();
@@ -208,7 +209,7 @@ export function mountSystem(main, api, scope, { section, activity, sections = DE
        for (const button of nav.children) button.setAttribute('aria-pressed', String(button.dataset.value === next));
        while (layout.firstChild !== saveRow) layout.firstChild.remove();
        // Every section is delivered (S12), and the router only names known sections (state.mjs).
-        const nodes = sections[next](layout, { api, activity, scope: sectionScope, drafts: store });
+        const nodes = sections[next](layout, { api, activity, scope: sectionScope, drafts: store, counts });
        layout.prepend(...nodes);
        if (next === 'overview') main.append(legacyPages);
        else legacyPages.remove();
Author	SHA1	Message	Date
Vito Sansevero	39730c2f77	Merge remote-tracking branch 'origin/main' into perf/523-image-memory-2 # Conflicts: # TESTING.md # py/api/__init__.py # py/api/admin.py	2026-10-31 17:46:05 -07:00
Vito	3cb90b0c1e	perf(images): read PNG text without decoding pixels (#523 ) (#577 ) * perf(images): read PNG text without decoding pixels (#523) Pillow's PngImageFile.text loads every pixel to reach text chunks written after IDAT. The scan's metadata batch and the image monitor now read PNG text through utils/png_text.py, which runs Pillow's own chunk handlers after the first IDAT and seeks over the pixel data. The legacy .text read is the floor: a differential test compares both readers on every fixture image, a curated corpus and a seeded fuzz. The only difference is a file whose pixel data is undecodable or cut short: it now keeps the text read before its pixels instead of reading nothing. Animated PNGs still use .text. * test(images): cover the text budget, XMP and layout guard of the PNG text walk (#523)	2026-11-01 00:04:27 +00:00
Vito	9a43b67008	test: reset a disposed fixture package's exit actions (#580 ) (#581 ) load_package imports the package under a test-only name; its exit actions stayed registered after cleanup_package removed it from sys.modules, so the suite ended with a ModuleNotFoundError from _kill_encoders_at_exit. cleanup_package now resets the package's exit signal.	2026-10-31 23:17:17 +00:00
Vito Sansevero	1cbc32f42a	Merge remote-tracking branch 'origin/main' into perf/523-image-memory-2 # Conflicts: # TESTING.md	2026-10-31 16:09:18 -07:00
Vito Sansevero	b264dac2e8	fix(images): start no image thread after the teardown; join retired job pools (#523 )	2026-10-31 15:58:25 -07:00
Vito Sansevero	77f9f09443	perf(images): run image work on two own pools and trim after jobs (#523 ) Image work no longer spreads over the default executor's threads, where glibc kept an image-sized malloc arena per thread. - Interactive image work (view previews behind their gate, workspace card facts, both inspector routes) runs on the API's fixed pm-image pool of 3 threads; the on-demand preview pool is merged into it. - map_parallel (scan metadata, thumbnail and convert jobs, the legacy gallery generation, capture previews) shares one long-lived pm-worker pool of Worker threads, reused between jobs. The package teardown stops it within its bound (part "workers"). - After a scan, thumbnail or convert job, glibc's malloc_trim(0) returns free pages once; a no-op on Windows, macOS and musl. - tools/perf/memory_drive.py measures RSS, heap, threads and arenas under the #523 load on the disposable browser test server (manual only).	2026-10-31 15:36:07 -07:00
Vito	c3c75c3369	fix(monitor): never store PromptSearchList search text as a prompt (#548 ) (#576 ) A PromptSearchList wired straight into CLIPTextEncode left the legacy-parser fallback tracing the positive to the node's search query ("fox"). Both the output scan and the live monitor stored that query as a prompt and linked every image of the batch to it. The fallback now leaves such files unlinked (the usual unresolved outcome) when its answer is the search text of a PromptSearchList upstream of a sampler's positive. An empty search text, and a search feeding only the negative, keep their previous outcome. The parity gate states f22 as a deliberate departure from the #170 parser, and the README says to wire the Search List through Prompt Manager Text to get the images linked.	2026-10-31 22:25:01 +00:00
Vito Sansevero	d92b2eabd9	test(images): cover the text budget, XMP and layout guard of the PNG text walk (#523 )	2026-10-31 15:18:02 -07:00
Vito Sansevero	e5868d96f2	perf(images): read PNG text without decoding pixels (#523 ) Pillow's PngImageFile.text loads every pixel to reach text chunks written after IDAT. The scan's metadata batch and the image monitor now read PNG text through utils/png_text.py, which runs Pillow's own chunk handlers after the first IDAT and seeks over the pixel data. The legacy .text read is the floor: a differential test compares both readers on every fixture image, a curated corpus and a seeded fuzz. The only difference is a file whose pixel data is undecodable or cut short: it now keeps the text read before its pixels instead of reading nothing. Animated PNGs still use .text.	2026-10-31 15:01:37 -07:00
Vito	93bf24f47d	fix(monitor): keep watchdog threads from holding the exit (#561 ) (#569 ) * fix(monitor): keep watchdog threads from holding the exit (#561) watchdog 7.0.0 on Windows starts a plain threading.Thread per watch for its ReadDirectoryChangesW reader (DirectoryChangeReader.start). Created from ComfyUI's main thread it is not a daemon, and the interpreter joins every non-daemon thread before it runs any atexit callback, so the package teardown that would stop the observer never ran and the process never exited. Start the observer from a short-lived daemon thread, so every thread it creates inherits the daemon flag. Run observer.stop(), which joins the emitters without a limit, on a daemon thread and wait for it only until the stop deadline; a later stop waits for the same call. Lift the watchdog<7 cap. Closes #561 * test(exit): exempt watchdog 7's process-wide inotify reader (#561) watchdog 7 on Linux reads every inotify watch through one InotifyFD daemon thread per process that no observer stop ends, by design. The exit and package tests counted every watchdog BaseThread as the package's, so the singleton failed them once the cap was lifted. Exclude it, and require instead that no non-daemon thread is left at exit.	2026-10-31 21:31:55 +00:00
Vito	e29c23de8d	chore(init): print a one-line startup banner (#574 ) Replace the version line, one Loaded line per node and the total with "[ComfyUI-PromptManager] Version: <v> - <n> tools loaded". Closes #573	2026-10-31 18:56:42 +00:00
Vito	b82e68a9d0	ci: stop testing macOS in CI (#572 ) macOS is not a popular ComfyUI platform. Drop the macOS legs from the portability workflow and its native-trash proof; Windows keeps its Python 3.12-3.14 legs. The macOS code stays, and README calls its support best-effort. Closes #571	2026-10-31 18:13:27 +00:00
Vito	fb02066cda	test(vocabulary): open the long-list check as a new page, not a reload (#567 ) (#570 ) Firefox restores the scroll position of a reloaded page's tag window, so its end scrolled into view and appended a second page before the 100-row check. Scrolling the window to its top before the reload did not always stick on CI. A fresh navigation has no position to restore.	2026-10-31 18:13:05 +00:00
Vito	ab17818792	fix(system): show the Images page count in System Overview (#568 ) System > Overview's Images metric read the overview's indexed generated_images rows, while the workspace Overview card and the Images sidebar badge show the Images page's own total, which also lists files no row links (110 against 122 on the demo library). The concept's System metric and sidebar badge are both `images.length`, so the metric now reads the shell's Images count: one source for all three, a dash until it is read or when its read failed, re-read on each visit. The Statistics dialog's "Library totals" were the legacy /stats raw row counts under the same nouns; they now read "Database records" with "Prompt records", "Tags on prompts" and "Image records". /stats and the legacy admin page are unchanged. Closes #555	2026-10-31 18:04:45 +00:00