Pair category repair (#550) and saved instructions + dialog picker (#431) (#590)

* feat(maintenance): repair old Pair prompts' empty category and tags (#550)

Squashed from PR #579. Closes #550.

* feat(tagging): saved instructions with a default per engine (#431)

Squashed from PR #587.

* feat(tagging): pick saved instructions in the tagging dialogs (#431)

Squashed from PR #588. Closes #431.

.github/workflows/portability.yml

@@ -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:
@@ -21,19 +21,23 @@ env:
   PYTHONFAULTHANDLER: "1"
 
 # Time limit rule as in test.yml: ceil(1.5 x the slowest of three recent green runs on main) + 3
-# minutes, never under 5. Measured 2026-10-26 in test.yml on runs 40257217681 (3be0ada),
-# 40240706568 (fb9134f), 40228167328 (f9835aa), before these legs moved here in #420:
-#   slowest py3.10 on windows-latest: 12m43s / 15m54s / 14m53s -> 27 min
-#     (macos 6m14s-8m30s, windows 11m42s-15m54s)
+# minutes, never under 5, timed as whole-job wall time. One matrix job, so the slowest leg of each
+# 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)
+#   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.
 jobs:
   test:
     name: py${{ matrix.python-version }} on ${{ matrix.os }}
     runs-on: ${{ matrix.os }}
-    timeout-minutes: 27
+    timeout-minutes: 15
     strategy:
       fail-fast: false
       matrix:
-        os: [windows-latest, macos-latest]
+        os: [windows-latest]
         python-version: ["3.12", "3.13", "3.14"]
 
     steps:
@@ -53,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
@@ -75,3 +79,39 @@ jobs:
         env:
           PM_TEST_SLOWEST: "40"
         run: python -m tests.run_suite -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:
+    name: native trash on a real mount
+    runs-on: ubuntu-latest
+    timeout-minutes: 6
+
+    steps:
+      - uses: actions/checkout@v7
+
+      - name: Set up Python
+        uses: actions/setup-python@v7
+        with:
+          python-version: "3.12"
+          cache: pip
+          cache-dependency-path: requirements-dev.txt
+
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip
+          pip install -r requirements-dev.txt
+
+      - name: Mount an ext4 image
+        run: |
+          truncate -s 64M "$RUNNER_TEMP/drive.img"
+          mkfs.ext4 -q "$RUNNER_TEMP/drive.img"
+          mkdir "$RUNNER_TEMP/drive"
+          sudo mount -o loop "$RUNNER_TEMP/drive.img" "$RUNNER_TEMP/drive"
+          sudo rmdir "$RUNNER_TEMP/drive/lost+found"
+          sudo chown "$(id -u):$(id -g)" "$RUNNER_TEMP/drive"
+          findmnt "$RUNNER_TEMP/drive"
+
+      - name: Native trash proof
+        timeout-minutes: 4
+        run: PM_TRASH_MOUNT="$RUNNER_TEMP/drive" python -m unittest tests.test_trash_mount_root -v

.github/workflows/test.yml

@@ -28,8 +28,8 @@ 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 keeps this 27-minute limit; this
-#     Linux-only matrix keeps it too until it is re-measured.
+#     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
 #   perf budget 30s / 29s / 27s, black and flake8 39s / 37s / 39s,
@@ -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 }}

.github/workflows/viewerjs-check.yml

@@ -41,9 +41,13 @@ concurrency:
 # 7.3m-12.2m), because --shard splits by test count and the slow concept specs sort into shard 1. The page
 # workflows now run in three shards per browser with the limits unchanged; the per-shard times are
 # re-measured after three green main runs.
+# #563: coverage shards reached their 18-minute step limit three times on 2026-10-31 with no failing
+# test (PR #560 shard 2/3: 814 passed in 17.9m; queue run 41206495451 shard 1/3: 827 passed; PR #564
+# shard 2/3 twice), against recent green shards of 11m-17m. Coverage now runs in four shards with the
+# limits unchanged; the per-shard times are re-measured after three green main runs.
 jobs:
   frontend-coverage:
-    name: frontend coverage (shard ${{ matrix.shard }}/3)
+    name: frontend coverage (shard ${{ matrix.shard }}/4)
     runs-on: ubuntu-latest
     # Layout comparisons depend on the fonts each browser falls back to, so browser jobs run in the
     # same pinned Playwright image as local runs (tag = @playwright/test in tools/vendor/package-lock.json;
@@ -59,7 +63,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        shard: [1, 2, 3]
+        shard: [1, 2, 3, 4]
     steps:
       - uses: actions/checkout@v7
       # The image's own Python 3.12 and Node 24: setup-python/setup-node install the runner's toolcache
@@ -88,13 +92,13 @@ jobs:
       - run: npm ci --prefix tools/vendor --ignore-scripts
       # Shard 1 also runs the node units and the viewer suite, so the vendor pins and the tooling tests
       # run on the lightest shard instead.
-      - if: matrix.shard == 3
+      - if: matrix.shard == 4
         run: npm run --prefix tools/vendor check
-      - if: matrix.shard == 3
+      - if: matrix.shard == 4
         run: npm run --prefix tools/vendor test
-      - name: Page workflows with coverage, shard ${{ matrix.shard }}/3
+      - name: Page workflows with coverage, shard ${{ matrix.shard }}/4
         timeout-minutes: 18
-        run: npm run --prefix tools/vendor test:coverage:shard -- ${{ matrix.shard }}/3
+        run: npm run --prefix tools/vendor test:coverage:shard -- ${{ matrix.shard }}/4
       - name: Upload raw coverage
         uses: actions/upload-artifact@v7
         with:
@@ -136,13 +140,13 @@ jobs:
         with:
           pattern: coverage-raw-*
           path: ${{ runner.temp }}/coverage-raw
-      # coverage:merge only checks that raw files exist; merging two of three shards would report
+      # coverage:merge only checks that raw files exist; merging three of four shards would report
       # partial coverage, so a missing or empty shard artifact fails here.
       - name: Verify every shard's raw coverage arrived
         shell: bash
         env:
           STAGED_RAW: ${{ runner.temp }}/coverage-raw
-          EXPECTED_SHARDS: 3
+          EXPECTED_SHARDS: 4
         run: |
           found=0
           for shard in $(seq 1 "$EXPECTED_SHARDS"); do

__init__.py

@@ -120,6 +120,7 @@ except Exception as e:
         logger = get_logger("prompt_manager.init")
         logger.error(f"Failed to register API routes: {e}")
     except Exception:
+        # logging failed too; a report must never stop the node loading
         pass
 
 # Shared database, set up with the monitoring system below. Stays None when that
@@ -279,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.
@@ -297,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}
@@ -344,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():
@@ -411,6 +425,7 @@ except Exception as e:
             f"Failed to register the exit signal: {e}"
         )
     except Exception:
+        # logging failed too; a report must never stop the node loading
         pass
 
 try:
@@ -428,16 +443,14 @@ except Exception as e:
             f"Failed to register the background work teardown: {e}"
         )
     except Exception:
+        # logging failed too; a report must never stop the node loading
         pass
 
 __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"
+)

database/image_removal.py

@@ -43,6 +43,7 @@ from .image_categories import DELETE_SQL as DELETE_CATEGORY_SQL
 from .image_removal_schema import (
     META_TABLE,
     TABLE,
+    bump_generation,
     clear_marks,
     has_marks_table,
     mark_of,
@@ -144,6 +145,23 @@ def removed_from_library(conn, path) -> bool:
     return False
 
 
+def mark_hides(marks: dict, locator: str, path) -> bool:
+    """Whether a removal mark in *marks* hides the file at *path* (blocking).
+
+    *marks* is ``{locator: (size, mtime_ns)}``. The mark of *locator* hides
+    the file while ``os.lstat(path)`` still has the size and ``mtime_ns`` it
+    was removed with; a file that cannot be looked at is not hidden.
+    """
+    mark = marks.get(locator)
+    if mark is None:
+        return False
+    try:
+        info = os.lstat(path)
+    except OSError:
+        return False
+    return (info.st_size, info.st_mtime_ns) == tuple(mark)
+
+
 def write_mark(conn, locator, size, mtime_ns):
     _write_mark(conn, locator, size, mtime_ns, utc_now_iso())
 
@@ -302,4 +320,29 @@ class ImageRemovalOperations:
         if not locators:
             return 0
         with self._pair_transaction() as conn:
-            return clear_marks(conn, locators)
+            removed = clear_marks(conn, locators)
+        # #498: again after the commit, so a reader that cached the marks
+        # between the in-transaction bump and the commit reloads them.
+        bump_generation()
+        return removed
+
+    def clear_hiding_marks(self, files) -> set:
+        """Delete the marks that still hide their file (#498, Return to library).
+
+        *files* are ``(locator, path)`` pairs. In one ``BEGIN IMMEDIATE``
+        transaction each mark is read again and deleted only while the file at
+        *path* still has its size and ``mtime_ns``. Returns the locators whose
+        marks went.
+        """
+        cleared = set()
+        files = list(files)
+        if not files:
+            return cleared
+        with self._pair_transaction() as conn:
+            for locator, path in files:
+                mark = mark_of(conn, locator)
+                if mark is not None and mark_hides({locator: mark}, locator, path):
+                    cleared.add(locator)
+            clear_marks(conn, sorted(cleared))
+        bump_generation()
+        return cleared

database/image_removal_schema.py

@@ -155,6 +155,11 @@ def _bump() -> None:
     _generation += 1
 
 
+def bump_generation() -> None:
+    """Move the generation after a commit (a cached read made before it reloads)."""
+    _bump()
+
+
 def has_marks_table(conn: sqlite3.Connection) -> bool:
     return (
         conn.execute(

database/library.py

@@ -7,6 +7,7 @@ from contextlib import contextmanager
 
 from .library_folders import folder_condition, list_library_folders
 from .library_tokens import library_edit_token
+from .prompt_archive import ARCHIVED, SHELVED
 
 _ELIGIBLE = """(EXISTS(SELECT 1 FROM prompt_pairs ep WHERE ep.positive_prompt_id=p.id)
  OR EXISTS(SELECT 1 FROM generated_images eg WHERE eg.prompt_id=p.id)
@@ -106,6 +107,12 @@ ARCHIVE_SCOPES = {
 }
 
 
+# Shelves (#443): the parents on the shelf, or the archived ones.
+PARENT_SCOPES = {"active": SHELVED, "archived": ARCHIVED}
+# A positive pair owner that is on the shelf, as a condition on alias pp.
+SHELVED_PAIR = SHELVED.replace("p.id", "pp.positive_prompt_id")
+
+
 def _escape(value: str) -> str:
     return value.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
 
@@ -270,6 +277,8 @@ class LibraryOperations:
         rows = conn.execute(
             "SELECT p.id,p.text,ls.title,p.category,p.rating,p.notes,p.created_at,"
             "p.run_count,p.last_used_at,"
+            + ARCHIVED
+            + " AS archived,"
             + (f"{sort_expression} AS _sort," if sort_expression else "")
             + "(SELECT COUNT(*) FROM prompt_pairs v WHERE v.positive_prompt_id=p.id "
             "AND v.archived=0) AS variant_count,"
@@ -293,6 +302,7 @@ class LibraryOperations:
             # Usage is shown, never compared: the token hashes owner fields only.
             row["run_count"] = row["run_count"] if type(row["run_count"]) is int else 0
             row["last_used_at"] = utc_iso(row["last_used_at"])
+            row["archived"] = bool(row["archived"])
             row["edit_token"] = library_edit_token(epoch, row)
         return result
 
@@ -310,10 +320,13 @@ class LibraryOperations:
         sort: str = "newest",
         limit: int = 40,
         after: dict | None = None,
+        shelf: str = "active",
     ) -> dict:
         _limit(limit)
         if sort not in _ORDERS:
             raise ValueError("Unknown library sort order")
+        if not isinstance(shelf, str) or shelf not in PARENT_SCOPES:
+            raise ValueError("shelf must be active or archived")
         where, params = _filter(
             text, category, [] if tags is None else tags, tag_mode, folder_prefixes
         )
@@ -328,15 +341,18 @@ class LibraryOperations:
             ceiling = after["ceiling"]
         where += " AND p.id<=?"
         params.append(ceiling)
-        # One SELECT counts the filtered parents and their active variants.
-        total, variant_total = conn.execute(
-            "SELECT COUNT(*),COALESCE(SUM((SELECT COUNT(*) FROM prompt_pairs v "
-            "WHERE v.positive_prompt_id=p.id AND v.archived=0)),0) FROM "
-            + _FROM
-            + " WHERE "
-            + where,
+        # One SELECT counts the filtered parents of this scope, their active
+        # variants, and the filtered archived parents (Show archived, #443).
+        scope = PARENT_SCOPES[shelf]
+        total, variant_total, archived_total = conn.execute(
+            f"SELECT COALESCE(SUM({scope}),0),"
+            f"COALESCE(SUM(CASE WHEN {scope} THEN (SELECT COUNT(*) "
+            "FROM prompt_pairs v WHERE v.positive_prompt_id=p.id "
+            "AND v.archived=0) ELSE 0 END),0),"
+            f"COALESCE(SUM({ARCHIVED}),0) FROM " + _FROM + " WHERE " + where,
             params,
         ).fetchone()
+        where += f" AND {scope}"
         expression, direction = _ORDERS[sort]
         if after is not None:
             operator = ">" if direction == "ASC" else "<"
@@ -369,6 +385,7 @@ class LibraryOperations:
             "items": rows,
             "total": total,
             "variant_total": variant_total,
+            "archived_total": archived_total,
             # Echoed so a client can tell this read applied its tag match.
             "tag_mode": tag_mode,
             "has_more": has_more,
@@ -386,6 +403,7 @@ class LibraryOperations:
         sort: str = "newest",
         limit: int = 40,
         after: dict | None = None,
+        shelf: str = "active",
     ) -> dict:
         with self.library_snapshot(include_ceiling=True) as (conn, epoch, ceiling):
             return self._list_library(
@@ -400,6 +418,7 @@ class LibraryOperations:
                 sort=sort,
                 limit=limit,
                 after=after,
+                shelf=shelf,
             )
 
     def _is_library_positive(self, conn: sqlite3.Connection, positive_id: int) -> bool:
@@ -644,7 +663,9 @@ class LibraryOperations:
             owners = {}
             order = []
             for row in conn.execute(
-                "SELECT p.id,p.text,ls.title,p.category,p.rating,p.notes FROM "
+                "SELECT p.id,p.text,ls.title,p.category,p.rating,p.notes,"
+                + ARCHIVED
+                + " AS archived FROM "
                 + _FROM
                 + " WHERE "
                 + _ELIGIBLE
@@ -657,6 +678,7 @@ class LibraryOperations:
                     "tags": [],
                     "rating": row["rating"],
                     "notes": row["notes"],
+                    "archived": bool(row["archived"]),
                     "variants": [],
                 }
                 order.append(row["id"])
@@ -701,8 +723,9 @@ class LibraryOperations:
         """System › Overview counts from one snapshot, writing nothing.
 
         ``positive_prompts`` and ``negative_variants`` are the unfiltered
-        Library's ``total`` and ``variant_total``: eligible parents and active
-        pairs (every pair's positive is eligible). ``images`` counts indexed
+        Library's ``total`` and ``variant_total``: eligible parents on the
+        shelf (not archived, #443) and their active pairs (every pair's
+        positive is eligible). ``images`` counts indexed
         ``generated_images`` rows, not files on disk. An image has no rating
         of its own (it shows its linked prompt's), so ``rated_images`` counts
         the indexed images whose linked prompt is rated. ``review_pending`` is
@@ -710,8 +733,14 @@ class LibraryOperations:
         """
         with self.library_snapshot() as (conn, _epoch, _ceiling):
             row = conn.execute(
-                "SELECT (SELECT COUNT(*) FROM prompts p WHERE " + _ELIGIBLE + "),"
-                "(SELECT COUNT(*) FROM prompt_pairs WHERE archived=0),"
+                "SELECT (SELECT COUNT(*) FROM prompts p WHERE "
+                + _ELIGIBLE
+                + " AND "
+                + SHELVED
+                + "),"
+                "(SELECT COUNT(*) FROM prompt_pairs pp WHERE pp.archived=0 AND "
+                + SHELVED_PAIR
+                + "),"
                 "(SELECT COUNT(*) FROM generated_images),"
                 "(SELECT COUNT(*) FROM generated_images gi JOIN prompts r "
                 "ON r.id=gi.prompt_id WHERE r.rating IS NOT NULL),"

database/library_archive.py

@@ -0,0 +1,473 @@
+"""Delete selected Library prompts; archive the paired ones (#443, D152).
+
+A selected Library parent is classified, in this order, as ``not_found``
+(missing, or not a Library parent), ``protected`` (tagged ``__protected__``;
+never changed), ``already_archived``, ``archive`` (the positive or negative of
+any saved pair, archived pair or not: archiving hides it from the shelf and
+removes nothing) or ``delete`` (in no pair: deleted exactly as the legacy
+delete does, through ``delete_prompt_rows``). Every command checks the
+storage epoch first and runs in one ``BEGIN IMMEDIATE`` transaction.
+
+The write executes what the user reviewed: it is given the reviewed archive
+and delete lists, and deletes only reviewed deletes the route backed up
+(``candidates``) whose fingerprint is unchanged since that backup.
+"""
+
+import hashlib
+import json
+import re
+import sqlite3
+from typing import Iterable
+
+from .library import _ELIGIBLE
+from .library_commands import _validate_bulk_ids, _validate_epoch
+from .models import utc_now_iso
+from .prompt_archive import ARCHIVED as _ARCHIVED
+from .prompt_archive import TABLE as ARCHIVE
+from .workflow_documents import documents_of_rows, release_documents
+
+try:
+    from ..utils.hashing import generate_prompt_hash
+except ImportError:
+    from utils.hashing import generate_prompt_hash
+
+# Ids bound per statement: the pair check binds each id twice, which stays
+# under SQLite's historical 999-variable limit (maintenance.py ID_BATCH).
+ID_BATCH = 400
+_PROTECTED = (
+    "EXISTS(SELECT 1 FROM prompt_tags pt JOIN tags t ON t.id=pt.tag_id "
+    "WHERE pt.prompt_id=p.id AND t.name='__protected__')"
+)
+_PAIRED = (
+    "EXISTS(SELECT 1 FROM prompt_pairs pp "
+    "WHERE pp.positive_prompt_id=p.id OR pp.negative_prompt_id=p.id)"
+)
+# #482: used as the negative of another positive's pair (archived or not):
+# never deleted, so that positive's variant stays (archive only).
+_NEGATIVE_OF_OTHER = (
+    "EXISTS(SELECT 1 FROM prompt_pairs op "
+    "WHERE op.negative_prompt_id=p.id AND op.positive_prompt_id!=p.id)"
+)
+# Classes in the order a prompt is tested; the first that holds wins.
+CLASSES = (
+    "not_found",
+    "protected",
+    "already_archived",
+    "not_archived",
+    "archive",
+    "delete",
+)
+# #492: the shelf a Delete… runs on. On the archived shelf an archived prompt is
+# classified by its pairs, and one restored since is ``not_archived``.
+SHELVES = ("active", "archived")
+
+
+def _check_shelf(shelf: str) -> bool:
+    """Whether ``shelf`` is the archived shelf (ValueError when unknown)."""
+    if shelf not in SHELVES:
+        raise ValueError("shelf must be active or archived")
+    return shelf == "archived"
+
+
+_DIGEST = re.compile(r"^[0-9a-f]{64}$")
+
+
+def _batches(ids: list) -> Iterable[list]:
+    for start in range(0, len(ids), ID_BATCH):
+        yield ids[start : start + ID_BATCH]
+
+
+def delete_prompt_rows(conn: sqlite3.Connection, prompt_ids: list[int]) -> int:
+    """Delete prompts and their image link rows; return the prompts deleted.
+
+    The legacy delete's statements (``delete_prompt``, ``bulk_delete_prompts``)
+    in the caller's transaction: image link rows first (``image_prompt_pairs``
+    cascades, review decisions keep a NULL image), then the prompts (tags and
+    Library state cascade), then the workflow documents no row references.
+    Image files are never touched. The caller checks pairs first.
+    """
+    documents = documents_of_rows(conn, "prompt_id", prompt_ids)
+    deleted = 0
+    for chunk in _batches(list(prompt_ids)):
+        marks = ",".join("?" for _ in chunk)
+        conn.execute(
+            f"DELETE FROM generated_images WHERE prompt_id IN ({marks})",  # nosec
+            chunk,
+        )
+        deleted += conn.execute(
+            f"DELETE FROM prompts WHERE id IN ({marks})", chunk  # nosec B608
+        ).rowcount
+    release_documents(conn, documents)
+    return deleted
+
+
+class LibraryArchiveOperations:
+    """PromptDatabase mixin: classify, archive or delete, and restore parents."""
+
+    def _classify_library_delete(
+        self, conn: sqlite3.Connection, prompt_ids: list[int], shelf: str = "active"
+    ) -> dict:
+        on_archived_shelf = _check_shelf(shelf)
+        found, deletable = {}, set()
+        for chunk in _batches(prompt_ids):
+            marks = ",".join("?" for _ in chunk)
+            for row in conn.execute(
+                f"SELECT p.id,{_PROTECTED},{_ARCHIVED},{_PAIRED},"  # nosec B608
+                f"{_NEGATIVE_OF_OTHER} "
+                f"FROM prompts p WHERE p.id IN ({marks}) AND {_ELIGIBLE}",
+                chunk,
+            ):
+                protected, archived, paired, negative_of_other = row[1:5]
+                if protected:
+                    found[row[0]] = "protected"
+                elif on_archived_shelf and not archived:
+                    found[row[0]] = "not_archived"
+                elif archived and not on_archived_shelf:
+                    found[row[0]] = "already_archived"
+                elif paired:
+                    found[row[0]] = "archive"
+                    if not negative_of_other:
+                        deletable.add(row[0])
+                else:
+                    found[row[0]] = "delete"
+        classes = {name: [] for name in CLASSES}
+        for pid in prompt_ids:
+            classes[found.get(pid, "not_found")].append(pid)
+        # #482: the archive ids Delete permanently may delete with their pairs.
+        classes["deletable"] = [pid for pid in classes["archive"] if pid in deletable]
+        return classes
+
+    @staticmethod
+    def _image_links(conn: sqlite3.Connection, prompt_ids: list[int]) -> int:
+        total = 0
+        for chunk in _batches(prompt_ids):
+            marks = ",".join("?" for _ in chunk)
+            total += conn.execute(
+                "SELECT COUNT(*) FROM generated_images "  # nosec B608
+                f"WHERE prompt_id IN ({marks})",
+                chunk,
+            ).fetchone()[0]
+        return total
+
+    @staticmethod
+    def _fingerprints(conn: sqlite3.Connection, prompt_ids: list[int]) -> dict:
+        """What a safety backup holds of each prompt a delete would remove.
+
+        The prompt row (usage included), its tag ids, its Library title and
+        its image link rows (count and newest id), as a JSON-safe list.
+        """
+        prints = {}
+        for chunk in _batches(prompt_ids):
+            marks = ",".join("?" for _ in chunk)
+            for row in conn.execute(
+                "SELECT p.id,p.text,p.hash,p.category,p.rating,p.notes,"  # nosec
+                "p.updated_at,p.run_count,p.last_used_at,"
+                "(SELECT group_concat(tag_id) FROM (SELECT tag_id FROM prompt_tags "
+                "WHERE prompt_id=p.id ORDER BY tag_id)),"
+                "(SELECT title FROM prompt_library_state WHERE prompt_id=p.id),"
+                "(SELECT COUNT(*) FROM generated_images WHERE prompt_id=p.id),"
+                "(SELECT MAX(id) FROM generated_images WHERE prompt_id=p.id) "
+                f"FROM prompts p WHERE p.id IN ({marks})",
+                chunk,
+            ):
+                prints[row[0]] = [
+                    value if isinstance(value, (int, str, type(None))) else repr(value)
+                    for value in tuple(row)[1:]
+                ]
+        return prints
+
+    @staticmethod
+    def _distinct_images(conn: sqlite3.Connection, prompt_ids: list[int]) -> int:
+        """Image rows of ``prompt_ids`` (each image once; #482 deletes them)."""
+        return LibraryArchiveOperations._image_links(conn, prompt_ids)
+
+    @staticmethod
+    def _pair_prints(conn: sqlite3.Connection, prompt_ids: list[int]) -> dict:
+        """#482: each prompt's own saved pairs, as the review saw them.
+
+        Per pair (by id): its negative, archived flag, metadata, usage, ordered
+        tag ids and its image-pair links (count and newest id).
+        """
+        prints = {pid: [] for pid in prompt_ids}
+        for chunk in _batches(prompt_ids):
+            marks = ",".join("?" for _ in chunk)
+            for row in conn.execute(
+                "SELECT pp.positive_prompt_id,pp.id,pp.negative_prompt_id,"  # nosec
+                "pp.archived,pp.title,pp.category,pp.rating,pp.notes,pp.run_count,"
+                "pp.last_used_at,pp.updated_at,"
+                "(SELECT group_concat(tag_id) FROM (SELECT tag_id FROM "
+                "prompt_pair_tags WHERE pair_id=pp.id ORDER BY tag_id)),"
+                "(SELECT COUNT(*) FROM image_prompt_pairs WHERE pair_id=pp.id),"
+                "(SELECT MAX(id) FROM image_prompt_pairs WHERE pair_id=pp.id) "
+                f"FROM prompt_pairs pp WHERE pp.positive_prompt_id IN ({marks}) "
+                "ORDER BY pp.id",
+                chunk,
+            ):
+                prints[row[0]].append(
+                    [
+                        v if isinstance(v, (int, str, type(None))) else repr(v)
+                        for v in tuple(row)[1:]
+                    ]
+                )
+        return prints
+
+    @classmethod
+    def _review_digests(cls, conn: sqlite3.Connection, prompt_ids: list[int]) -> dict:
+        """#482: one digest per deletable prompt of its row and its pairs."""
+        prompts = cls._fingerprints(conn, prompt_ids)
+        pairs = cls._pair_prints(conn, prompt_ids)
+        return {
+            pid: hashlib.sha256(
+                json.dumps(
+                    [prompts.get(pid), pairs[pid]], separators=(",", ":")
+                ).encode("utf-8")
+            ).hexdigest()
+            for pid in prompt_ids
+        }
+
+    @staticmethod
+    def _paired_counts(
+        conn: sqlite3.Connection, prompt_ids: list[int], selected: list[int]
+    ) -> dict:
+        """#482: what Delete permanently would remove with ``prompt_ids``."""
+        pairs = archived = variants = 0
+        negatives = set()
+        for chunk in _batches(prompt_ids):
+            marks = ",".join("?" for _ in chunk)
+            for archived_flag, negative, positive in conn.execute(
+                "SELECT archived,negative_prompt_id,positive_prompt_id "  # nosec
+                f"FROM prompt_pairs WHERE positive_prompt_id IN ({marks})",
+                chunk,
+            ):
+                pairs += 1
+                archived += 1 if archived_flag else 0
+                if negative is not None:
+                    variants += 1
+                    if negative != positive:
+                        negatives.add(negative)
+        return {
+            "pairs": pairs,
+            "archived_pairs": archived,
+            "variants": variants,
+            "kept_negatives": len(negatives - set(selected)),
+        }
+
+    @staticmethod
+    def _checked_ids(prompt_ids: list[int], expected_epoch: str) -> list[int]:
+        _validate_epoch(expected_epoch)
+        _validate_bulk_ids(prompt_ids)
+        return list(dict.fromkeys(prompt_ids))
+
+    def classify_library_delete(
+        self, prompt_ids: list[int], *, expected_epoch: str, shelf: str = "active"
+    ) -> dict:
+        """What a delete of ``prompt_ids`` would do now (reads only).
+
+        Returns the id lists per class, in request order, ``image_links``
+        (the image link rows the ``delete`` class would remove) and
+        ``fingerprints`` of the ``delete`` class (``_fingerprints``). On the
+        archived ``shelf`` (#492) archived prompts are classified by their pairs.
+        """
+        _check_shelf(shelf)
+        prompt_ids = self._checked_ids(prompt_ids, expected_epoch)
+        with self.library_snapshot() as (conn, _epoch, _ceiling):
+            epoch = self._library_epoch(conn, expected_epoch)
+            classes = self._classify_library_delete(conn, prompt_ids, shelf)
+            deletable = classes["deletable"]
+            return {
+                "library_epoch": epoch,
+                **classes,
+                "image_links": self._image_links(conn, classes["delete"]),
+                "fingerprints": self._fingerprints(conn, classes["delete"]),
+                # #482: Delete permanently, with the deletable prompts' pairs.
+                "review_digests": self._review_digests(conn, deletable),
+                "paired_counts": self._paired_counts(conn, deletable, prompt_ids),
+                "paired_image_links": self._distinct_images(conn, deletable),
+            }
+
+    @staticmethod
+    def _repair_hash(conn: sqlite3.Connection, prompt_id: int) -> None:
+        """Give an archived prompt its canonical hash, so ingestion finds it.
+
+        Scans, the monitor and node saves resolve prompts by hash; a stale or
+        missing one would make them insert a visible copy (Q3). When another
+        row already holds the canonical hash, lookups find that row and no new
+        copy is made, so nothing is written.
+        """
+        text, stored = conn.execute(
+            "SELECT text,hash FROM prompts WHERE id=?", (prompt_id,)
+        ).fetchone()
+        canonical = generate_prompt_hash(text) if isinstance(text, str) else None
+        if canonical is None or stored == canonical:
+            return
+        if conn.execute(
+            "SELECT 1 FROM prompts WHERE hash=? AND id!=?", (canonical, prompt_id)
+        ).fetchone():
+            return
+        conn.execute("UPDATE prompts SET hash=? WHERE id=?", (canonical, prompt_id))
+
+    def archive_or_delete_library_prompts(
+        self,
+        archive_ids: list[int],
+        delete_ids: list[int],
+        *,
+        expected_epoch: str,
+        candidates: dict | None,
+        delete_paired: dict | None = None,
+        shelf: str = "active",
+    ) -> dict:
+        """Archive and delete what the user reviewed, all or nothing.
+
+        ``archive_ids`` and ``delete_ids`` are the reviewed classes.
+        ``candidates`` maps each reviewed delete the route backed up to its
+        fingerprint then (None: every reviewed delete, unchecked; tests and
+        callers without a backup). A reviewed delete is deleted only when it
+        still classifies ``delete`` and its fingerprint is unchanged; one that
+        is now paired, and a reviewed archive now unpaired, are archived (the
+        restorable direction); anything else is kept and counted.
+
+        ``delete_paired`` (#482) maps reviewed archive ids the user chose to
+        Delete permanently to their review digest. Each is deleted with its
+        own saved pairs only while it is still deletable and its digest is
+        unchanged; otherwise it is archived and counted ``changed``.
+
+        On the archived ``shelf`` (#492) the ids are archived already: nothing
+        is archived or rewritten, every reviewed id not deleted (protected or
+        changed) counts ``changed``, and one restored since ``not_archived``.
+        """
+        on_archived_shelf = _check_shelf(shelf)
+        _validate_epoch(expected_epoch)
+        for ids in (archive_ids, delete_ids):
+            if not isinstance(ids, list) or any(
+                type(pid) is not int or not 1 <= pid <= 2**63 - 1 for pid in ids
+            ):
+                raise ValueError("prompt ids must be lists of positive integers")
+        reviewed_delete = set(delete_ids)
+        delete_paired = {} if delete_paired is None else delete_paired
+        if not isinstance(delete_paired, dict) or any(
+            type(pid) is not int
+            or not isinstance(digest, str)
+            or not _DIGEST.match(digest)
+            for pid, digest in delete_paired.items()
+        ):
+            raise ValueError("delete_paired must map prompt ids to review digests")
+        if not set(delete_paired) <= set(archive_ids):
+            raise ValueError("A permanent delete must be a reviewed archive id")
+        if not archive_ids and not delete_ids:
+            raise ValueError("Nothing to delete or archive")
+        if reviewed_delete & set(archive_ids):
+            raise ValueError("A prompt cannot be both archived and deleted")
+        prompt_ids = list(dict.fromkeys([*archive_ids, *delete_ids]))
+        with self._pair_transaction() as conn:
+            epoch = self._library_epoch(conn, expected_epoch)
+            classes = self._classify_library_delete(conn, prompt_ids, shelf)
+            now_prints = self._fingerprints(conn, classes["delete"])
+            deletes, archives, changed = [], [], 0
+            paired_now = set(classes["deletable"])
+            chosen = [pid for pid in classes["archive"] if pid in delete_paired]
+            digests = self._review_digests(
+                conn, [pid for pid in chosen if pid in paired_now]
+            )
+            paired_deletes = []
+            for pid in classes["archive"]:
+                if pid not in delete_paired:
+                    archives.append(pid)
+                elif digests.get(pid) == delete_paired[pid]:
+                    paired_deletes.append(pid)
+                else:
+                    archives.append(pid)  # changed since the review: restorable
+                    changed += 1
+            for pid in classes["delete"]:
+                if pid not in reviewed_delete:
+                    archives.append(pid)  # reviewed as archive, now unpaired
+                    changed += 1 if pid in delete_paired else 0
+                elif candidates is None or candidates.get(pid) == now_prints[pid]:
+                    deletes.append(pid)
+                else:
+                    changed += 1
+            changed += len(classes["protected"])
+            if on_archived_shelf:
+                # Already archived: the rest stay as they are, each one counted.
+                archives = []
+                changed = len(prompt_ids) - len(deletes) - len(paired_deletes)
+                changed -= len(classes["not_found"]) + len(classes["not_archived"])
+            image_links = self._image_links(conn, [*deletes, *paired_deletes])
+            now = utc_now_iso()
+            for pid in archives:
+                self._repair_hash(conn, pid)
+            conn.executemany(
+                f"INSERT INTO {ARCHIVE}(prompt_id, archived_at) VALUES (?, ?)",
+                [(pid, now) for pid in archives],
+            )
+            # A changed row is the newest edit the restore fingerprint reads.
+            conn.executemany(
+                "UPDATE prompts SET updated_at=? WHERE id=?",
+                [(now, pid) for pid in archives],
+            )
+            # #482: the image-pair links (RESTRICT) go first, then the pairs
+            # (pair tags CASCADE, review decisions SET NULL), then the prompts.
+            pairs_removed = 0
+            for chunk in _batches(paired_deletes):
+                marks = ",".join("?" for _ in chunk)
+                pairs_of = (
+                    "SELECT id FROM prompt_pairs "  # nosec B608
+                    f"WHERE positive_prompt_id IN ({marks})"
+                )
+                conn.execute(
+                    f"DELETE FROM image_prompt_pairs WHERE pair_id IN ({pairs_of})",
+                    chunk,
+                )
+                pairs_removed += conn.execute(
+                    "DELETE FROM prompt_pairs "  # nosec B608
+                    f"WHERE positive_prompt_id IN ({marks})",
+                    chunk,
+                ).rowcount
+            deleted = delete_prompt_rows(conn, [*deletes, *paired_deletes])
+            return {
+                "library_epoch": epoch,
+                "deleted": deleted,
+                "archived": len(archives),
+                "already_archived": len(classes["already_archived"]),
+                "not_archived": len(classes["not_archived"]),
+                "not_found": len(classes["not_found"]),
+                "changed": changed,
+                "image_links": image_links,
+                "pairs_removed": pairs_removed,
+            }
+
+    def restore_library_prompts(
+        self, prompt_ids: list[int], *, expected_epoch: str
+    ) -> dict:
+        """Put archived prompts back on the shelf, all or nothing.
+
+        Every requested id with an archive row is restored, whatever its
+        protection or Library parent state: restoring only un-hides it.
+        """
+        prompt_ids = self._checked_ids(prompt_ids, expected_epoch)
+        with self._pair_transaction() as conn:
+            epoch = self._library_epoch(conn, expected_epoch)
+            restored, existing = [], set()
+            for chunk in _batches(prompt_ids):
+                marks = ",".join("?" for _ in chunk)
+                for pid, archived in conn.execute(
+                    f"SELECT p.id,{_ARCHIVED} FROM prompts p "  # nosec B608
+                    f"WHERE p.id IN ({marks})",
+                    chunk,
+                ):
+                    existing.add(pid)
+                    if archived:
+                        restored.append(pid)
+            now = utc_now_iso()
+            conn.executemany(
+                f"DELETE FROM {ARCHIVE} WHERE prompt_id=?", [(pid,) for pid in restored]
+            )
+            conn.executemany(
+                "UPDATE prompts SET updated_at=? WHERE id=?",
+                [(now, pid) for pid in restored],
+            )
+            return {
+                "library_epoch": epoch,
+                "restored": len(restored),
+                "unchanged": len(existing) - len(restored),
+                "not_found": len(prompt_ids) - len(existing),
+            }

database/library_move.py

@@ -590,6 +590,7 @@ class _Move:
             try:
                 os.rmdir(self.plan.target_dir)
             except OSError:
+                # not empty or already gone: the undo leaves it alone
                 pass
 
 

database/maintenance.py

@@ -2,7 +2,8 @@
 
 Four operations, each the native owner of a legacy ``/maintenance`` one and
 built on the same queries, so a preview counts what the legacy cleanup would
-change:
+change (``workflow_storage`` and #550's ``pair_metadata``, which fills old
+Pair prompts' empty category and tags, live in their own modules):
 
 - ``duplicate_prompts``: prompts whose text is equal ignoring case and outer
   spaces (``DUPLICATE_PROMPTS_SQL``, legacy ``cleanup_duplicates``); each
@@ -15,6 +16,8 @@ change:
   ``cleanup_orphaned_images``), counted only under a source that can be read
   now, as the health check's ``missing_files`` judges them
   (``health.link_states``). It deletes database rows only, never a file.
+  It also forgets the removal marks (#480) that hide nothing any more: the
+  file is gone or changed, under a readable source (#501, ``_stale_marks``).
 
 ``preview_maintenance`` reads one snapshot on its own ``mode=ro`` connection
 (``health.readonly_uri``) and closes it in ``finally``; files are looked at
@@ -41,6 +44,8 @@ from pathlib import Path
 
 from .execution_contexts import LIVE_HINT_CTES
 from .health import BATCH_ROWS, link_states, readable_roots, readonly_uri
+from .image_removal_schema import TABLE as MARKS_TABLE
+from .image_removal_schema import clear_marks, has_marks_table
 from .pairs import PairDependencyError
 from .workflow_documents import documents_of_rows, release_documents
 
@@ -56,6 +61,9 @@ MAINTENANCE_OPERATIONS = (
     "missing_links",
     # Applied as its own batched job (database/workflow_storage.py).
     "workflow_storage",
+    # #550: fill old Pair prompts' empty category and tags
+    # (database/pair_metadata_repair.py).
+    "pair_metadata",
 )
 MAX_DETAILS = 50  # detail rows a plan lists
 MAX_RESULT_ROWS = 200  # changed records an apply describes (the job lists 200)
@@ -85,6 +93,12 @@ PROTECTIONS = (
         "with Library state (a title or an explicit parent)",
         "with Library state (a title or an explicit parent)",
     ),
+    (
+        # #443: an archived prompt is restorable, never pruned.
+        "EXISTS (SELECT 1 FROM prompt_archive pa WHERE pa.prompt_id = p.id)",
+        "archived in the Library",
+        "archived in the Library",
+    ),
     (
         "p.id IN (SELECT prompt_id FROM live_hint_prompts)",
         "used by recorded runs",
@@ -93,7 +107,8 @@ PROTECTIONS = (
 )
 
 # Exempt references: images, pairs, the __protected__ tag, Library parent
-# state, and live recorded-run hints (computed once, see LIVE_HINT_CTES).
+# state, the Library archive, and live recorded-run hints (computed once,
+# see LIVE_HINT_CTES).
 ORPHANED_PROMPTS_SQL = (
     "WITH "
     + LIVE_HINT_CTES
@@ -187,6 +202,8 @@ class MaintenanceDraft:
     protected: dict
     media: dict
     details: list = field(default_factory=list)
+    # missing_links only: locators of removal marks whose file is gone (#501).
+    marks: list = field(default_factory=list)
 
     def public(self):
         """The plan fields a response may carry (no fingerprint, no targets)."""
@@ -289,6 +306,90 @@ def _delete_ids(conn, table, ids):
     return removed
 
 
+def _mark_rows(conn, only=None):
+    """``[(locator, size, mtime_ns)]`` of the removal marks (all, or those
+    among *only*), sorted by locator."""
+    if not has_marks_table(conn):
+        return []
+    sql = f"SELECT locator, size, mtime_ns FROM {MARKS_TABLE}"  # nosec B608
+    if only is None:
+        return sorted(tuple(row) for row in conn.execute(sql))
+    found = []
+    for batch in _batches(sorted(only)):
+        found.extend(
+            tuple(row)
+            for row in conn.execute(
+                f"{sql} WHERE locator IN ({_marks(batch)})", batch  # nosec B608
+            )
+        )
+    return sorted(found)
+
+
+def _lists_something(folder):
+    """Whether *folder* holds at least one entry (an unplugged drive's mount
+    point, or an unbound volume, is an empty folder that reads fine)."""
+    try:
+        with os.scandir(folder) as entries:
+            return next(entries, None) is not None
+    except OSError:
+        return False
+
+
+def _name_listed(path):
+    """Whether *path*'s folder lists its name, compared under ``normcase``.
+
+    A locator is ``normcase``d; on a case-sensitive NTFS folder ``lstat`` of
+    the lower-cased spelling fails for a file that is there.
+    """
+    folder, name = os.path.split(path)
+    wanted = os.path.normcase(name)
+    try:
+        with os.scandir(folder) as entries:
+            return any(os.path.normcase(entry.name) == wanted for entry in entries)
+    except OSError:
+        return True  # cannot tell: keep the mark
+
+
+def _stale_mark(locator, size, mtime_ns):
+    """Whether the file of a mark is gone or has other bytes (blocking)."""
+    try:
+        info = os.lstat(locator)
+    except (FileNotFoundError, NotADirectoryError):
+        return os.path.isdir(os.path.dirname(locator)) and not _name_listed(locator)
+    except OSError:
+        return False
+    return (info.st_size, info.st_mtime_ns) != (size, mtime_ns)
+
+
+def _stale_marks(marks, roots):
+    """``[(locator, owner)]`` of the marks that hide nothing any more (#501).
+
+    *marks* are ``(locator, size, mtime_ns)`` rows. A mark is stale when its
+    file was changed, or is gone from a folder that still exists. Only marks
+    under a source that can be read now and lists something are judged (an
+    unplugged drive's marks, or its empty mount point's, stay); any doubt
+    keeps the mark. *owner* is the index of the longest source holding it.
+    Locators are ``normcase(realpath)`` paths, so the sources are too.
+    """
+    readable = readable_roots(roots)
+    prefixes = [
+        os.path.normcase(os.path.realpath(root)).rstrip("\\/") + os.sep
+        for root in roots
+    ]
+    judged = {}
+    stale = []
+    for locator, size, mtime_ns in marks:
+        owners = [i for i, prefix in enumerate(prefixes) if locator.startswith(prefix)]
+        if not owners:
+            continue  # under no source now (a source removed from the settings)
+        owner = max(owners, key=lambda i: len(prefixes[i]))
+        if owner not in judged:
+            judged[owner] = readable[owner] and _lists_something(roots[owner])
+        if judged[owner] and _stale_mark(locator, size, mtime_ns):
+            stale.append((locator, owner))
+    return stale
+
+
 def _read_snapshot(db_path, read):
     """``read(conn)`` on one ``mode=ro`` snapshot; the connection always closes."""
     conn = sqlite3.connect(readonly_uri(db_path), uri=True)
@@ -431,8 +532,15 @@ class MaintenanceOperations:
         if operation == "workflow_storage":
             return self._plan_workflow_storage()
         if operation == "missing_links":
-            rows = _read_snapshot(self.model.db_path, self._link_rows)
-            return self._plan_missing_links(rows, roots)
+            rows, marks = _read_snapshot(
+                self.model.db_path,
+                lambda conn: (self._link_rows(conn), _mark_rows(conn)),
+            )
+            return self._plan_missing_links(rows, roots, _stale_marks(marks, roots))
+        if operation == "pair_metadata":
+            from .pair_metadata_repair import plan_pair_metadata
+
+            return _read_snapshot(self.model.db_path, plan_pair_metadata)
         reader = (
             self._plan_duplicate_prompts
             if operation == "duplicate_prompts"
@@ -563,13 +671,15 @@ class MaintenanceOperations:
         )
 
     @staticmethod
-    def _links_fingerprint(candidates):
+    def _links_fingerprint(candidates, stale=()):
         fingerprint = _Fingerprint("missing_links")
         for row, owner in candidates:
             fingerprint.add([row[0], row[1], row[3], owner])
+        for locator, owner in stale:
+            fingerprint.add(["mark", locator, owner])
         return fingerprint.hexdigest()
 
-    def _plan_missing_links(self, rows, roots):
+    def _plan_missing_links(self, rows, roots, stale=()):
         readable = readable_roots(roots)
         candidates, unavailable, outside = [], 0, 0
         for row, owner, state in link_states(rows, roots, readable):
@@ -597,29 +707,47 @@ class MaintenanceOperations:
                 )
             )
         count = len(candidates)
+        parts = []
+        if count:
+            parts.append(
+                _plural(count, "link to a missing file", "links to missing files")
+            )
+        if stale:
+            parts.append(
+                _plural(
+                    len(stale),
+                    "removed image whose file is gone or changed",
+                    "removed images whose files are gone or changed",
+                )
+            )
+        details = [
+            {
+                "label": _relative(row[1], roots[owner]),
+                "detail": f"{_source_name(roots[owner], owner)} · the file is missing",
+            }
+            for row, owner in candidates
+        ] + [
+            {
+                "label": _relative(locator, os.path.realpath(roots[owner])),
+                "detail": (
+                    f"{_source_name(roots[owner], owner)} · removed image, "
+                    "the file is gone or changed"
+                ),
+            }
+            for locator, owner in stale
+        ]
         return MaintenanceDraft(
             operation="missing_links",
-            fingerprint=self._links_fingerprint(candidates),
+            fingerprint=self._links_fingerprint(candidates, stale),
             targets=[row[0] for row, _owner in candidates],
             scope={
-                "count": count,
-                "label": (
-                    _plural(count, "link to a missing file", "links to missing files")
-                    if count
-                    else "No links to missing files"
-                ),
+                "count": count + len(stale),
+                "label": "; ".join(parts) or "No links to missing files",
             },
             protected={"count": unavailable + outside, "reasons": reasons},
             media={"deletes_files": False, "label": LINKS_ONLY},
-            details=[
-                {
-                    "label": _relative(row[1], roots[owner]),
-                    "detail": (
-                        f"{_source_name(roots[owner], owner)} · the file is missing"
-                    ),
-                }
-                for row, owner in candidates[:MAX_DETAILS]
-            ],
+            details=details[:MAX_DETAILS],
+            marks=[locator for locator, _owner in stale],
         )
 
     def _plan_optimize(self):
@@ -669,14 +797,19 @@ class MaintenanceOperations:
 
     # ── apply ─────────────────────────────────────────────────────────
 
-    def apply_maintenance(self, operation, fingerprint, targets, roots=()) -> dict:
+    def apply_maintenance(
+        self, operation, fingerprint, targets, roots=(), marks=()
+    ) -> dict:
         """Apply a reviewed plan exactly, or raise PlanChanged (blocking).
 
         Inside ``BEGIN IMMEDIATE`` the fingerprint is computed again on the
         write connection; when it differs nothing is written. ``targets`` and
         ``roots`` are the plan's. Returns ``{changed, rows}``: how many
         records changed and up to ``MAX_RESULT_ROWS`` descriptions of them
-        (``{prompt_id[, kept_id]}`` or ``{image_path}``).
+        (``{prompt_id[, kept_id]}`` or ``{image_path}``). ``marks`` are a
+        missing-links plan's stale removal marks; that apply also returns
+        ``marks``, how many it forgot. ``pair_metadata`` returns every filled
+        prompt, uncapped (its job keeps the complete record, #550).
         """
         if operation not in MAINTENANCE_OPERATIONS or operation == "workflow_storage":
             raise ValueError(f"Unknown maintenance operation {operation!r}")
@@ -689,9 +822,16 @@ class MaintenanceOperations:
         appliers = {
             "duplicate_prompts": self._apply_duplicate_prompts,
             "unused_components": self._apply_unused_components,
-            "missing_links": self._apply_missing_links,
         }
         with self._pair_transaction() as conn:
+            if operation == "pair_metadata":
+                from .pair_metadata_repair import apply_pair_metadata
+
+                return apply_pair_metadata(conn, fingerprint, targets)
+            if operation == "missing_links":
+                return self._apply_missing_links(
+                    conn, fingerprint, targets, roots, list(marks)
+                )
             return appliers[operation](conn, fingerprint, targets, roots)
 
     def _apply_duplicate_prompts(self, conn, fingerprint, targets, _roots):
@@ -720,7 +860,7 @@ class MaintenanceOperations:
         ]
         return {"changed": changed, "rows": rows}
 
-    def _apply_missing_links(self, conn, fingerprint, targets, roots):
+    def _apply_missing_links(self, conn, fingerprint, targets, roots, marks):
         rows = []
         for batch in _batches(sorted(targets)):
             rows.extend(
@@ -737,10 +877,14 @@ class MaintenanceOperations:
             if state != "missing":  # back, or its source went away
                 raise PlanChanged("missing_links")
             candidates.append((row, owner))
-        if self._links_fingerprint(candidates) != fingerprint:
+        stale = _stale_marks(_mark_rows(conn, marks) if marks else [], roots)
+        if [locator for locator, _owner in stale] != marks:  # a mark went or is valid
+            raise PlanChanged("missing_links")
+        if self._links_fingerprint(candidates, stale) != fingerprint:
             raise PlanChanged("missing_links")
         changed = _delete_ids(conn, "generated_images", [row[0] for row in rows])
         return {
             "changed": changed,
+            "marks": clear_marks(conn, marks),
             "rows": [{"image_path": row[1]} for row in rows[:MAX_RESULT_ROWS]],
         }

database/managed_connection.py

@@ -35,38 +35,51 @@ from __future__ import annotations
 
 import contextlib
 import logging
+import operator
+import sys
 import threading
 import time
 import weakref
-from typing import Iterable, Iterator, List, Optional
+from collections import deque
+from typing import Any, Callable, Iterable, Iterator, List, Optional
 
 import sqlite3
 
 CLOSED_MESSAGE = "Cannot operate on a closed database."
 # Rows fetched per step when a cursor is iterated: the connection is marked
 # in use once per chunk instead of once per row (iteration costs about what
-# a plain cursor costs). A loop that stops early leaves up to this many rows
-# fetched; ``fetchone``/``fetchmany`` after a partial loop skip them.
+# a plain cursor costs). Prefetched rows belong to the cursor so iteration
+# and explicit fetches always advance the same position.
 ITER_CHUNK = 128
 
 
 class ManagedCursor(sqlite3.Cursor):
     """A cursor whose calls into SQLite count as its connection being in use."""
 
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        super().__init__(*args, **kwargs)
+        self._pending = deque()
+        self._closed = False
+
     def execute(self, sql, parameters=(), /):
-        conn = self.connection
-        conn._enter()
-        try:
-            return super().execute(sql, parameters)
-        finally:
-            conn._leave()
+        return self._run_statement(super().execute, sql, parameters)
 
     def executemany(self, sql, seq_of_parameters, /):
+        return self._run_statement(super().executemany, sql, seq_of_parameters)
+
+    def _run_statement(self, operation: Callable, *args: Any) -> ManagedCursor:
         conn = self.connection
         conn._enter()
+        description = self.description
+        completed = False
         try:
-            return super().executemany(sql, seq_of_parameters)
+            result = operation(*args)
+            completed = True
+            return result
         finally:
+            # Argument parsing can fail before SQLite replaces the old statement.
+            if completed or self.description is not description:
+                self._pending.clear()
             conn._leave()
 
     def executescript(self, sql_script, /):
@@ -81,6 +94,8 @@ class ManagedCursor(sqlite3.Cursor):
         conn = self.connection
         conn._enter()
         try:
+            if self._pending and not self._closed:
+                return self._pending.popleft()
             return super().fetchone()
         finally:
             conn._leave()
@@ -89,7 +104,20 @@ class ManagedCursor(sqlite3.Cursor):
         conn = self.connection
         conn._enter()
         try:
-            return super().fetchmany(*args, **kwargs)
+            # Let SQLite validate the original arguments, including version-specific
+            # handling of zero/negative sizes, before touching prefetched rows.
+            rows = super().fetchmany(*args, **kwargs)
+            if not self._pending:
+                return rows
+            size = operator.index(
+                kwargs.get("size", args[0] if args else self.arraysize)
+            )
+            # Python 3.12 reads all rows for size=0; 3.13+ reads none.
+            if size == 0 and sys.version_info >= (3, 13):
+                return rows
+            self._pending.extend(rows)
+            count = len(self._pending) if size <= 0 else min(size, len(self._pending))
+            return [self._pending.popleft() for _ in range(count)]
         finally:
             conn._leave()
 
@@ -97,25 +125,31 @@ class ManagedCursor(sqlite3.Cursor):
         conn = self.connection
         conn._enter()
         try:
-            return super().fetchall()
+            rows = super().fetchall()
+            pending = list(self._pending)
+            self._pending.clear()
+            return pending + rows
         finally:
             conn._leave()
 
     def __iter__(self) -> Iterator:
-        return self._chunks()
-
-    def _chunks(self) -> Iterator:
-        while True:
-            rows = self.fetchmany(ITER_CHUNK)
-            if not rows:
-                return
-            yield from rows
+        return self
 
     def __next__(self):
         conn = self.connection
+        pending = self._pending
+        # Buffered Python rows need no SQLite lock, but retirement and cursor
+        # close must still stop an iterator obtained before either operation.
+        if conn._retired or self._closed:
+            raise sqlite3.ProgrammingError(CLOSED_MESSAGE)
+        if pending:
+            return pending.popleft()
         conn._enter()
         try:
-            return super().__next__()
+            pending.extend(super().fetchmany(ITER_CHUNK))
+            if not pending:
+                raise StopIteration
+            return pending.popleft()
         finally:
             conn._leave()
 
@@ -126,7 +160,10 @@ class ManagedCursor(sqlite3.Cursor):
         except sqlite3.ProgrammingError:
             return super().close()  # the connection is closed or closing
         try:
-            return super().close()
+            result = super().close()
+            self._pending = deque()
+            self._closed = True
+            return result
         finally:
             conn._leave()
 
@@ -293,6 +330,9 @@ class ManagedConnection(sqlite3.Connection):
             cursors = list(self._cursors)
         try:
             for cursor in cursors:
+                if isinstance(cursor, ManagedCursor):
+                    cursor._pending = deque()
+                    cursor._closed = True
                 with contextlib.suppress(sqlite3.Error):
                     sqlite3.Cursor.close(cursor)
             super().close()

database/models.py

@@ -45,6 +45,7 @@ except ImportError:
     from utils.hashing import register_prompt_functions
 from .image_categories import image_categories_schema_error, migrate_image_categories
 from .image_removal_schema import image_removals_schema_error, migrate_image_removals
+from .prompt_archive import migrate_prompt_archive, prompt_archive_schema_error
 from .workflow_documents import (
     DOCUMENT_ID_COLUMNS,
     migrate_workflow_documents,
@@ -367,6 +368,8 @@ class PromptModel:
         migrate_autotag_state(conn)
         # Image categories (#332): independent of every table above.
         migrate_image_categories(conn)
+        # The Library parent archive (#443): independent of every table above.
+        migrate_prompt_archive(conn)
         # Removal marks (#480): independent of every table above.
         migrate_image_removals(conn)
         # Last: the older generated_images rebuilds run only before file_path.
@@ -1245,6 +1248,11 @@ class PromptModel:
             )
             if categories_error:
                 return False, categories_error
+            archive_error = prompt_archive_schema_error(
+                conn, allow_missing=not require_library_state
+            )
+            if archive_error:
+                return False, archive_error
             removals_error = image_removals_schema_error(
                 conn, allow_missing=not require_library_state
             )

database/operations.py

@@ -25,6 +25,7 @@ from .review import ReviewOperations
 from .review_decisions import ReviewDecisionOperations
 from .saved_views import SavedViewOperations
 from .library_commands import LibraryCommandOperations
+from .library_archive import LibraryArchiveOperations, delete_prompt_rows
 from .vocabulary import VocabularyOperations
 from .execution_history import ExecutionHistoryOperations
 from .execution_contexts import ExecutionContextOperations
@@ -33,7 +34,7 @@ from .history_reads import HistoryReadOperations
 from .health import HealthOperations, iter_consistency_issues
 from .duplicate_copies import DuplicateCopyOperations
 from .workflow_storage import WorkflowStorageOperations
-from .category_import import CategoryImportOperations, EmptyReplacement  # noqa: F401
+from .category_import import CategoryImportOperations
 from .image_categories import ImageCategoryOperations
 from .workflow_documents import (
     DocumentReader,
@@ -449,6 +450,7 @@ class PromptDatabase(
     PairOperations,
     LibraryOperations,
     LibraryCommandOperations,
+    LibraryArchiveOperations,
     VocabularyOperations,
     ExecutionHistoryOperations,
     ExecutionContextOperations,
@@ -2351,16 +2353,10 @@ class PromptDatabase(
         Returns:
             Number of prompts actually deleted.
         """
-        count = 0
         with self._pair_transaction() as conn:
             self._ensure_prompts_unpaired(conn, prompt_ids)
-            documents = documents_of_rows(conn, "prompt_id", prompt_ids)
-            for pid in prompt_ids:
-                conn.execute("DELETE FROM generated_images WHERE prompt_id = ?", (pid,))
-                cursor = conn.execute("DELETE FROM prompts WHERE id = ?", (pid,))
-                if cursor.rowcount > 0:
-                    count += 1
-            release_documents(conn, documents)
+            # The workspace Library's delete runs the same statements (#443).
+            count = delete_prompt_rows(conn, list(dict.fromkeys(prompt_ids)))
             conn.commit()
         return count
 

database/pair_metadata_repair.py

@@ -0,0 +1,452 @@
+"""Pair category and tags repair (#550): the ``pair_metadata`` maintenance plan.
+
+Before #546 a Pair node run saved its category and tags on the saved pair
+only; the positive prompt it created stayed Uncategorized and untagged. This
+plan fills the **empty** category and the **empty** tag set of those prompts
+from their saved pairs (``docs/egm/post-3.4.1/550-design.md``):
+
+- Considered: prompts the Pair node created as a positive (their earliest
+  pair, in either role, has them as the positive and its ``created_at`` is
+  within ``PAIR_CREATED_TOLERANCE`` of theirs) whose category is blank
+  and/or that have no tags.
+- Sources: the prompt's non-archived pairs where it is the positive. Pairs
+  with a blank category or no tags are ignored for that field. A field fills
+  when exactly one value remains (category: trimmed, case-sensitive; tags:
+  the tag-id set).
+- Kept (protected), in this order: archived prompt, edited prompt
+  (``updated_at`` not within the tolerance of ``created_at``), pairs that
+  disagree, a tag set over ``MAX_LIBRARY_TAGS``.
+
+The apply runs in the caller's ``BEGIN IMMEDIATE`` transaction, recomputes
+the fingerprint, then writes guarded statements (the field must still be
+empty) that never touch a pair, a negative or ``updated_at``. A guard that
+hits nothing raises ``PlanChanged`` so the whole transaction rolls back.
+"""
+
+import datetime
+
+from . import maintenance as _maintenance
+from .library_commands import MAX_LIBRARY_TAGS
+from .maintenance import (
+    MAX_DETAILS,
+    MaintenanceDraft,
+    PlanChanged,
+    _excerpt,
+    _Fingerprint,
+    _marks,
+    _plural,
+)
+
+OPERATION = "pair_metadata"
+# Seconds: the Pair save stamps its new prompt and its pair with separate
+# utc_now_iso() calls in one transaction (microseconds apart).
+PAIR_CREATED_TOLERANCE = 2.0
+# Fill prompts one plan applies; the rest are applied by the next review.
+# Bounds the complete record a job keeps in memory (about 100 bytes a row).
+MAX_PLAN_FILLS = 20000
+DETAIL_VALUES = 3
+DETAIL_TAGS = 5
+MEDIA_LABEL = "No image file or saved pair is changed."
+EMPTY_LABEL = "No Pair prompts are missing a category or tags"
+RECORD_NOTE = (
+    "The job's report lists every filled field. Download it to keep it: "
+    "Activity forgets jobs after a restart."
+)
+# Protected classes in precedence order: (key, detail, reason one, many).
+EDITED_DETAIL = (
+    "Kept: edited after it was saved; its empty category or tags may be intentional"
+)
+PROTECTED = (
+    (
+        "archived",
+        "Kept: archived in the Library",
+        "archived in the Library",
+        "archived in the Library",
+    ),
+    (
+        "edited",
+        EDITED_DETAIL,
+        "edited after it was saved",
+        "edited after they were saved",
+    ),
+    ("conflict", None, "where pairs disagree", "where pairs disagree"),
+    (
+        "too-many-tags",
+        f"Kept: its pair has more than {MAX_LIBRARY_TAGS} tags",
+        f"whose pair has more than {MAX_LIBRARY_TAGS} tags",
+        f"whose pair has more than {MAX_LIBRARY_TAGS} tags",
+    ),
+)
+_PROTECTED_ORDER = [key for key, *_rest in PROTECTED]
+# Detail rows: what the user can act on (a conflict) first.
+DETAIL_ORDER = ("conflict", "edited", "archived", "too-many-tags")
+
+_CANDIDATES_SQL = """
+    SELECT p.id, p.hash, p.text, p.created_at, p.updated_at,
+           COALESCE(TRIM(p.category), '') = '' AS blank_category,
+           NOT EXISTS (SELECT 1 FROM prompt_tags pt WHERE pt.prompt_id = p.id)
+               AS untagged,
+           EXISTS (SELECT 1 FROM prompt_archive pa WHERE pa.prompt_id = p.id)
+               AS archived,
+           (SELECT CASE WHEN pp.positive_prompt_id = p.id
+                        THEN pp.created_at END
+              FROM prompt_pairs pp
+             WHERE pp.positive_prompt_id = p.id OR pp.negative_prompt_id = p.id
+             ORDER BY pp.id LIMIT 1) AS first_pair_at
+      FROM prompts p
+     WHERE EXISTS (SELECT 1 FROM prompt_pairs pp WHERE pp.positive_prompt_id = p.id)
+       AND (COALESCE(TRIM(p.category), '') = ''
+            OR NOT EXISTS (SELECT 1 FROM prompt_tags pt WHERE pt.prompt_id = p.id))
+     ORDER BY p.id
+"""
+_SOURCES_SQL = (
+    "SELECT id, positive_prompt_id, TRIM(COALESCE(category, '')) FROM prompt_pairs "
+    "WHERE archived = 0 AND positive_prompt_id IN ({marks}) ORDER BY id"
+)
+_PAIR_TAGS_SQL = (
+    "SELECT ppt.pair_id, t.id, t.name FROM prompt_pair_tags ppt "
+    "JOIN tags t ON t.id = ppt.tag_id WHERE ppt.pair_id IN ({marks}) "
+    "ORDER BY ppt.pair_id, t.id"
+)
+
+
+def parse_stamp(value):
+    """A stored timestamp as an aware UTC datetime, or None when unreadable."""
+    if not isinstance(value, str) or not value.strip():
+        return None
+    try:
+        stamp = datetime.datetime.fromisoformat(value.strip())
+    except ValueError:
+        return None
+    if stamp.tzinfo is None:
+        stamp = stamp.replace(tzinfo=datetime.timezone.utc)
+    return stamp
+
+
+def _close(first, second):
+    """Whether two stored stamps parse and lie within the tolerance."""
+    a, b = parse_stamp(first), parse_stamp(second)
+    if a is None or b is None:
+        return False
+    return abs((a - b).total_seconds()) <= PAIR_CREATED_TOLERANCE
+
+
+def _batches(ids):
+    size = _maintenance.ID_BATCH
+    for start in range(0, len(ids), size):
+        yield ids[start : start + size]
+
+
+def _sources(conn, ids):
+    """``{prompt_id: [(pair_id, category, ((tag_id, name), ...)), ...]}``."""
+    pairs = {}
+    for batch in _batches(ids):
+        for pair_id, prompt_id, category in conn.execute(
+            _SOURCES_SQL.format(marks=_marks(batch)), batch
+        ):
+            pairs.setdefault(prompt_id, []).append([pair_id, category, []])
+    by_pair = {pair[0]: pair for group in pairs.values() for pair in group}
+    pair_ids = sorted(by_pair)
+    for batch in _batches(pair_ids):
+        for pair_id, tag_id, name in conn.execute(
+            _PAIR_TAGS_SQL.format(marks=_marks(batch)), batch
+        ):
+            by_pair[pair_id][2].append((tag_id, name))
+    return {
+        prompt_id: [(pid, category, tuple(tags)) for pid, category, tags in group]
+        for prompt_id, group in pairs.items()
+    }
+
+
+def _field(values, guard):
+    """``(class, value)`` of one field from its distinct values, or None."""
+    if not values:
+        return None
+    if guard:
+        return (guard, None)
+    if len(values) > 1:
+        return ("conflict", sorted(values))
+    return ("fill", next(iter(values)))
+
+
+def _classify(row, sources):
+    """The result of one considered prompt, or None when nothing applies."""
+    guard = None
+    if row["archived"]:
+        guard = "archived"
+    elif not _close(row["updated_at"], row["created_at"]):
+        guard = "edited"
+    category = tags = None
+    tag_pair = None
+    if row["blank_category"]:
+        category = _field({c for _p, c, _t in sources if c}, guard)
+    if row["untagged"]:
+        sets = {}
+        for pair_id, _category, pair_tags in sources:
+            if pair_tags:
+                sets.setdefault(frozenset(pair_tags), pair_id)
+        tags = _field(set(sets), guard)
+        if tags and tags[0] == "fill":
+            tag_pair = sets[tags[1]]
+            if len(tags[1]) > MAX_LIBRARY_TAGS:
+                tags = ("too-many-tags", None)
+            else:
+                tags = ("fill", tuple(sorted(tags[1])))
+        elif tags and tags[0] == "conflict":
+            tags = ("conflict", [sorted(s) for s in sorted(sets, key=sorted)])
+    if category is None and tags is None:
+        return None
+    return {
+        "id": row["id"],
+        "hash": row["hash"],
+        "text": row["text"],
+        "category": category,
+        "tags": tags,
+        "tag_pair": tag_pair if tags and tags[0] == "fill" else None,
+        "pairs": [pair_id for pair_id, _c, _t in sources],
+    }
+
+
+def _state(conn):
+    """``(results, fingerprint)``: every considered prompt's result, by id."""
+    considered = []
+    for row in conn.execute(_CANDIDATES_SQL):
+        row = dict(
+            zip(
+                (
+                    "id",
+                    "hash",
+                    "text",
+                    "created_at",
+                    "updated_at",
+                    "blank_category",
+                    "untagged",
+                    "archived",
+                    "first_pair_at",
+                ),
+                tuple(row),
+            )
+        )
+        if _close(row["created_at"], row["first_pair_at"]):
+            considered.append(row)
+    sources = _sources(conn, [row["id"] for row in considered])
+    fingerprint = _Fingerprint(OPERATION)
+    results = []
+    for row in considered:
+        result = _classify(row, sources.get(row["id"], []))
+        if result is None:
+            continue
+        results.append(result)
+        fingerprint.add(
+            [
+                result["id"],
+                result["hash"],
+                result["category"],
+                result["tags"],
+                result["tag_pair"],
+                result["pairs"],
+            ]
+        )
+    return results, fingerprint.hexdigest()
+
+
+def _fills(result):
+    return (
+        bool(result["category"] and result["category"][0] == "fill"),
+        bool(result["tags"] and result["tags"][0] == "fill"),
+    )
+
+
+def _targets(results):
+    targets = []
+    for result in results:
+        category, tags = _fills(result)
+        if category or tags:
+            targets.append([result["id"], category, tags])
+    return targets[:MAX_PLAN_FILLS]
+
+
+def _protected_classes(result):
+    """The protected classes of a result's fields, in precedence order."""
+    found = {
+        field[0]
+        for field in (result["category"], result["tags"])
+        if field and field[0] != "fill"
+    }
+    return [key for key in _PROTECTED_ORDER if key in found]
+
+
+def _more(values, limit):
+    shown = ", ".join(values[:limit])
+    return shown + (", …" if len(values) > limit else "")
+
+
+def _fill_detail(result):
+    parts = []
+    category, tags = _fills(result)
+    if category:
+        parts.append(f"category “{result['category'][1]}”")
+    if tags:
+        names = sorted(name for _id, name in result["tags"][1])
+        noun = "tag" if len(names) == 1 else "tags"
+        parts.append(f"{len(names)} {noun}: {_more(names, DETAIL_TAGS)}")
+    pairs = [str(pair_id) for pair_id in result["pairs"]]
+    label = "pair" if len(pairs) == 1 else "pairs"
+    return f"Fills {' and '.join(parts)} ({label} {_more(pairs, DETAIL_VALUES)})"
+
+
+def _protected_detail(result, key):
+    if key != "conflict":
+        return next(detail for k, detail, *_r in PROTECTED if k == key)
+    if result["category"] and result["category"][0] == "conflict":
+        values = result["category"][1]
+        return f"Kept: pairs disagree on category: {_more(values, DETAIL_VALUES)}"
+    return "Kept: pairs disagree on tags"
+
+
+def _kept(results):
+    """One row per kept prompt, its review reasons without "Kept: " (#550)."""
+    rows = []
+    for result in results:
+        classes = set(_protected_classes(result))
+        if classes:
+            rows.append(
+                {
+                    "prompt_id": result["id"],
+                    "reasons": [
+                        _protected_detail(result, key).removeprefix("Kept: ")
+                        for key in DETAIL_ORDER
+                        if key in classes
+                    ],
+                }
+            )
+    return rows
+
+
+def _details(results, targets):
+    rows = []
+    for key in DETAIL_ORDER:
+        for result in results:
+            if key in _protected_classes(result):
+                rows.append(
+                    {
+                        "label": _excerpt(result["text"]),
+                        "detail": _protected_detail(result, key),
+                    }
+                )
+    planned = {target[0] for target in targets}
+    for result in results:
+        if result["id"] in planned:
+            rows.append(
+                {"label": _excerpt(result["text"]), "detail": _fill_detail(result)}
+            )
+    if not targets:
+        return rows[:MAX_DETAILS]
+    return rows[: MAX_DETAILS - 1] + [{"label": "Record", "detail": RECORD_NOTE}]
+
+
+def _scope_label(targets, total):
+    if not targets:
+        return EMPTY_LABEL
+    categories = sum(1 for _id, category, _tags in targets if category)
+    tag_sets = sum(1 for _id, _category, tags in targets if tags)
+    label = (
+        f"{_plural(len(targets), 'prompt')} to fill: "
+        f"{_plural(categories, 'category', 'categories')}, "
+        f"{_plural(tag_sets, 'tag set')}"
+    )
+    if total > len(targets):
+        label += (
+            f" (the first {len(targets):,} of {total:,}; review again for the rest)"
+        )
+    return label
+
+
+def plan_pair_metadata(conn):
+    """The reviewed ``pair_metadata`` plan on ``conn``'s snapshot."""
+    results, fingerprint = _state(conn)
+    targets = _targets(results)
+    total = sum(1 for result in results if any(_fills(result)))
+    counts = {key: 0 for key in _PROTECTED_ORDER}
+    held = 0
+    for result in results:
+        classes = _protected_classes(result)
+        held += bool(classes)
+        for key in classes:
+            counts[key] += 1
+    reasons = [
+        f"{counts[key]:,} {one if counts[key] == 1 else many}"
+        for key, _detail, one, many in PROTECTED
+        if counts[key]
+    ]
+    return MaintenanceDraft(
+        operation=OPERATION,
+        fingerprint=fingerprint,
+        targets=targets,
+        scope={"count": len(targets), "label": _scope_label(targets, total)},
+        protected={"count": held, "reasons": reasons},
+        media={"deletes_files": False, "label": MEDIA_LABEL},
+        details=_details(results, targets),
+    )
+
+
+def _fill_categories(conn, fills):
+    """Set each blank category; raises PlanChanged when one is no longer blank."""
+    cursor = conn.executemany(
+        "UPDATE prompts SET category = ? WHERE id = ? "
+        "AND COALESCE(TRIM(category), '') = ''",
+        fills,
+    )
+    if cursor.rowcount != len(fills):
+        raise PlanChanged(OPERATION)
+
+
+def _fill_tags(conn, fills, expected):
+    """Copy each source pair's tags onto its untagged prompt."""
+    cursor = conn.executemany(
+        "INSERT INTO prompt_tags(prompt_id, tag_id) "
+        "SELECT ?, ppt.tag_id FROM prompt_pair_tags ppt "
+        "JOIN tags t ON t.id = ppt.tag_id WHERE ppt.pair_id = ? "
+        "AND NOT EXISTS (SELECT 1 FROM prompt_tags WHERE prompt_id = ?)",
+        fills,
+    )
+    if cursor.rowcount != expected:
+        raise PlanChanged(OPERATION)
+
+
+def apply_pair_metadata(conn, fingerprint, targets):
+    """Apply the reviewed fills on ``conn`` (inside BEGIN IMMEDIATE).
+
+    Returns ``{changed, categories, tag_sets, rows, kept}``; ``rows`` lists
+    every filled prompt (``{prompt_id, category?, tags?}``) and ``kept`` every
+    prompt the plan keeps (``{prompt_id, reasons}``, in the review's wording
+    and order), both never capped.
+    """
+    results, current = _state(conn)
+    if current != fingerprint or _targets(results) != targets:
+        raise PlanChanged(OPERATION)
+    by_id = {result["id"]: result for result in results}
+    categories, tags, rows = [], [], []
+    tag_count = 0
+    for prompt_id, fill_category, fill_tags in targets:
+        result = by_id[prompt_id]
+        row = {"prompt_id": prompt_id}
+        if fill_category:
+            row["category"] = result["category"][1]
+            categories.append((row["category"], prompt_id))
+        if fill_tags:
+            row["tags"] = sorted(name for _id, name in result["tags"][1])
+            tags.append((prompt_id, result["tag_pair"], prompt_id))
+            tag_count += len(result["tags"][1])
+        rows.append(row)
+    if categories:
+        _fill_categories(conn, categories)
+    if tags:
+        _fill_tags(conn, tags, tag_count)
+    return {
+        "changed": len(rows),
+        "categories": len(categories),
+        "tag_sets": len(tags),
+        "rows": rows,
+        "kept": _kept(results),
+    }

database/pairs.py

@@ -112,8 +112,14 @@ class PairOperations:
         tags: Optional[List[str]] = None,
         rating: Optional[int] = None,
         notes: Optional[str] = None,
+        fill_new_positive: bool = False,
     ) -> int:
-        """Save an ordered pair atomically; repeat saves preserve its metadata."""
+        """Save an ordered pair atomically; repeat saves preserve its metadata.
+
+        ``fill_new_positive`` also gives the positive prompt the category and
+        tags, only if this save inserted its row (#546). An existing prompt
+        never changes, so a cleared category stays cleared.
+        """
         with self._pair_transaction() as conn:
             return self._save_pair(
                 conn,
@@ -124,6 +130,7 @@ class PairOperations:
                 tags=tags,
                 rating=rating,
                 notes=notes,
+                fill_new_positive=fill_new_positive,
             )
 
     def _save_pair(
@@ -137,6 +144,7 @@ class PairOperations:
         tags: Optional[List[str]] = None,
         rating: Optional[int] = None,
         notes: Optional[str] = None,
+        fill_new_positive: bool = False,
     ) -> int:
         """Save an ordered pair atomically; repeat saves preserve its metadata."""
         if not isinstance(positive_text, str) or not positive_text.strip():
@@ -144,8 +152,13 @@ class PairOperations:
         if not isinstance(negative_text, str):
             raise ValueError("Negative prompt text must be a string; use '' for empty")
         # Validate metadata before any component insertion.
-        _validate_metadata(title, category, tags, rating, notes)
-        positive = self._pair_component(conn, positive_text)
+        tags = _validate_metadata(title, category, tags, rating, notes)
+        positive, created = self._pair_component_outcome(conn, positive_text)
+        if created and fill_new_positive:
+            conn.execute(
+                "UPDATE prompts SET category = ? WHERE id = ?", (category, positive)
+            )
+            self._sync_prompt_tags(conn, positive, tags or [])
         negative = (
             self._pair_component(conn, negative_text) if negative_text.strip() else None
         )

database/prompt_archive.py

@@ -0,0 +1,107 @@
+"""Library parent archive (#443, D152): prompts hidden from the shelf, restorable.
+
+A row means the Library parent is archived; no row means it is on the shelf.
+Nothing else about the prompt changes: its text, metadata, tags, run count,
+image links, pairs and recorded runs stay. Restore deletes the row. Deleting
+the prompt (any path) drops the row by cascade.
+
+A table of its own, not a column: ``prompt_library_state`` is validated by its
+exact definition, so a new column there would make older builds refuse the
+upgraded database, while older builds ignore an unknown table (they show the
+archived prompts again and keep the rows). Optional in old backups, required
+once a database is current.
+"""
+
+import re
+import sqlite3
+
+TABLE = "prompt_archive"
+MAX_STAMP_BYTES = 64
+# What ``utc_now_iso`` writes starts with a date; anything else was not written here.
+_STAMP_GLOB = "[0-9][0-9][0-9][0-9]-[0-9][0-9]-[0-9][0-9]T*"
+_DDL = (
+    f"CREATE TABLE {TABLE} (prompt_id INTEGER PRIMARY KEY, "
+    "archived_at TEXT NOT NULL, "
+    "FOREIGN KEY(prompt_id) REFERENCES prompts(id) ON DELETE CASCADE)"
+)
+_COLUMNS = (("prompt_id", "INTEGER", 0, 1), ("archived_at", "TEXT", 1, 0))
+_FOREIGN_KEYS = {("prompt_id", "prompts", "id", "CASCADE")}
+# A predicate on ``p`` (the prompts alias): the parent is not archived.
+SHELVED = f"NOT EXISTS(SELECT 1 FROM {TABLE} pa WHERE pa.prompt_id=p.id)"
+ARCHIVED = f"EXISTS(SELECT 1 FROM {TABLE} pa WHERE pa.prompt_id=p.id)"
+
+
+def _compact(sql: str) -> str:
+    return re.sub(r"\s+", "", sql or "").lower()
+
+
+def _structure_error(conn: sqlite3.Connection, allow_missing: bool) -> str | None:
+    objects = conn.execute(
+        "SELECT type,sql FROM sqlite_master WHERE name=? "
+        "OR (type IN ('index','trigger') AND tbl_name=?)",
+        (TABLE, TABLE),
+    ).fetchall()
+    if not objects:
+        return None if allow_missing else "Missing prompt archive schema"
+    tables = [sql for kind, sql in objects if kind == "table"]
+    if len(tables) != 1 or len(objects) != 1:
+        return "Prompt archive name is reserved by an unsupported schema object"
+    if _compact(tables[0]) != _compact(_DDL):
+        return "Prompt archive has an invalid table definition"
+    columns = tuple(
+        (row[1], row[2].upper(), row[3], row[5])
+        for row in conn.execute(f"PRAGMA table_info({TABLE})")
+    )
+    if columns != _COLUMNS:
+        return "Prompt archive has invalid columns"
+    foreign_keys = {
+        (row[3], row[2], row[4], row[6])
+        for row in conn.execute(f"PRAGMA foreign_key_list({TABLE})")
+    }
+    if foreign_keys != _FOREIGN_KEYS:
+        return "Prompt archive has invalid foreign key definitions"
+    return None
+
+
+def prompt_archive_schema_error(
+    conn: sqlite3.Connection, *, allow_missing: bool = True
+) -> str | None:
+    """Reject an unsupported archive schema or values; reads only.
+
+    A missing table is acceptable for backups that predate it unless
+    ``allow_missing`` is false.
+    """
+    error = _structure_error(conn, allow_missing)
+    if error is not None:
+        return error
+    if not conn.execute(
+        "SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", (TABLE,)
+    ).fetchone():
+        return None
+    if conn.execute(f"PRAGMA foreign_key_check({TABLE})").fetchone():
+        return "Prompt archive contains foreign key violations"
+    if conn.execute(
+        f"SELECT 1 FROM {TABLE} WHERE typeof(archived_at)!='text' "  # nosec B608
+        f"OR length(CAST(archived_at AS BLOB))>{MAX_STAMP_BYTES} "
+        f"OR archived_at NOT GLOB '{_STAMP_GLOB}' LIMIT 1"
+    ).fetchone():
+        return "Prompt archive contains invalid values"
+    return None
+
+
+def migrate_prompt_archive(conn: sqlite3.Connection) -> None:
+    """Install the table atomically; validate before and after (no backfill)."""
+    conn.execute("SAVEPOINT prompt_archive_schema")
+    try:
+        error = _structure_error(conn, True)
+        if error:
+            raise sqlite3.DatabaseError(error)
+        conn.execute(_DDL.replace("CREATE TABLE", "CREATE TABLE IF NOT EXISTS", 1))
+        error = _structure_error(conn, False)
+        if error:
+            raise sqlite3.DatabaseError(error)
+        conn.execute("RELEASE prompt_archive_schema")
+    except BaseException:
+        conn.execute("ROLLBACK TO prompt_archive_schema")
+        conn.execute("RELEASE prompt_archive_schema")
+        raise

database/restore_stage.py

@@ -149,6 +149,7 @@ def library_summary(conn: sqlite3.Connection) -> Dict[str, int]:
     collections (#331).
     """
     from .library import _ELIGIBLE
+    from .prompt_archive import SHELVED, TABLE as ARCHIVE
 
     has_state = conn.execute(
         "SELECT 1 FROM sqlite_master WHERE type='table' AND "
@@ -157,9 +158,15 @@ def library_summary(conn: sqlite3.Connection) -> Dict[str, int]:
     has_pairs = conn.execute(
         "SELECT 1 FROM sqlite_master WHERE type='table' AND name='prompt_pairs'"
     ).fetchone()
+    has_archive = conn.execute(
+        "SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", (ARCHIVE,)
+    ).fetchone()
     if has_state and has_pairs:
+        # Archived parents (#443) are off the shelf, as in library_counts.
         positives = conn.execute(
-            "SELECT COUNT(*) FROM prompts p WHERE " + _ELIGIBLE
+            "SELECT COUNT(*) FROM prompts p WHERE "
+            + _ELIGIBLE
+            + (" AND " + SHELVED if has_archive else "")
         ).fetchone()[0]
     else:
         positives = _count(conn, "prompts")

database/review_decisions.py

@@ -5,8 +5,12 @@ A decision is one row of ``media_review_decisions`` for one item revision.
 ingester's compatible-only rules (``_ingest_resolved_pair_in``) in the same
 ``_pair_transaction`` as the decision row, the item's new state and revision
 and the review index revision; any conflict or later failure rolls all of it
-back. ``unresolved`` records the decision and links nothing. No decision
-counts a run, touches usage or calls an execution writer.
+back. ``manual`` (#536) is the same but for a Library prompt the reviewer chose:
+its link's provenance names a manual choice and carries no prompt sides, so it
+cannot pose as file evidence, and the decision's ``candidate_hash`` (the table
+requires one) is the digest of that choice. ``unresolved`` records the decision
+and links nothing. No decision counts a run, touches usage or calls an execution
+writer.
 
 Callers verify the file first (the capture must be the item's bytes and the
 current file must still hash to them); nothing here reads the filesystem.
@@ -17,6 +21,7 @@ import sqlite3
 
 from .annotation_schema import BYTES_TABLE
 from .annotations import _new_id
+from .library import _ELIGIBLE, _FROM, SHELVED
 from .library_commands import LibraryCommandError
 from .models import utc_now_iso
 from .review import (
@@ -30,9 +35,28 @@ from .review import (
     same_name_rows,
 )
 from .review_changes import notify_review_change
-from .review_schema import DECISIONS_TABLE, ITEMS_TABLE, canonical_evidence
+from .review_schema import (
+    DECISIONS_TABLE,
+    ITEMS_TABLE,
+    canonical_evidence,
+    evidence_digest,
+)
 
 DECISION_KIND = "user-selection"
+MANUAL_KIND = "manual-link"
+_MANUAL_POSITIVE = (
+    f"SELECT p.id,p.text FROM {_FROM} "  # nosec B608
+    f"WHERE p.id=? AND {_ELIGIBLE} AND {SHELVED}"
+)
+_MANUAL_PAIR = (
+    "SELECT COALESCE(n.text,'') AS negative FROM prompt_pairs pp "
+    "LEFT JOIN prompts n ON n.id=pp.negative_prompt_id "
+    "WHERE pp.id=? AND pp.positive_prompt_id=? AND pp.archived=0"
+)
+_ARCHIVED_BARE_PAIR = (
+    "SELECT 1 FROM prompt_pairs WHERE positive_prompt_id=? "
+    "AND negative_prompt_id IS NULL AND archived=1"
+)
 _INSERT_DECISION = (
     f"INSERT INTO {DECISIONS_TABLE}(id,item_id,decision_revision,"  # nosec B608
     "created_at,outcome,candidate_hash,evidence,pair_id,image_id) "
@@ -59,6 +83,8 @@ class ReviewCommandError(LibraryCommandError):
     ``fingerprint`` the current dependency fingerprint, when known;
     ``changed`` marks an association conflict caused by links that changed
     since the reviewer read them (rather than an incompatible choice).
+    ``prompt-missing``: the Library prompt or recipe of a manual link is gone,
+    archived or not that prompt's.
     """
 
     def __init__(
@@ -93,6 +119,7 @@ class ReviewDecisionOperations:
         sha256: str,
         outcome: str,
         candidate: dict | None = None,
+        manual: dict | None = None,
         image_path: str | None = None,
         metadata: dict | None = None,
         expected_fingerprint: str | None = None,
@@ -101,7 +128,9 @@ class ReviewDecisionOperations:
     ) -> dict:
         """Record the decision; return ``{item, total, pair}``.
 
-        ``selected`` needs the evidence ``candidate`` (with its ``hash``), the
+        ``selected`` needs the evidence ``candidate`` (with its ``hash``) or
+        ``manual`` (``{positive_id, pair_id}``; a null ``pair_id`` is no
+        negative), the
         ``image_path`` the scan and the monitor link under, the captured
         ``metadata`` and the ``expected_fingerprint`` the reviewer saw.
         ``aliases`` are the other spellings legacy rows of the same physical
@@ -115,9 +144,12 @@ class ReviewDecisionOperations:
         if outcome not in ("selected", "unresolved"):
             raise ValueError("outcome must be selected or unresolved")
         if outcome == "selected" and (
-            not isinstance(candidate, dict) or not image_path
+            (not isinstance(candidate, dict)) == (not isinstance(manual, dict))
+            or not image_path
         ):
-            raise ValueError("A selected decision needs a candidate and a file")
+            raise ValueError(
+                "A selected decision needs one candidate or manual choice and a file"
+            )
         try:
             with self._pair_transaction() as conn:
                 decided = self._decide_in(
@@ -128,6 +160,7 @@ class ReviewDecisionOperations:
                     sha256=sha256,
                     outcome=outcome,
                     candidate=candidate,
+                    manual=manual,
                     image_path=image_path,
                     metadata=metadata or {},
                     expected_fingerprint=expected_fingerprint,
@@ -158,13 +191,24 @@ class ReviewDecisionOperations:
             raise _Rollback(ReviewCommandError("file-changed", epoch))
         decision_id = _new_id()
         pair = None
-        candidate = command["candidate"] if command["outcome"] == "selected" else None
+        selected = command["outcome"] == "selected"
+        candidate = command["candidate"] if selected else None
+        manual = command["manual"] if selected else None
+        candidate_hash = candidate["hash"] if candidate is not None else None
         if candidate is not None:
             pair = self._apply_candidate(
                 conn, epoch, current, decision_id, total, **command
             )
+        elif manual is not None:
+            pair = self._apply_manual(
+                conn, epoch, current, decision_id, total, **command
+            )
+            manual = {"positive_id": manual["positive_id"], "pair_id": pair["id"]}
+            candidate_hash = evidence_digest(
+                canonical_evidence({"kind": MANUAL_KIND, **manual})
+            )
         evidence = {
-            "kind": DECISION_KIND,
+            "kind": MANUAL_KIND if manual is not None else DECISION_KIND,
             "decision_id": decision_id,
             "outcome": command["outcome"],
             "reviewed_digest": command["sha256"],
@@ -174,6 +218,8 @@ class ReviewDecisionOperations:
             "dependency_fingerprint": command["expected_fingerprint"],
             "pair": pair,
         }
+        if manual is not None:
+            evidence["manual"] = manual
         now = utc_now_iso()
         self._insert_review_decision(
             conn,
@@ -183,7 +229,7 @@ class ReviewDecisionOperations:
                 current["revision"] + 1,
                 now,
                 command["outcome"],
-                candidate["hash"] if candidate is not None else None,
+                candidate_hash,
                 canonical_evidence(evidence),
                 pair["id"] if pair else None,
                 pair["image_id"] if pair else None,
@@ -204,6 +250,77 @@ class ReviewDecisionOperations:
         self, conn, epoch, current, decision_id, total, **command
     ) -> dict:
         """Link the candidate's pair, compatible-only; raise on any conflict."""
+        candidate = command["candidate"]
+        positive, negative = candidate["positive"], candidate["negative"]
+        provenance = {
+            "kind": DECISION_KIND,
+            "decision_id": decision_id,
+            "reviewed_digest": command["sha256"],
+            "candidate_hash": candidate["hash"],
+            "resolver_version": current["resolver_version"],
+            "positive": positive,
+            "negative": negative,
+        }
+        return self._link_compatible(
+            conn, epoch, current, total, positive, negative, provenance, command
+        )
+
+    def _apply_manual(self, conn, epoch, current, decision_id, total, **command):
+        """Link the chosen Library prompt, compatible-only; raise otherwise."""
+        missing = _Rollback(
+            ReviewCommandError("prompt-missing", epoch, item=current, total=total)
+        )
+        positive_id = command["manual"]["positive_id"]
+        pair_id = command["manual"]["pair_id"]
+        row = conn.execute(_MANUAL_POSITIVE, (positive_id,)).fetchone()
+        if row is None:
+            raise missing
+        positive_text, negative_text = row[1], ""
+        if pair_id is not None:
+            chosen = conn.execute(_MANUAL_PAIR, (pair_id, positive_id)).fetchone()
+            if chosen is None:
+                raise missing
+            negative_text = chosen[0]
+        elif conn.execute(_ARCHIVED_BARE_PAIR, (positive_id,)).fetchone():
+            raise missing
+        provenance = {
+            "kind": MANUAL_KIND,
+            "decision_id": decision_id,
+            "reviewed_digest": command["sha256"],
+            "resolver_version": current["resolver_version"],
+            "positive_id": positive_id,
+            "pair_id": pair_id,
+        }
+        # The sides name no provenance: only the ingester reads them.
+        pair = self._link_compatible(
+            conn,
+            epoch,
+            current,
+            total,
+            {"text": positive_text},
+            {"text": negative_text},
+            provenance,
+            command,
+        )
+        # A prompt whose stored hash is stale reaches another row: never link that.
+        reached = conn.execute(
+            "SELECT positive_prompt_id FROM prompt_pairs WHERE id=?", (pair["id"],)
+        ).fetchone()[0]
+        if reached != positive_id or pair_id not in (None, pair["id"]):
+            raise missing
+        return pair
+
+    def _link_compatible(
+        self,
+        conn,
+        epoch,
+        current,
+        total,
+        positive,
+        negative,
+        provenance,
+        command,
+    ) -> dict:
         # Imported here: pair_ingestion imports operations, which composes this.
         from .pair_ingestion import _ingest_resolved_pair_in
 
@@ -242,17 +359,6 @@ class ReviewDecisionOperations:
             # Enrich the one row of this file under the spelling it was
             # linked with, so the ingester's compatibility rules see it.
             image_path = rows[0][1]
-        candidate = command["candidate"]
-        positive, negative = candidate["positive"], candidate["negative"]
-        provenance = {
-            "kind": DECISION_KIND,
-            "decision_id": decision_id,
-            "reviewed_digest": command["sha256"],
-            "candidate_hash": candidate["hash"],
-            "resolver_version": current["resolver_version"],
-            "positive": positive,
-            "negative": negative,
-        }
         result = _ingest_resolved_pair_in(
             conn,
             self,

database/safety_backup.py

@@ -46,6 +46,7 @@ def remove_quietly(*paths):
             if path and os.path.exists(path):
                 os.remove(path)
         except OSError:
+            # cleanup after a failure must not replace that failure
             pass
 
 

database/vocabulary.py

@@ -20,7 +20,8 @@ counted, hashed into the token and renamed with the prompts and pairs.
 
 import sqlite3
 
-from .library import _ELIGIBLE, UNTAGGED, _escape, _limit
+from .library import _ELIGIBLE, SHELVED_PAIR, UNTAGGED, _escape, _limit
+from .prompt_archive import SHELVED
 from .library_commands import _TOKEN, LibraryCommandError, _validate_epoch
 from .library_tokens import canonical_token
 from .models import utc_now_iso
@@ -40,17 +41,21 @@ _IMAGE_FILES = (
 
 def _tag_rows(page: str, order: str = "page.id") -> str:
     """Count rows for the registry rows selected by ``page`` (id, name)."""
-    # Eligible parent components carrying a tag in the page, plus the
-    # positive parent of every active pair carrying it (always eligible).
+    # Eligible parent components on the shelf carrying a tag in the page,
+    # plus the positive parent of every active pair carrying it (always
+    # eligible) when that parent is on the shelf: archived parents (#443)
+    # are hidden with their pairs, so Browse lists what was counted.
     return (
         "WITH page AS (" + page + "),"
         "owners(tag_id,pid) AS ("
         "SELECT pt.tag_id,p.id FROM prompts p "
         "JOIN prompt_tags pt ON pt.prompt_id=p.id "
         "WHERE pt.tag_id IN (SELECT id FROM page) AND " + _ELIGIBLE + " "
+        "AND " + SHELVED + " "
         "UNION SELECT ppt.tag_id,pp.positive_prompt_id FROM prompt_pairs pp "
         "JOIN prompt_pair_tags ppt ON ppt.pair_id=pp.id "
-        "WHERE pp.archived=0 AND ppt.tag_id IN (SELECT id FROM page)),"
+        "WHERE pp.archived=0 AND " + SHELVED_PAIR + " "
+        "AND ppt.tag_id IN (SELECT id FROM page)),"
         "images(tag_id,n) AS (" + _IMAGE_FILES + " "
         "WHERE mit.tag_id IN (SELECT id FROM page) GROUP BY mit.tag_id) "
         "SELECT page.id,page.name,COUNT(DISTINCT owners.pid) AS prompt_count,"
@@ -61,7 +66,6 @@ def _tag_rows(page: str, order: str = "page.id") -> str:
     )
 
 
-_TAG_PAGE = _tag_rows("SELECT id,name FROM tags WHERE id>? ORDER BY id LIMIT ?")
 # #444 tag sorts: sort -> (key expression on tags/counted, direction). Ties and
 # "created" follow the id, so every order is total and keyset pages are exact.
 TAG_SORTS = {
@@ -79,8 +83,10 @@ _RANKED_TAG_PAGE = (
     "WITH owners(tag_id,pid) AS ("
     "SELECT pt.tag_id,p.id FROM prompts p "
     "JOIN prompt_tags pt ON pt.prompt_id=p.id WHERE " + _ELIGIBLE + " "
+    "AND " + SHELVED + " "
     "UNION SELECT ppt.tag_id,pp.positive_prompt_id FROM prompt_pairs pp "
-    "JOIN prompt_pair_tags ppt ON ppt.pair_id=pp.id WHERE pp.archived=0),"
+    "JOIN prompt_pair_tags ppt ON ppt.pair_id=pp.id WHERE pp.archived=0 "
+    "AND " + SHELVED_PAIR + "),"
     "counted AS (SELECT t.id,t.name,COUNT(DISTINCT o.pid) AS prompt_count "
     "FROM tags t LEFT JOIN owners o ON o.tag_id=t.id WHERE {search} "
     "GROUP BY t.id),"
@@ -94,7 +100,13 @@ _RANKED_TAG_PAGE = (
     "ORDER BY page.prompt_count {direction},page.id"
 )
 _UNTAGGED_COUNT = (
-    "SELECT COUNT(*) FROM prompts p WHERE (" + _ELIGIBLE + ") AND " + UNTAGGED
+    "SELECT COUNT(*) FROM prompts p WHERE ("
+    + _ELIGIBLE
+    + ") AND "
+    + UNTAGGED
+    # Untagged prompts opens the shelf, which hides archived parents (#443).
+    + " AND "
+    + SHELVED
 )
 _TAG_ONE = _tag_rows("SELECT id,name FROM tags WHERE id=?")
 # Every owner kind a tag can sit on, for the delete confirmation.
@@ -134,7 +146,9 @@ def _validate_tag_id(tag_id: object) -> None:
 
 
 # Category values in Tags & categories (#332 §7): the prompt-backed values
-# (nonblank values on eligible parents or on active pairs, grouped exactly, so
+# (nonblank values on eligible parents or on active pairs; ``prompt_count``
+# counts only parents on the shelf, so a value only archived parents (#443)
+# hold stays listed with 0, as a tag does; grouped exactly, so
 # whitespace and case variants stay distinct; /library/options lists the same
 # prompt-backed set with its own query) plus the values only image rows hold.
 # A prompt-backed value is ordered by its oldest owning positive; an image-only
@@ -143,15 +157,17 @@ def _validate_tag_id(tag_id: object) -> None:
 # files missing on disk. No FULL OUTER JOIN, so it runs on SQLite 3.31+.
 _IMAGE_ONLY_FIRST_PID = 2**63 - 1
 _CATEGORY_ROWS = (
-    "WITH owners(value,pid) AS ("
-    "SELECT p.category,p.id FROM prompts p WHERE p.category IS NOT NULL "
-    "AND TRIM(p.category)!='' AND " + _ELIGIBLE + " "
-    "UNION SELECT pp.category,pp.positive_prompt_id FROM prompt_pairs pp "
+    "WITH owners(value,pid,shelved) AS ("
+    "SELECT p.category,p.id," + SHELVED + " FROM prompts p "
+    "WHERE p.category IS NOT NULL AND TRIM(p.category)!='' AND " + _ELIGIBLE + " "
+    "UNION SELECT pp.category,pp.positive_prompt_id," + SHELVED_PAIR + " "
+    "FROM prompt_pairs pp "
     "WHERE pp.archived=0 AND pp.category IS NOT NULL AND TRIM(pp.category)!=''),"
     "image_values(value,image_count) AS ("
     "SELECT category,COUNT(*) FROM image_categories GROUP BY category),"
     "grouped(value,first_pid,prompt_count) AS ("
-    "SELECT value,MIN(pid),COUNT(DISTINCT pid) FROM owners GROUP BY value "
+    "SELECT value,MIN(pid),COUNT(DISTINCT CASE WHEN shelved THEN pid END) "
+    "FROM owners GROUP BY value "
     "UNION ALL SELECT value," + str(_IMAGE_ONLY_FIRST_PID) + ",0 FROM image_values "
     "WHERE value NOT IN (SELECT value FROM owners)),"
     "counted AS (SELECT g.value,g.first_pid,g.prompt_count,"

docs/prompt-send-bridge.md

@@ -18,9 +18,13 @@ protocol and peer checks reject unrelated messages. Same-origin scripts are
 trusted; this transport is not authentication against other scripts on that origin.
 Unsupported messaging is reported without breaking existing ComfyUI nodes.
 
+Apply works in one click: without a preview, the sender requests the ticket itself
+and applies it. Preview remains optional, and shows the exact change first.
+
 An apply timeout or abort after posting means the outcome is unknown. It cannot
 prove the host cancelled the write. The sender must inspect a fresh destination
-preview before deciding to send again; automatic retry must not overwrite edits.
+preview or choose the destination again before sending again; automatic retry must
+not overwrite edits. An error the host returned is a refusal: nothing was written.
 
 ## PromptManager adapter
 
@@ -31,7 +35,9 @@ field. Unknown/ambiguous metadata must not become an empty string.
 
 Previews show original and requested text plus positive prepend/append composition.
 Connected composition is marked unknown rather than guessing upstream values.
-Connected destination fields reject writes. Applying rechecks graph, node, widgets,
+Connected destination fields reject writes. The sender compares the destination's revision
+(node object and graph version) from discovery with the one at the preview, and refuses if the workflow changed.
+Applying rechecks graph, node, widgets,
 values and connections. Ordinary serialization cache synchronization is accepted
 only when it preserves live values. Both requested roles are checked before writing.
 

prompt_manager_pair.py

@@ -230,6 +230,8 @@ class PromptManagerPair(PromptManagerFocused):
             negative_text,
             category=category.strip() or None,
             tags=self._parse_tags(tags),
+            # The Library shows the prompt's own category and tags, not the pair's.
+            fill_new_positive=True,
         )
         pair = self.db.get_prompt_pair(pair_id)
         prompt_id = pair["positive_prompt_id"]

py/api/__init__.py

@@ -19,6 +19,7 @@ Split into domain-specific mixins for maintainability:
 - SystemRoutesMixin: System health check, Overview and usage statistics
 - LibraryLocationRoutesMixin: where the library lives, Move library, Keep it here
 - MaintenanceRoutesMixin: reviewed maintenance plans (preview, apply)
+- LibraryDeleteRoutesMixin: the Library's Delete… (archive paired) and Restore
 - WorkflowStorageRoutesMixin: store each workflow once, then Reclaim space
 - DuplicateFilesMixin: duplicate image files as find and removal jobs
 - LoraImportRoutesMixin: the reviewed, staged LoRA reference import job
@@ -40,8 +41,10 @@ from aiohttp import web
 from PIL import Image
 
 from .prompts import PromptRoutesMixin
+from .revalidation import revalidated_response
 from .pairs import PairRoutesMixin
 from .library import LibraryRoutesMixin
+from .library_delete import LibraryDeleteRoutesMixin
 from .vocabulary import VocabularyRoutesMixin
 from .annotations import AnnotationRoutesMixin
 from .review import ReviewRoutesMixin
@@ -75,12 +78,15 @@ from .lora_integration import LoraIntegrationMixin
 from .lora_import_routes import LoraImportRoutesMixin
 from .image_categories import ImageCategoryRoutesMixin
 from .image_delete import ImageDeleteMixin
+from .image_return import ImageReturnMixin
 from .source_rescope import SourceRescopeMixin
 
 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
@@ -91,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
 
 
@@ -314,6 +322,7 @@ async def _gzip_middleware(request, handler):
 class PromptManagerAPI(
     PairRoutesMixin,
     LibraryRoutesMixin,
+    LibraryDeleteRoutesMixin,
     VocabularyRoutesMixin,
     AnnotationRoutesMixin,
     ReviewRoutesMixin,
@@ -345,6 +354,7 @@ class PromptManagerAPI(
     LoraImportRoutesMixin,
     ImageCategoryRoutesMixin,
     ImageDeleteMixin,
+    ImageReturnMixin,
     SourceRescopeMixin,
 ):
     """REST API handler for PromptManager operations and web interface.
@@ -520,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())
@@ -661,40 +679,45 @@ class PromptManagerAPI(
 
         @routes.get("/prompt_manager/workspace")
         async def serve_workspace_ui(request: web.Request) -> web.Response:
-            return self._serve_html_page("workspace.html", "Workspace")
+            return self._serve_html_page(request, "workspace.html", "Workspace")
 
         @routes.get("/prompt_manager/web")
         async def serve_web_ui(request):
-            return self._serve_html_page("index.html", "Web UI", cache=False)
+            return self._serve_html_page(request, "index.html", "Web UI", cache=False)
 
         @routes.get("/prompt_manager/gallery.html")
         async def serve_gallery_ui(request):
-            return self._serve_html_page("metadata.html", "Gallery")
+            return self._serve_html_page(request, "metadata.html", "Gallery")
 
         @routes.get("/prompt_manager/admin")
         async def serve_admin_ui(request):
-            return self._serve_html_page("admin.html", "Admin UI")
+            return self._serve_html_page(request, "admin.html", "Admin UI")
 
         @routes.get("/prompt_manager/gallery")
         async def serve_gallery_admin_ui(request):
-            return self._serve_html_page("gallery.html", "Gallery")
+            return self._serve_html_page(request, "gallery.html", "Gallery")
 
         # ── Static file serving ───────────────────────────────────────
 
         @routes.get("/prompt_manager/lib/{filepath:.*}")
         async def serve_lib_static(request):
             """Serve static library files (JS, CSS) from web/lib directory."""
-            return self._serve_static_file("lib", request.match_info.get("filepath"))
+            return self._serve_static_file(
+                request, "lib", request.match_info.get("filepath")
+            )
 
         @routes.get("/prompt_manager/js/{filepath:.*}")
         async def serve_js_static(request):
             """Serve static JavaScript files from web/js directory."""
-            return self._serve_static_file("js", request.match_info.get("filepath"))
+            return self._serve_static_file(
+                request, "js", request.match_info.get("filepath")
+            )
 
         # ── Register domain-specific routes from mixins ───────────────
 
         self._register_pair_routes(routes)
         self._register_library_routes(routes)
+        self._register_library_delete_routes(routes)
         self._register_vocabulary_routes(routes)
         self._register_annotation_routes(routes)
         self._register_review_routes(routes)
@@ -719,6 +742,7 @@ class PromptManagerAPI(
         self._register_lora_import_routes(routes)
         self._register_image_category_routes(routes)
         self._register_image_delete_routes(routes)
+        self._register_image_return_routes(routes)
         self._register_rescope_routes(routes)
 
         # Register gzip compression middleware (once)
@@ -735,8 +759,11 @@ class PromptManagerAPI(
 
         self.logger.info("All routes registered with decorator pattern")
 
-    def _serve_static_file(self, subdir, filepath):
-        """Serve ``web/<subdir>/<filepath>``; 403 outside it, 404 when missing."""
+    def _serve_static_file(self, request, subdir, filepath):
+        """Serve ``web/<subdir>/<filepath>``; 403 outside it, 404 when missing.
+
+        A served file is revalidated on every reuse (#478, ``revalidation``).
+        """
         root = os.path.join(_get_project_root(), "web", subdir)
         exact = STATIC_EXACT_FILES.get((subdir, filepath))
         # An exact file passes the same containment check, its extension only.
@@ -751,10 +778,13 @@ class PromptManagerAPI(
             return web.Response(text="Not Found", status=404)
         with open(file_path, "rb") as f:
             content = f.read()
-        return web.Response(body=content, content_type=content_type)
+        return revalidated_response(request, content, content_type)
 
-    def _serve_html_page(self, filename, title, cache=True):
-        """Serve ``web/<filename>``; 404 when missing, 500 when unreadable."""
+    def _serve_html_page(self, request, filename, title, cache=True):
+        """Serve ``web/<filename>``; 404 when missing, 500 when unreadable.
+
+        A served page is revalidated on every reuse (#478, ``revalidation``).
+        """
         html_path = os.path.join(_get_project_root(), "web", filename)
         try:
             if cache and html_path in self._html_cache:
@@ -771,7 +801,9 @@ class PromptManagerAPI(
                     content_type="text/html",
                     status=404,
                 )
-            return web.Response(text=content, content_type="text/html", charset="utf-8")
+            return revalidated_response(
+                request, content.encode("utf-8"), "text/html", charset="utf-8"
+            )
         except Exception:
             self.logger.exception(f"Failed to load {title}")
             return web.Response(
@@ -1028,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.
@@ -1036,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}"

py/api/admin.py

@@ -36,10 +36,21 @@ except ImportError:
 
 try:
     from ...utils import exit_signal
+    from ...utils.media_folders import (
+        in_trash_below_roots,
+        in_trash_folder,
+        is_skipped_folder,
+    )
 except ImportError:
     from utils import exit_signal
+    from utils.media_folders import (
+        in_trash_below_roots,
+        in_trash_folder,
+        is_skipped_folder,
+    )
 
 try:
+    from ...database.image_removal import mark_hides
     from ...database.pair_ingestion import SCAN_MARKS, ingest_resolved_pair
     from ...database.run_attribution import (
         RecordedOutput,
@@ -60,6 +71,7 @@ try:
         decide_file_prompt,
     )
 except ImportError:
+    from database.image_removal import mark_hides
     from database.pair_ingestion import SCAN_MARKS, ingest_resolved_pair
     from database.run_attribution import (
         RecordedOutput,
@@ -118,10 +130,11 @@ DUPLICATE_FIND_LEGACY_TITLE = "Find duplicate image files (gallery page)"
 def _collect_media_files(output_dirs, extensions):
     """Media files under ``output_dirs`` (blocking), bounded and symlink-free.
 
-    Walks each root without following symlinks, skips ``thumbnails``
-    directories and symlinked files, never descends more than MAX_SCAN_DEPTH
-    levels and stops after MAX_SCAN_FILES files in total. Extensions match
-    case-insensitively and each path is listed once.
+    Walks each root without following symlinks, skips ``thumbnails`` and
+    trash folders (``utils.media_folders``) and symlinked files, never
+    descends more than MAX_SCAN_DEPTH levels and stops after MAX_SCAN_FILES
+    files in total. Extensions match case-insensitively and each path is
+    listed once.
     """
     wanted = {ext.lower() for ext in extensions}
     found = []
@@ -136,7 +149,7 @@ def _collect_media_files(output_dirs, extensions):
             dirnames[:] = sorted(
                 d
                 for d in dirnames
-                if d != "thumbnails"
+                if not is_skipped_folder(d)
                 and depth < MAX_SCAN_DEPTH
                 and not os.path.islink(os.path.join(current, d))
             )
@@ -163,7 +176,7 @@ def _physical_file_key(path, stat):
     volumes on Windows) falls back to the resolved, case-folded path.
     """
     if stat.st_ino:
-        return ("inode", stat.st_dev, stat.st_ino)
+        return ("inode", (stat.st_dev, stat.st_ino))
     return ("path", os.path.normcase(os.path.realpath(str(path))))
 
 
@@ -236,7 +249,12 @@ SCAN_FINAL_STATES = {
     "error": "failed",
 }
 # The scope as the scan log's first line names it (the review dialog's words).
-SCAN_SCOPE_LOG = {"new": "new and unresolved files", "all": "all files"}
+SCAN_SCOPE_LOG = {
+    "new": "new and unresolved files",
+    "all": "all files",
+    "returned": "images returned to the library",
+}
+RETURN_JOB_TITLE = "Return images to the library"  # #498
 MIN_BATCHES_FOR_ESTIMATE = 2  # "about X min left" waits for this many batches
 # Log line prefix per outcome group (spec: + added, = known, ? review, ! error,
 # - no usable metadata).
@@ -521,6 +539,7 @@ def _diagnose_output_dirs(configured_dirs, public_path=None):
 
         candidates.append(folder_paths.get_output_directory())
     except ImportError:
+        # outside ComfyUI there is no output folder to add
         pass
 
     output_dirs = []
@@ -893,11 +912,8 @@ class AdminRoutesMixin:
                     stat = os.stat(path)
                 except OSError:
                     continue
-                mark = marks.get(os.path.normcase(os.path.realpath(str(path))))
-                if mark is not None and tuple(mark) == (
-                    stat.st_size,
-                    stat.st_mtime_ns,
-                ):
+                locator = os.path.normcase(os.path.realpath(str(path)))
+                if locator in marks and mark_hides(marks, locator, path):
                     continue
                 key = _physical_file_key(path, stat)
                 if key not in seen:
@@ -1082,13 +1098,21 @@ class AdminRoutesMixin:
         return result
 
     def _delete_one_duplicate(self, image_path, output_path):
-        """Delete one file and its thumbnail; return a failure reason or None."""
+        """Delete one file and its thumbnail; return a failure reason or None.
+
+        What is removed is exactly the folder's real path plus the name: a
+        link is removed itself, never its target, and the preview removed is
+        that name's. Both the link's place and what it leads to are checked.
+        """
         file_path = Path(image_path)
         if not file_path.is_absolute():
             file_path = output_path / file_path
 
         try:
-            rel_path = file_path.resolve().relative_to(output_path.resolve())
+            real_root = output_path.resolve()
+            rel_path = file_path.resolve().relative_to(real_root)
+            exact = Path(os.path.realpath(file_path.parent)) / file_path.name
+            exact_rel = exact.relative_to(real_root)
         except ValueError:
             self.logger.warning(
                 f"Attempted to delete file outside output directory: {image_path}"
@@ -1097,15 +1121,19 @@ class AdminRoutesMixin:
 
         if file_path.suffix.lower() not in DELETABLE_MEDIA_EXTENSIONS:
             return "not a media file"
-        if "thumbnails" in rel_path.parts:
+        if "thumbnails" in rel_path.parts or "thumbnails" in exact_rel.parts:
             return "thumbnails are managed by the gallery"
+        if in_trash_below_roots(file_path, [output_path]) or in_trash_folder(
+            str(exact_rel)
+        ):
+            return "it is in a trash folder"  # never unlinked from a trash (#543)
 
         if not file_path.is_file():
             return "file not found"
 
-        os.remove(file_path)
+        os.remove(exact)
         self.logger.info(f"Deleted duplicate image: {image_path}")
-        _remove_thumbnail(output_path.resolve(), rel_path, self.logger)
+        _remove_thumbnail(real_root, exact_rel, self.logger)
         return None
 
     async def get_statistics(self, request):
@@ -1379,13 +1407,17 @@ class AdminRoutesMixin:
             }
         )
 
-    def _start_scan_job(self, files=DEFAULT_SCAN_SCOPE, roots=None, output_dirs=None):
+    def _start_scan_job(
+        self, files=DEFAULT_SCAN_SCOPE, roots=None, output_dirs=None, returned=None
+    ):
         """Register the scan job synchronously; its runner starts on the loop.
 
         ``files`` is the scope (``SCAN_SCOPES``). ``roots`` selects sources by
         their index in ``output_dirs`` (the effective sources the caller
         resolved); None scans every source, resolved when the job starts.
-        Raises JobBusy while another record-changing job runs.
+        Raises JobBusy while another record-changing job runs. ``returned``
+        (#498, ``[(path, locator)]``) makes it "Return images to the library":
+        those files' removal marks go first, then only they are scanned.
         """
         if files not in SCAN_SCOPES:
             raise ValueError(f"Unknown scan scope: {files!r}")
@@ -1393,20 +1425,26 @@ class AdminRoutesMixin:
             output_dirs is None or not roots or max(roots) >= len(output_dirs)
         ):
             raise ValueError("A scoped scan needs the sources its roots index")
+        scope = {"files": files, "roots": None if roots is None else list(roots)}
+        if returned is not None:
+            scope = {"files": "returned", "roots": None, "images": len(returned)}
         job = self._jobs.start(
             SCAN_JOB_KIND,
-            SCAN_JOB_TITLE,
+            SCAN_JOB_TITLE if returned is None else RETURN_JOB_TITLE,
             record_changing=True,
-            scope={"files": files, "roots": None if roots is None else list(roots)},
-            runner=functools.partial(self._run_scan_job, output_dirs=output_dirs),
+            scope=scope,
+            runner=functools.partial(
+                self._run_scan_job, output_dirs=output_dirs, returned=returned
+            ),
             groups=SCAN_GROUP_NAMES,
         )
         self._scan_job = job
         return job
 
-    async def _run_scan_job(self, job, output_dirs=None):
+    async def _run_scan_job(self, job, output_dirs=None, returned=None):
         final = None
-        async for payload in self._scan_images_events(job, output_dirs):
+        extra = {} if returned is None else {"returned": returned}
+        async for payload in self._scan_images_events(job, output_dirs, **extra):
             job.publish(payload)
             final = payload.get("type")
         state = SCAN_FINAL_STATES.get(final, "finished")
@@ -1416,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)
@@ -1444,7 +1483,7 @@ class AdminRoutesMixin:
             )
         return response
 
-    async def _scan_images_events(self, job, output_dirs=None):
+    async def _scan_images_events(self, job, output_dirs=None, returned=None):
         """Yield progress payloads while scanning the output directories.
 
         Every effective source is walked, or only those ``job.scope["roots"]``
@@ -1482,9 +1521,16 @@ class AdminRoutesMixin:
                 0, f"Scanning {len(walked)} directory(ies) for media files..."
             )
 
-            media_files = await self._run_in_executor(
-                self._collect_output_media_sync, walked
-            )
+            if returned is None:
+                media_files = await self._run_in_executor(
+                    self._collect_output_media_sync, walked
+                )
+            else:
+                # #498: only the files whose marks this job deleted.
+                media_files = await self._run_in_executor(
+                    self._clear_returned_marks, returned
+                )
+                job.log_line(f"Returned {_plural(len(media_files), 'image')}")
             counts = dict.fromkeys(SCAN_COUNT_KEYS, 0)
             job.update(
                 total=len(media_files),
@@ -1623,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
 

py/api/annotation_suggest.py

@@ -13,7 +13,6 @@ import math
 from ..tagging_defaults import (  # noqa: F401 - re-exported for the route
     LLM_MODEL_TYPES as LLM_MODELS,
     MAX_INSTRUCTION_BYTES,
-    MAX_INSTRUCTION_CHARS,
     MODEL_TYPES,
     WD14_MODEL_TYPES as WD14_MODELS,
     validate_template,
@@ -103,11 +102,14 @@ def model_refusal(status: dict, model_type: str, model_name: str):
     return None
 
 
-def run_suggestions(service, capture_path: str, model_type: str, options: dict):
+def run_suggestions(
+    service, capture_path: str, model_type: str, options: dict, skip=None
+):
     """Blocking: load (or swap to) *model_type*, then suggest for the capture.
 
     The caller holds the model gate for the whole call; nothing is ever
-    downloaded (``load_model`` refuses a model that is not on disk).
+    downloaded (``load_model`` refuses a model that is not on disk). ``skip``
+    is the request's or job's compiled Skipped tags (#431).
     """
     if not service.is_model_loaded() or service.get_loaded_model_type() != model_type:
         service.load_model(model_type, options["use_gpu"])
@@ -117,6 +119,7 @@ def run_suggestions(service, capture_path: str, model_type: str, options: dict):
         prompt=options.get("instructions"),
         general_threshold=options.get("general_threshold"),
         character_threshold=options.get("character_threshold"),
+        skip=skip,
     )
 
 

py/api/annotations.py

@@ -35,6 +35,7 @@ from .media_target import media_target as _target
 from .images import _unavailable_link, _verified_title
 from .prompts import bad_request
 from .. import model_gate
+from ..tagging_defaults import SkipMatcher
 
 try:
     from ...database.annotations import (
@@ -640,6 +641,8 @@ class AnnotationRoutesMixin:
             owner = model_gate.new_owner("image-annotations")
             if not model_gate.try_acquire(owner):
                 return model_busy_response()
+            # #431: the Skipped tags as they are when the request is accepted.
+            skip = SkipMatcher(AutotagConfig.SKIPPED_TAGS).counting()
             try:
                 # From here the pin is the gate's to end: the worker ends it,
                 # or ``on_skipped`` does when the worker never starts.
@@ -653,6 +656,7 @@ class AnnotationRoutesMixin:
                     capture,
                     model_type,
                     options,
+                    skip,
                     on_skipped=functools.partial(store.release_use, token),
                 )
             except MissingBackendPackages as exc:
@@ -677,6 +681,8 @@ class AnnotationRoutesMixin:
                 "model_type": model_type,
                 "options": options,
                 "suggestions": suggestion_rows(pairs),
+                # How many distinct suggestions the Skipped tags hid (#431).
+                "skipped": len(skip.hidden),
             },
             headers={"Cache-Control": "no-store"},
         )
@@ -732,10 +738,10 @@ def _byte_range(request: web.Request, size: int):
     return start, stop
 
 
-def _pinned_suggestions(store, token, service, capture, model_type, options):
+def _pinned_suggestions(store, token, service, capture, model_type, options, skip=None):
     """Worker thread: infer on the capture, then end its pin (even on error)."""
     try:
-        return run_suggestions(service, capture.path, model_type, options)
+        return run_suggestions(service, capture.path, model_type, options, skip)
     finally:
         store.release_use(token)
 

py/api/autotag_routes.py

@@ -23,6 +23,11 @@ from .prompts import thumbnail_url_for
 from . import thumbnail_settings
 from .. import model_gate
 
+try:
+    from ...utils.media_folders import in_trash_below_roots, is_skipped_folder
+except ImportError:
+    from utils.media_folders import in_trash_below_roots, is_skipped_folder
+
 from aiohttp import web
 
 _INTERNAL_ERROR = "An internal error occurred. Check server logs for details."
@@ -35,6 +40,12 @@ BATCH_PREVIEW_BODY_MAX = 1024
 APPLY_BODY_MAX = 64 * 1024  # POST /autotag/apply: one image and its tags
 _HEX64 = re.compile(r"[0-9a-f]{64}")
 BATCH_JOB_BODY_MAX = 32 * 1024
+# Skipped tags preview (#431): 200 entries of 50 characters, with room for JSON.
+SKIPPED_PREVIEW_BODY_MAX = 64 * 1024
+SKIPPED_PREVIEW_LEASE = "Skipped tags preview"
+SKIPPED_PREVIEW_SAMPLE = 12
+# WD14 label lists the preview found missing or unreadable (logged once each).
+_MISSING_LABEL_LISTS = set()
 REVIEW_REQUIRED = (
     "Replacing tags needs a review of each image. Use Preview tagging review, "
     "or choose Skip or Merge for a batch run."
@@ -67,6 +78,12 @@ MODEL_DOWNLOAD_BODY_MAX = 1024
 _download_titles = {}
 
 
+def _instruction_engine(model_type: str) -> str:
+    """The language-model engine whose default instructions ``model_type``
+    reads: itself, or GGUF for a WD14 engine (which takes none)."""
+    return model_type if model_type in ("gguf", "hf") else "gguf"
+
+
 def _holder_label(holder: Optional[str]) -> str:
     if not holder:
         return ANOTHER_OPERATION
@@ -242,6 +259,10 @@ class AutotagRoutesMixin:
         async def preview_autotag_batch_route(request):
             return await self.preview_autotag_batch(request)
 
+        @routes.post("/prompt_manager/autotag/skipped-tags/preview")
+        async def preview_skipped_tags_route(request):
+            return await self.preview_skipped_tags(request)
+
         @routes.post("/prompt_manager/autotag/start")
         async def start_autotag_route(request):
             return await self.start_autotag(request)
@@ -279,7 +300,15 @@ class AutotagRoutesMixin:
                     "success": True,
                     "models": public_models(status, loaded_type),
                     # The saved tagging defaults (#328, System › Auto-tagging).
-                    "default_prompt": AutotagConfig.TEMPLATE,
+                    # The saved engine's default instructions (GGUF for a
+                    # WD14 engine); kept one release for cached pages (#431).
+                    "default_prompt": AutotagConfig.default_instructions_text(
+                        _instruction_engine(AutotagConfig.MODEL_TYPE)
+                    ),
+                    # #431: Built-in first, with every text, and each
+                    # language-model engine's default.
+                    "instructions": AutotagConfig.all_instructions(),
+                    "default_instructions": dict(AutotagConfig.DEFAULT_INSTRUCTIONS),
                     "model_loaded": loaded,
                     "loaded_model_type": loaded_type,
                     "wd14_general_threshold": AutotagConfig.GENERAL_THRESHOLD,
@@ -419,6 +448,7 @@ class AutotagRoutesMixin:
         title = f"Tagging model download · {label}"
         _download_titles.clear()
         _download_titles[owner] = title
+        admitted = False
         try:
             job = self._jobs.start(
                 MODEL_DOWNLOAD_JOB_KIND,
@@ -434,13 +464,13 @@ class AutotagRoutesMixin:
                     self._run_model_download, service, model_type, owner
                 ),
             )
-        except BaseException as refused:
-            # JobBusy, or anything else the registry raised: give the gate back.
-            model_gate.release(owner)
-            _download_titles.pop(owner, None)
-            if isinstance(refused, JobBusy):
-                return None, busy_response(refused)
-            raise
+            admitted = True
+        except JobBusy as refused:
+            return None, busy_response(refused)
+        finally:
+            if not admitted:  # JobBusy, or anything else: give the gate back.
+                model_gate.release(owner)
+                _download_titles.pop(owner, None)
 
         def ended(_task):
             model_gate.release(owner)
@@ -533,7 +563,10 @@ class AutotagRoutesMixin:
         else:
             prompt = params.get("prompt")
             template = prompt if isinstance(prompt, str) and prompt.strip() else None
-            options = {"instructions": template or AutotagConfig.TEMPLATE}
+            options = {
+                "instructions": template
+                or AutotagConfig.default_instructions_text(model_type)
+            }
         try:
             _model, options = validate_options(
                 model_type,
@@ -693,6 +726,86 @@ class AutotagRoutesMixin:
             return None, "existing must be one of " + ", ".join(EXISTING_POLICIES)
         return MODELS[model_type], None
 
+    async def preview_skipped_tags(self, request):
+        """Which known tags a Skipped tags list would hide (#431); no write.
+
+        ``{entries}`` is checked as a save checks it (the same 400 sentence).
+        The known tags are the library's plus the general and character names
+        of each downloaded WD14 label list; no model is loaded and no image is
+        opened. Refused (409 ``busy``) while an exclusive job runs.
+        """
+        from ..tagging_defaults import validate_skipped_tags
+
+        data, error = await self._batch_body(
+            request, ("entries",), SKIPPED_PREVIEW_BODY_MAX
+        )
+        if error is not None:
+            return error
+        try:
+            entries = validate_skipped_tags(data["entries"])
+        except ValueError as exc:
+            return web.json_response({"success": False, "error": str(exc)}, status=400)
+        try:
+            async with self._jobs.lease(SKIPPED_PREVIEW_LEASE):
+                result = await self._run_in_executor(self._skipped_preview, entries)
+        except JobBusy as busy:
+            return busy_response(busy)
+        except Exception:
+            self.logger.exception("The Skipped tags preview failed")
+            return web.json_response(
+                {"success": False, "error": _INTERNAL_ERROR}, status=500
+            )
+        return web.json_response({"success": True, **result})
+
+    def _skipped_preview(self, entries):
+        """Worker thread: the preview's counts over the known tags."""
+        from ..autotag import MODELS, get_autotag_service
+        from ..tagging_defaults import (
+            WD14_MODEL_TYPES,
+            SkipMatcher,
+            wd14_label_names,
+        )
+
+        library = self.db.get_all_tags()
+        known = {}  # case-folded -> (0 library / 1 label list, spelling)
+        for name in library:
+            known.setdefault(name.casefold(), (0, name))
+        lists = []
+        models_dir = getattr(get_autotag_service(), "models_dir", None)
+        for model_type in sorted(WD14_MODEL_TYPES, key=lambda t: MODELS[t]["name"]):
+            if models_dir is None:
+                break
+            path = Path(models_dir) / MODELS[model_type]["subdir"] / "selected_tags.csv"
+            try:
+                names = wd14_label_names(path)
+            except (OSError, ValueError) as exc:
+                if path not in _MISSING_LABEL_LISTS:  # logged once per list
+                    _MISSING_LABEL_LISTS.add(path)
+                    self.logger.debug(f"No WD14 label list for {model_type}: {exc}")
+                continue
+            _MISSING_LABEL_LISTS.discard(path)
+            lists.append(MODELS[model_type]["name"])
+            for name in names:
+                known.setdefault(name.casefold(), (1, name))
+        matchers = [(entry, SkipMatcher([entry])) for entry in entries]
+        whole = SkipMatcher(entries)
+        hidden = [item for item in known.values() if whole.matches(item[1])]
+        hidden.sort(key=lambda item: (item[0], item[1].casefold()))
+        per_entry = [
+            {
+                "entry": entry,
+                "count": sum(1 for _, name in hidden if matcher.matches(name)),
+            }
+            for entry, matcher in matchers
+        ]
+        return {
+            "count": len(hidden),
+            "sample": [name for _, name in hidden[:SKIPPED_PREVIEW_SAMPLE]],
+            "sample_limit": SKIPPED_PREVIEW_SAMPLE,
+            "per_entry": per_entry,
+            "sources": {"library": len(library), "wd14": lists},
+        }
+
     async def preview_autotag_batch(self, request):
         """Review Tag output images: counts, model, backup and disk; no write.
 
@@ -844,9 +957,12 @@ class AutotagRoutesMixin:
         """Return the real path when it is a media file inside an output directory."""
         if Path(str(raw_path)).suffix.lower() not in _MEDIA_EXTENSIONS:
             return None
-        for output_dir in self._get_all_output_dirs():
+        roots = list(self._get_all_output_dirs())
+        for output_dir in roots:
             if path_is_within(raw_path, output_dir):
                 real = os.path.realpath(raw_path)
+                if in_trash_below_roots(real, roots):
+                    return None  # thrown away, not an output (#543)
                 return real if os.path.isfile(real) else None
         return None
 
@@ -993,7 +1109,10 @@ class AutotagRoutesMixin:
         else:
             prompt = data.get("prompt")
             template = prompt if isinstance(prompt, str) and prompt.strip() else None
-            options = {"instructions": template or AutotagConfig.TEMPLATE}
+            options = {
+                "instructions": template
+                or AutotagConfig.default_instructions_text(model_type)
+            }
         use_gpu = data.get("use_gpu", True)
         options["use_gpu"] = use_gpu if isinstance(use_gpu, bool) else True
         return validate_options(
@@ -1033,12 +1152,12 @@ class AutotagRoutesMixin:
                 return self._bad_request('existing must be "skip" or "merge"')
             try:
                 options = self._single_options(data, model_type)
+            except OptionError as error:
+                return self._bad_request(str(error))
             except (ValueError, TypeError):
                 return self._bad_request(
                     "general_threshold and character_threshold must be numeric"
                 )
-            except OptionError as error:
-                return self._bad_request(str(error))
             target, error = await self._select_autotag_target(data)
             if error is not None:
                 return error
@@ -1087,6 +1206,8 @@ class AutotagRoutesMixin:
 
     async def _suggest_single(self, target, model_type, options, base):
         from ..autotag import MODELS, MissingBackendPackages, get_autotag_service
+        from ..config import AutotagConfig
+        from ..tagging_defaults import SkipMatcher
         from .annotation_suggest import model_refusal
         from .annotations import _pinned_suggestions, _settled, _submit
 
@@ -1103,6 +1224,8 @@ class AutotagRoutesMixin:
         owner = model_gate.new_owner("legacy-single")
         if not model_gate.try_acquire(owner):
             return model_busy_response()
+        # #431: the legacy review hides the Skipped tags too (19A).
+        skip = SkipMatcher(AutotagConfig.SKIPPED_TAGS)
         store = self._capture_store()
         try:
             epoch = await self._run_in_executor(self.db.get_execution_epoch)
@@ -1142,6 +1265,7 @@ class AutotagRoutesMixin:
                     capture,
                     model_type,
                     options,
+                    skip,
                     on_skipped=functools.partial(store.release_use, capture.token),
                 )
             except MissingBackendPackages as exc:
@@ -1319,8 +1443,9 @@ class AutotagRoutesMixin:
     def _collect_scan_files(output_path: Path):
         """Media files under ``output_path`` from a bounded, symlink-free walk.
 
-        Descends at most ``_SCAN_MAX_DEPTH`` levels, skips ``thumbnails``
-        trees and symlinked files, and stops after ``_SCAN_MAX_FILES`` files.
+        Descends at most ``_SCAN_MAX_DEPTH`` levels, skips ``thumbnails`` and
+        trash folders (``utils.media_folders``) and symlinked files, and stops
+        after ``_SCAN_MAX_FILES`` files.
         Returns ``(paths, truncated)``.
         """
         found = []
@@ -1330,7 +1455,7 @@ class AutotagRoutesMixin:
             dirnames[:] = sorted(
                 name
                 for name in dirnames
-                if name != "thumbnails" and depth + 1 <= _SCAN_MAX_DEPTH
+                if not is_skipped_folder(name) and depth + 1 <= _SCAN_MAX_DEPTH
             )
             for name in sorted(filenames):
                 if Path(name).suffix.lower() not in _MEDIA_EXTENSIONS:
@@ -1357,6 +1482,10 @@ class AutotagRoutesMixin:
             files, truncated = await self._run_in_executor(
                 self._collect_scan_files, output_path
             )
+            # #501: a file removed from the library is not offered for tagging.
+            files = await self._run_in_executor(
+                self._without_removed, output_path, files
+            )
             has_thumbnails = (output_path / "thumbnails").is_dir()
 
             images = []

py/api/backup_restore.py

@@ -971,6 +971,7 @@ class BackupRestoreRoutesMixin:
         line = restored_line(staged.summary)
         job.log_line(line)
         job.update(progress=100, status_line=line)
+        return None
 
 
 async def _outlive_cancel(future):

py/api/duplicate_files.py

@@ -574,12 +574,13 @@ class DuplicateFilesMixin:
             return _fail_job(job, "sources-changed", SOURCES_CHANGED)
         ok, backup = await self._safety_backup_step(job, db, backup_on)
         if not ok:
-            return
+            return None
         try:
             await self._remove_duplicates(job, db, plan, backup)
         finally:
             if backup:
                 await self._prune_backup_pool(job, db)
+        return None
 
     async def _remove_duplicates(self, job, db, plan, backup):
         targets = plan.payload["targets"]

py/api/duplicate_review.py

@@ -16,6 +16,11 @@ from pathlib import Path
 from .admin import DELETABLE_MEDIA_EXTENSIONS
 from .maintenance import DUPLICATE_FILES
 
+try:
+    from ...utils.media_folders import in_trash_below_roots, is_trash_folder
+except ImportError:
+    from utils.media_folders import in_trash_below_roots, is_trash_folder
+
 # Keep entries one review takes; with paths of a few hundred characters the
 # list stays inside the preview body cap (DUPLICATE_PREVIEW_BODY_MAX, 1 MiB).
 MAX_KEEP = 2000
@@ -197,6 +202,8 @@ def file_problem(
     parts = Path(os.path.relpath(lexical, base)).parts
     if "thumbnails" in parts or ".." in parts:
         return OUTSIDE_SOURCE
+    if any(is_trash_folder(part) for part in parts[:-1]):
+        return OUTSIDE_SOURCE  # never removed from a trash folder (#543)
     if os.path.splitext(lexical)[1].lower() not in DELETABLE_MEDIA_EXTENSIONS:
         return OUTSIDE_SOURCE
     try:
@@ -212,6 +219,8 @@ def file_problem(
     try:
         if not _lexically_inside(_canonical_path(path), [_canonical_path(root)]):
             return OUTSIDE_SOURCE  # a linked folder on the way
+        if in_trash_below_roots(path, [root]):
+            return OUTSIDE_SOURCE  # a linked folder into a trash folder
         if keeper is not None:
             same = keeper_problem(path, root, info, keeper, keeper_root)
             if same:

py/api/duplicate_spellings.py

@@ -27,7 +27,7 @@ def _identity(path):
     except OSError:
         return None
     if info.st_ino:
-        return ("inode", info.st_dev, info.st_ino)
+        return ("inode", (info.st_dev, info.st_ino))
     return ("path", os.path.normcase(os.path.realpath(str(path))))
 
 

py/api/history.py

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

py/api/history_current.py

@@ -17,6 +17,7 @@ from aiohttp import web
 
 from . import thumbnail_settings
 from .images import _canonical, _public_path, SERVABLE_EXTENSIONS, VIDEO_EXTENSIONS
+from .images import in_trash_below_roots
 from .images import existing_media_thumbnail, media_thumbnail_rel
 
 try:
@@ -198,6 +199,9 @@ def resolve_current_candidate(
         if namespace is None or not namespace.is_dir():
             return _unavailable("root_unavailable")
         target = _canonical(lexical)
+        if in_trash_below_roots(target, [namespace]):
+            # Every kind and the outputs listing come through here (#543).
+            return _unavailable("missing")
         if not _authorized(target, namespace, scope):
             return _unavailable("not_authorized")
         if target.suffix.lower() not in SERVABLE_EXTENSIONS:
@@ -248,6 +252,10 @@ def _open_current_target(
     A *preview* (a file under ``thumbnails/``) may also be JPEG XL (#317)."""
     if _canonical(target) != target:
         raise CurrentLocationUnavailable("not_authorized", stale=True)
+    if in_trash_below_roots(target, [namespace]):
+        # Also at the open: the cover chooses its preview after the candidate
+        # check, and a preview can be a link into a trash folder (#543).
+        raise CurrentLocationUnavailable("missing")
     if not _authorized(target, namespace, scope):
         raise CurrentLocationUnavailable("not_authorized")
     suffix = target.suffix.lower()

py/api/image_categories.py

@@ -232,15 +232,17 @@ class ImageCategoryRoutesMixin:
             return None, _coded(400, "too-many-images", TOO_MANY)
         return tuple(selection.targets), None
 
-    async def _category_request(self, request, fields: frozenset):
+    async def _category_request(self, request, fields: frozenset, data=None):
         """``(data, targets, roots, None)`` or ``(None, None, None, error)``.
 
         The images are ``items`` (1 to 500 targets) or ``filter`` (all images
-        an Images listing filter admits), besides *fields*.
+        an Images listing filter admits), besides *fields*. *data* is the body
+        when the caller has already read it (a body is read once).
         """
-        data, error = await _json_body_or_error(request)
-        if error is not None:
-            return None, None, None, error
+        if data is None:
+            data, error = await _json_body_or_error(request)
+            if error is not None:
+                return None, None, None, error
         names = set(data) - fields
         if set(data) & fields != fields or names not in ({"items"}, {"filter"}):
             return (

py/api/image_delete.py

@@ -47,12 +47,16 @@ from .maintenance import MAINTENANCE_PREVIEW_LEASE, _fail_job, _failure, backup_
 
 try:
     from ...database import image_removal_schema
+    from ...database.image_removal import mark_hides
     from ...database.annotation_schema import MAX_LOCATOR_LENGTH
     from ...utils import trash
+    from ...utils.media_folders import is_trash_folder
 except ImportError:
     from database import image_removal_schema
+    from database.image_removal import mark_hides
     from database.annotation_schema import MAX_LOCATOR_LENGTH
     from utils import trash
+    from utils.media_folders import is_trash_folder
 
 IMAGE_DELETE = "delete_images"
 MODES = ("library", "trash")
@@ -61,7 +65,9 @@ TITLES = {
     "trash": "Delete images (move to the trash)",
 }
 MAX_DETAIL_ROWS = 50
-TRASH_FOLDERS = frozenset({".trash", ".trashes", "$recycle.bin"})
+# The system moved the file but its place in a trash could not be shown (#499).
+TRASH_UNCONFIRMED = "trash-unconfirmed"
+RECORDS_FAILED_CODE = "records-failed"
 
 # Reason codes (design §4.4) and the sentence each shows.
 REASONS = {
@@ -80,6 +86,21 @@ REASONS = {
     trash.TRASH_IN_ROOT: "This drive's trash is inside an output folder.",
     trash.LOCKED: "The file is open in another program. Close it and try again.",
     trash.FAILED: "The file could not be moved to the trash.",
+    trash.NO_PERMISSION: (
+        "PromptManager isn't allowed to move this file (no permission, or it is "
+        "locked in Finder)."
+    ),
+    trash.CLOUD_SYNC: (
+        "Files in iCloud Drive or a cloud-synced folder can't be moved to the "
+        "trash here."
+    ),
+    TRASH_UNCONFIRMED: (
+        "Moved by macOS, but its place in the trash could not be confirmed."
+    ),
+    trash.WINDOWS_RECYCLE: (
+        "Moving files to the Recycle Bin isn't available on Windows: files larger "
+        "than the bin's capacity would be deleted permanently."
+    ),
     "not-attempted": "Not attempted: the job stopped after an error.",
 }
 _PROBLEM_CODES = {
@@ -93,19 +114,12 @@ SOURCES_CHANGED = "The sources changed since the review; nothing was changed."
 RECORDS_FAILED = "The image records could not be removed; the job stopped."
 
 
-def _is_trash_folder(name: str) -> bool:
-    folded = name.casefold()
-    return folded in TRASH_FOLDERS or (
-        folded.startswith(".trash-") and folded[7:].isdigit()
-    )
-
-
 def lexical_problem(relative: str) -> str | None:
     """The refusal code a root-relative POSIX path earns before any file access."""
     parts = relative.split("/")
     if "thumbnails" in parts[:-1]:
         return "thumbnail"
-    if any(_is_trash_folder(part) for part in parts[:-1]):
+    if any(is_trash_folder(part) for part in parts[:-1]):
         return "trash-folder"
     if os.path.splitext(parts[-1])[1].lower() not in DELETABLE_MEDIA_EXTENSIONS:
         return "not-media"
@@ -316,18 +330,19 @@ class ImageDeleteMixin:
             return _fail_job(job, "sources-changed", SOURCES_CHANGED)
         ok, backup = await self._safety_backup_step(job, db, backup_on)
         if not ok:
-            return
+            return None
         try:
             await self._delete_images(job, db, targets, mode, backup)
         finally:
             if backup:
                 await self._prune_backup_pool(job, db)
+        return None
 
     async def _delete_images(self, job, db, targets, mode, backup):
         job.update(total=len(targets), status_line=TITLES[mode])
         roots = [str(root) for root in self._get_all_output_dirs()]
         done_group = "trashed" if mode == "trash" else "removed"
-        changed, failed = 0, False
+        changed, failed = 0, None
         for done, target in enumerate(targets, start=1):
             if failed:
                 self._record_delete(job, "not_changed", target, "not-attempted")
@@ -336,7 +351,8 @@ class ImageDeleteMixin:
                 self._delete_one_image, db, target, mode, roots
             )
             if group == "errors":
-                failed = True
+                failed = code or RECORDS_FAILED_CODE
+                text = REASONS[code] if code else RECORDS_FAILED
                 job.record(
                     "errors",
                     {
@@ -344,9 +360,9 @@ class ImageDeleteMixin:
                         "root": target["root_label"],
                         "root_index": target["root_index"],
                         "root_scope": target["root_scope"],
-                        "detail": RECORDS_FAILED,
+                        "detail": text,
                     },
-                    f"! {target['root_label']}/{target['rel']} · {RECORDS_FAILED}",
+                    f"! {target['root_label']}/{target['rel']} · {text}",
                 )
             elif group == done_group:
                 changed += 1
@@ -365,8 +381,10 @@ class ImageDeleteMixin:
         job.log_line(line)
         job.update(progress=100, status_line=f"{line} · {backup_note(backup)}")
         if failed:
-            job.scope["failure"] = "records-failed"
-            job.publish({"type": "error", "code": "records-failed", "message": line})
+            # "records-failed" names every stop at an errors row (the page's
+            # "Stopped after N of M" toast); the row says which error it was.
+            job.scope["failure"] = RECORDS_FAILED_CODE
+            job.publish({"type": "error", "code": failed, "message": line})
             job.finish("failed")
 
     @staticmethod
@@ -405,7 +423,19 @@ class ImageDeleteMixin:
                 label = trash.move_to_trash(target["abs"], roots)
             except trash.TrashRefused as refused:
                 code = refused.code if refused.code in REASONS else trash.FAILED
+                self.logger.info(
+                    "Not moved to the trash (%s): %s %s",
+                    code,
+                    target["rel"],
+                    refused.detail,
+                )
                 return "not_changed", code, None
+            except trash.TrashUnconfirmed as unconfirmed:
+                # The file left its folder: keep its records, stop the job.
+                self.logger.warning(
+                    "Trash move not confirmed: %s %s", target["rel"], unconfirmed.detail
+                )
+                return "errors", TRASH_UNCONFIRMED, None
             mark = None
         else:
             mark = (target["size"], target["mtime_ns"])
@@ -443,37 +473,49 @@ class ImageDeleteMixin:
         if not marks:
             return set()
         located = self._workspace_locators(listings)
-        hidden = set()
-        for locator, (size, mtime_ns) in marks.items():
-            entry = located.get(locator)
-            if entry is None:
-                continue
-            try:
-                info = os.lstat(entry.path)
-            except OSError:
-                continue
-            if (info.st_size, info.st_mtime_ns) == (size, mtime_ns):
-                hidden.add(entry.normalized)
-        return hidden
+        return {
+            entry.normalized
+            for locator, entry in ((locator, located.get(locator)) for locator in marks)
+            if entry is not None and mark_hides(marks, locator, entry.path)
+        }
 
     def _removed_paths(self, paths) -> set:
         """The absolute paths among *paths* a removal mark hides (blocking)."""
         marks = self._removal_marks()
         if not marks:
             return set()
-        hidden = set()
-        for path in paths:
-            mark = marks.get(os.path.normcase(os.path.realpath(str(path))))
-            if mark is None:
-                continue
-            try:
-                info = os.lstat(path)
-            except OSError:
-                continue
-            if (info.st_size, info.st_mtime_ns) == tuple(mark):
-                hidden.add(str(path))
+        return {
+            str(path)
+            for path in paths
+            if mark_hides(marks, os.path.normcase(os.path.realpath(str(path))), path)
+        }
+
+    def _removal_check_under(self, root):
+        """``hidden(path) -> bool`` for files a walk of *root* found (blocking).
+
+        A walk never follows a link below *root*, so a file's locator is
+        ``normcase(join(realpath(root), relative))``: no ``realpath`` per
+        file, and an ``lstat`` only for a marked one. Without marks it
+        answers False at once.
+        """
+        marks = self._removal_marks()
+        if not marks:
+            return lambda _path: False
+        root = os.fspath(root)
+        real = os.path.realpath(root)
+
+        def hidden(path):
+            relative = os.path.relpath(os.fspath(path), root)
+            locator = os.path.normcase(os.path.join(real, relative))
+            return mark_hides(marks, locator, path)
+
         return hidden
 
+    def _without_removed(self, root, paths) -> list:
+        """*paths* found by a walk of *root*, less the hidden ones (blocking)."""
+        removed = self._removal_check_under(root)
+        return [path for path in paths if not removed(path)]
+
 
 def parse_mode(value):
     """The apply's ``mode``, or None when it is not one of MODES."""

py/api/image_return.py

@@ -0,0 +1,148 @@
+"""Return removed images to the library (#498).
+
+``POST /prompt_manager/images/removed/return {items | filter}`` names files a
+removal mark hides (#480, "Remove from the library only"): ``items`` are 1 to
+500 listing identities, ``filter`` the removed view's listing query
+(``removed: "1"`` plus the view's other parameters, at most 500 images). Only
+this route takes ``removed``: every other bulk route refuses it, because an
+explicit write there would show a file again without re-reading it.
+
+The route answers 202 ``{job, returned, not_removed}`` after starting a
+``scan`` job titled "Return images to the library" (admission ``records``:
+409 ``busy`` beside a scan, restore or delete, nothing changed). The job
+first deletes the marks that still hide their file (same size and
+``mtime_ns``), in one transaction, then scans exactly those files with the
+scan's own pipeline: prompts, links, pairs and Needs review items come back
+from the files' embedded metadata. Image tags, auto-tag marks, the own
+category, collection memberships and the favorite were deleted by the remove
+and do not come back. No file is moved, written or deleted.
+"""
+
+import os
+from pathlib import Path
+
+from aiohttp import web
+
+from .image_categories import TOO_MANY, _coded
+from .images import (
+    ListingSourceChanged,
+    _favorite_relative_key,
+    _json_body_or_error,
+    parse_listing_filter,
+)
+from .image_tag_jobs import MAX_JOB_TARGETS
+from .jobs import JobBusy
+from .job_routes import busy_response
+
+try:
+    from ...database.image_removal import mark_hides
+except ImportError:
+    from database.image_removal import mark_hides
+
+NONE_REMOVED = "None of the selected images are removed from the library."
+SOURCE_CHANGED = "The image sources changed. Reload Images and try again."
+
+
+def _bad(message):
+    return web.json_response({"success": False, "error": message}, status=400)
+
+
+class ImageReturnMixin:
+    """Return to library (#498): clear the marks, then scan those files."""
+
+    def _register_image_return_routes(self, routes):
+        # A fixed path beside the dynamic /images/{image_id} routes.
+        @routes.post("/prompt_manager/images/removed/return")
+        async def return_removed_images_route(request):
+            return await self.return_removed_images(request)
+
+    async def return_removed_images(self, request):
+        data, error = await _json_body_or_error(request)
+        if error is not None:
+            return error
+        if set(data) == {"filter"}:
+            targets, roots, error = await self._removed_filter_targets(data["filter"])
+        else:
+            _data, targets, roots, error = await self._category_request(
+                request, frozenset(), data=data
+            )
+        if error is not None:
+            return error
+        files = {}
+        for index in dict.fromkeys(t[0] for t in targets):
+            files[index] = await self._get_gallery_files(roots[index])
+        returned, not_removed = await self._run_in_executor(
+            self._hidden_targets, roots, targets, files
+        )
+        if not returned:
+            return _coded(400, "no-images", NONE_REMOVED)
+        try:
+            job = self._start_scan_job(
+                files="all", output_dirs=list(roots), returned=returned
+            )
+        except JobBusy as busy:
+            return busy_response(busy)
+        return web.json_response(
+            {
+                "success": True,
+                "job": job.summary(),
+                "returned": len(returned),
+                "not_removed": not_removed,
+            },
+            status=202,
+        )
+
+    async def _removed_filter_targets(self, raw):
+        """``(targets, roots, None)`` for the removed view's filter, else an error."""
+        if not isinstance(raw, dict) or raw.get("removed") != "1":
+            return None, None, _bad('filter must hold removed: "1"')
+        rest = {key: value for key, value in raw.items() if key != "removed"}
+        try:
+            parse_listing_filter(rest)
+            selection = await self.resolve_listing_filter(
+                {**rest, "removed": "1"}, MAX_JOB_TARGETS
+            )
+        except ListingSourceChanged:
+            return None, None, _coded(409, "source-changed", SOURCE_CHANGED)
+        except ValueError as exc:
+            return None, None, _bad(str(exc))
+        if selection.truncated:
+            return None, None, _coded(400, "too-many-images", TOO_MANY)
+        if not selection.targets:
+            return None, None, _coded(400, "no-images", NONE_REMOVED)
+        return tuple(selection.targets), list(self._get_all_output_dirs()), None
+
+    def _hidden_targets(self, roots, targets, files):
+        """``([(path, locator)], not_removed)``: the targets a mark hides now.
+
+        Each target is matched to its listed spelling (the walk's). One
+        physical file named twice (nested roots) is returned once; a target
+        that is not listed or not hidden counts in ``not_removed``.
+        """
+        marks = self._removal_marks()
+        listings, seen, returned, not_removed = {}, set(), [], 0
+        for index, _scope, relative in targets:
+            root = roots[index]
+            if index not in listings:
+                listings[index] = self._favorite_listing(root, files[index])
+            spelled = listings[index].get(_favorite_relative_key(relative))
+            if spelled is None:
+                not_removed += 1
+                continue
+            path = os.path.join(str(root), *spelled.split("/"))
+            locator = os.path.normcase(os.path.realpath(path))
+            if locator in seen:
+                continue
+            if not mark_hides(marks, locator, path):
+                not_removed += 1
+                continue
+            seen.add(locator)
+            returned.append((Path(path), locator))
+        return returned, not_removed
+
+    def _clear_returned_marks(self, returned) -> list:
+        """The paths whose marks this deleted (blocking; one transaction)."""
+        cleared = self.db.clear_hiding_marks(
+            [(locator, str(path)) for path, locator in returned]
+        )
+        return [path for path, locator in returned if locator in cleared]

py/api/image_tag_jobs.py

@@ -39,6 +39,7 @@ from typing import NamedTuple
 from aiohttp import web
 
 from .. import model_gate
+from ..tagging_defaults import SkipMatcher
 from .admin import _plural, _time_left
 from .annotation_suggest import (
     OptionError,
@@ -170,6 +171,8 @@ class _Run:
     owner: object
     service: object = None
     roots: dict = field(default_factory=dict)
+    # The job's Skipped tags, compiled once at its start (#431).
+    skip: object = None
 
 
 @dataclass(frozen=True)
@@ -183,6 +186,9 @@ class _Spec:
     options: dict | None = None
     # A listing filter the start resolves into ``targets`` (#392).
     filter: dict | None = None
+    # The Skipped tags when the job was requested (#431); a later save does
+    # not change a running job.
+    skip: tuple = ()
 
     @property
     def title(self) -> str:
@@ -238,7 +244,11 @@ def _parse_auto(targets, data):
         )
     except OptionError as exc:
         return None, bad_request(str(exc))
-    return _Spec("auto", targets, model_type=model_type, options=options), None
+    skip = tuple(AutotagConfig.SKIPPED_TAGS)
+    return (
+        _Spec("auto", targets, model_type=model_type, options=options, skip=skip),
+        None,
+    )
 
 
 def parse_image_tag_request(data: dict):
@@ -329,6 +339,8 @@ class AutotagRun:
     keep_in_memory: bool
     backup_on: bool
     owner: object
+    # The Skipped tags when the job was started (#431).
+    skip: tuple = ()
 
 
 @dataclass(frozen=True)
@@ -400,6 +412,14 @@ def _load_model(service, model_type: str, use_gpu: bool) -> None:
         raise RuntimeError(f"{model_type} could not be loaded")
 
 
+def _skipped_tags_line(entries: tuple) -> str:
+    """The job log's one line about the Skipped tags it runs with (#431)."""
+    if not entries:
+        return "Skipped tags: none"
+    count = len(entries)
+    return f"Skipped tags: {count} {'entry' if count == 1 else 'entries'}"
+
+
 def _refusal(status: int, code: str, message: str) -> Refusal:
     response = web.json_response(
         {"success": False, "code": code, "error": message}, status=status
@@ -507,6 +527,7 @@ class ImageTagJobsMixin:
         released when the task ends, also before its runner ever starts.
         """
         from ..autotag import MODELS, get_autotag_service
+        from ..config import AutotagConfig
 
         backup_on = self._safety_backup_policy() == "on"
         status = await self._run_in_executor(get_autotag_service().get_models_status)
@@ -529,8 +550,15 @@ class ImageTagJobsMixin:
             response = model_busy_response()
             return None, Refusal(409, "model-busy", MODEL_BUSY_ERROR, response)
         run = AutotagRun(
-            model_type, existing, options, keep_in_memory, backup_on, owner
+            model_type,
+            existing,
+            options,
+            keep_in_memory,
+            backup_on,
+            owner,
+            skip=tuple(AutotagConfig.SKIPPED_TAGS),
         )
+        admitted = False
         try:
             job = self._jobs.start(
                 AUTOTAG_BATCH_JOB_KIND,
@@ -541,11 +569,12 @@ class ImageTagJobsMixin:
                 groups=AUTOTAG_GROUP_NAMES,
                 runner=functools.partial(self._run_autotag_batch, run),
             )
-        except BaseException as refused:
-            model_gate.release(owner)
-            if isinstance(refused, JobBusy):
-                return None, _busy_refusal(refused)
-            raise
+            admitted = True
+        except JobBusy as refused:
+            return None, _busy_refusal(refused)
+        finally:
+            if not admitted:
+                model_gate.release(owner)
         job.task.add_done_callback(lambda _task: model_gate.release(owner))
         return job, None
 
@@ -596,7 +625,9 @@ class ImageTagJobsMixin:
             return _batch_cancelled(job)
         job.publish({"type": "phase", "phase": "model"})
         job.update(status_line="Loading the tagging model")
-        spec = _Spec("auto", (), model_type=run.model_type, options=run.options)
+        spec = _Spec(
+            "auto", (), model_type=run.model_type, options=run.options, skip=run.skip
+        )
         service = await self._prepare_tagging_model(job, spec, run.owner)
         if service is None:
             return "failed"
@@ -622,6 +653,7 @@ class ImageTagJobsMixin:
             f"Tagging started · {job.scope.get('label') or run.model_type} · "
             f"{_plural(total, 'image')} · {policy}"
         )
+        job.log_line(_skipped_tags_line(spec.skip))
         if job.cancelled:
             return _batch_cancelled(job)
         remaining = await self._autotag_prepass(job, targets, run.existing == "skip")
@@ -631,7 +663,9 @@ class ImageTagJobsMixin:
         spec = dataclasses.replace(
             spec, targets=tuple(item.target for item in remaining)
         )
-        loop_run = _Run(job, spec, epoch, run.owner, service)
+        loop_run = _Run(
+            job, spec, epoch, run.owner, service, skip=SkipMatcher(spec.skip)
+        )
         if not await self._tag_each_file(loop_run, done_before=done):
             job.update(status_line=STORAGE_RESTORED)
             job.log_line(f"Tagging stopped · {STORAGE_RESTORED}")
@@ -737,6 +771,9 @@ class ImageTagJobsMixin:
             started += f" · {spec.model_type}"
         job.log_line(started)
         run = _Run(job, spec, epoch, owner)
+        if spec.mode == "auto":
+            job.log_line(_skipped_tags_line(spec.skip))
+            run.skip = SkipMatcher(spec.skip)
         try:
             if spec.mode == "auto":
                 run.service = await self._prepare_tagging_model(job, spec, owner)
@@ -892,6 +929,7 @@ class ImageTagJobsMixin:
                 capture,
                 spec.model_type,
                 spec.options,
+                run.skip,
                 on_skipped=functools.partial(store.release_use, capture.token),
             )
         except Exception:

py/api/images.py

@@ -39,10 +39,8 @@ from .thumbnail_encode import (
     PREVIEW_HELD,
     EncoderClosed,
     PreviewWriteHeld,
-    replace_unless_held,
     run_encoder,
     sweep_stray_temps,
-    temp_sibling,
     thumbnail_frame,
     writer_settings,
     write_thumbnail,
@@ -66,6 +64,12 @@ try:
     from ...database.models import normalize_image_path
     from ...database.run_attribution import recorded_output_lookup
     from ...utils.card_seed import card_seed, header_params, parsed_positive
+    from ...utils.media_folders import (
+        in_skipped_folder,
+        in_trash_below_roots,
+        in_trash_folder,
+        is_skipped_folder,
+    )
     from ...utils.parallel import map_parallel
     from ...utils.json_strict import dumps_strict
     from ...utils.inspection_response import (
@@ -79,7 +83,7 @@ try:
         OpenerUnavailable,
         RevealGate,
         RevealTargetChanged,
-        is_local_request,
+        is_same_origin_request,
         reveal_executor,
         reveal_in_file_manager,
     )
@@ -91,6 +95,12 @@ except ImportError:
     from database.models import normalize_image_path
     from database.run_attribution import recorded_output_lookup
     from utils.card_seed import card_seed, header_params, parsed_positive
+    from utils.media_folders import (
+        in_skipped_folder,
+        in_trash_below_roots,
+        in_trash_folder,
+        is_skipped_folder,
+    )
     from utils.parallel import map_parallel
     from utils.json_strict import dumps_strict
     from utils.inspection_response import (
@@ -104,7 +114,7 @@ except ImportError:
         OpenerUnavailable,
         RevealGate,
         RevealTargetChanged,
-        is_local_request,
+        is_same_origin_request,
         reveal_executor,
         reveal_in_file_manager,
     )
@@ -275,10 +285,20 @@ def _forbidden(message="Access denied"):
     return web.json_response({"success": False, "error": message}, status=403)
 
 
-def _media_access_error(path, allowed_dirs):
-    """Return a 403 response when *path* must not be served, else None.
+def _file_not_found():
+    return web.json_response(
+        {"success": False, "error": "Image file not found"},
+        status=404,
+        headers={"Cache-Control": "no-store"},
+    )
 
-    Fails closed: an empty *allowed_dirs* denies everything.
+
+def _media_access_error(path, allowed_dirs):
+    """The response refusing *path*, or None when it may be read.
+
+    Fails closed: an empty *allowed_dirs* denies everything (403). A file in a
+    trash folder below its root is thrown away, not an output: a 404, as if it
+    were not there (#543), also through a link or another spelling.
     """
     if Path(path).suffix.lower() not in SERVABLE_EXTENSIONS:
         return _forbidden("Only media files can be served")
@@ -286,6 +306,8 @@ def _media_access_error(path, allowed_dirs):
         return _forbidden("No allowed image directories configured")
     if not any(_is_within(path, root) for root in allowed_dirs):
         return _forbidden()
+    if in_trash_below_roots(path, allowed_dirs):
+        return _file_not_found()
     return None
 
 
@@ -315,11 +337,6 @@ def _thumbnail_batch_size(workers):
     return max(4, workers * 2)
 
 
-# Preview files are written by thumbnail_encode in the configured format
-# (thumbnail_settings, #317); these names stay importable from here.
-_temp_sibling = temp_sibling
-_replace_unless_held = replace_unless_held
-
 # A preview that could not be replaced (#249 S5 M6). Windows refuses to rename
 # over a file another process has open (WinError 5 or 32), such as a preview
 # the gallery is serving; the old preview stays and the file counts as an error.
@@ -372,6 +389,7 @@ def _video_placeholder_image(thumbnail_size):
             font=font,
         )
     except (OSError, AttributeError):
+        # a placeholder without its label still works
         pass
     return img
 
@@ -475,6 +493,7 @@ def _preview_names(folders, folder):
                     except OSError:
                         continue
         except OSError:
+            # an unreadable folder lists as empty; nothing there to cache
             pass
         folders[folder] = names
     return names
@@ -793,12 +812,12 @@ def _reveal_not_found():
     )
 
 
-def _not_local():
+def _not_same_origin():
     return web.json_response(
         {
             "success": False,
-            "code": "not-local",
-            "error": "Show in folder works only on the computer running ComfyUI.",
+            "code": "not-same-origin",
+            "error": "This page's address does not match the server's.",
         },
         status=403,
     )
@@ -915,7 +934,8 @@ def _name_suffix(name):
 def _walk_media(output_path, max_files):
     """Yield the ``os.DirEntry`` of each media file under *output_path*.
 
-    Symlinks are never followed, ``thumbnails/`` is skipped, descent stops
+    Symlinks are never followed, ``thumbnails/`` and trash folders are
+    skipped (``utils.media_folders``, #501), descent stops
     at MAX_SCAN_DEPTH and at most *max_files* entries are produced. The order
     is ``os.walk``'s (top-down, subfolders in ``os.scandir`` order, files by
     name), which the gallery's ties in mtime keep. Each entry's type comes
@@ -927,8 +947,6 @@ def _walk_media(output_path, max_files):
     produced = 0
     while stack:
         top, rel_parts = stack.pop()
-        if "thumbnails" in rel_parts:
-            continue
         dirs, files = [], []
         try:
             with os.scandir(top) as entries:
@@ -952,7 +970,7 @@ def _walk_media(output_path, max_files):
         if len(rel_parts) >= MAX_SCAN_DEPTH:
             continue
         for entry in reversed(dirs):
-            if not entry.is_symlink():
+            if not entry.is_symlink() and not is_skipped_folder(entry.name):
                 stack.append((entry.path, rel_parts + (entry.name,)))
 
 
@@ -976,8 +994,10 @@ def _settled_entry(root, relative):
     *folders, name = relative.split(os.sep)
     if _name_suffix(name).lower() not in MEDIA_EXTENSIONS:
         return None
-    if "thumbnails" in folders or len(folders) > MAX_SCAN_DEPTH:
+    if in_skipped_folder(folders) or len(folders) > MAX_SCAN_DEPTH:
         return None
+    if in_trash_below_roots(path, [root]):
+        return None  # a linked folder on the way leads into a trash folder
     folder = os.fspath(root)
     for part in ("", *folders):
         folder = os.path.join(folder, part) if part else folder
@@ -1159,7 +1179,10 @@ def _workspace_filters(query: Mapping) -> dict:
     the linked positive, and ``favorites=1`` keeps only favorites (the
     Favorites page, #249 D51). ``collection=<id>`` (1 to 9 digits) keeps only
     that collection's members (#331); it wins over ``favorites`` (a page never
-    sets both). Raises ValueError for any unknown or oversized value.
+    sets both). ``removed=1`` lists only the files a removal mark hides (#498,
+    Show removed images); such a file has no records, so it never combines
+    with ``favorites``, ``collection``, ``category``, ``tags`` or the rating
+    sort. Raises ValueError for any unknown or oversized value.
     """
     text = query.get("q", "").strip()
     folded = text.casefold()
@@ -1174,6 +1197,12 @@ def _workspace_filters(query: Mapping) -> dict:
     )
     favorites = query.get("favorites", "0")
     collection = query.get("collection", "")
+    removed = query.get("removed", "0")
+    if removed not in ("0", "1") or (
+        removed == "1"
+        and (favorites == "1" or collection or category or tags or order == "rating")
+    ):
+        raise ValueError("Invalid gallery filters")
     if (
         favorites not in ("0", "1")
         or (collection != "" and _COLLECTION_ID.fullmatch(collection) is None)
@@ -1201,6 +1230,7 @@ def _workspace_filters(query: Mapping) -> dict:
         "tags": tags,
         "favorites": favorites == "1" and not collection,
         "collection": int(collection) if collection else None,
+        "removed": removed == "1",
     }
 
 
@@ -1503,7 +1533,7 @@ class ImageRoutesMixin:
         async def clear_image_favorite_route(request):
             return await self.update_image_favorite(request, favorite=False)
 
-        # POST, local-only: opens this computer's file manager (D38).
+        # POST, same-origin: opens the server computer's file manager (D38).
         @routes.post("/prompt_manager/images/reveal")
         async def reveal_image_route(request):
             return await self.reveal_image(request)
@@ -2072,6 +2102,7 @@ class ImageRoutesMixin:
                         filters["tags"],
                         filters["favorites"],
                         filters["collection"],
+                        filters["removed"],
                     ]
                 ).encode()
             ).hexdigest()
@@ -2101,6 +2132,11 @@ class ImageRoutesMixin:
                     self._select_workspace_page, listings, filters, after, limit, scope
                 )
             )
+            # #498: Show removed images (N), every removed file in the selected
+            # source(s) whatever the filters (the hidden set is cached work).
+            removed_total = len(
+                await self._run_in_executor(self._hidden_entries, listings)
+            )
         except CollectionMissing:
             return collection_not_found()
         except Exception:
@@ -2127,6 +2163,7 @@ class ImageRoutesMixin:
             "limit": limit,
             "has_more": has_more,
             "next_cursor": next_cursor,
+            "removed_total": removed_total,
         }
         if positive_count is _COUNT_OMITTED:
             del result["positive_count"]
@@ -2450,9 +2487,12 @@ class ImageRoutesMixin:
             def key(entry):
                 return _workspace_key(entry, order)
 
-        # #480: files removed from the library stay on disk, hidden here.
+        # #480: files removed from the library stay on disk, hidden here; the
+        # removed view (#498) lists only them.
         hidden = self._hidden_entries(listings)
-        if hidden:
+        if filters.get("removed"):
+            entries = [entry for entry in entries if entry.normalized in hidden]
+        elif hidden:
             entries = [entry for entry in entries if entry.normalized not in hidden]
         return (
             entries,
@@ -2740,6 +2780,9 @@ class ImageRoutesMixin:
             if _has_traversal(filepath):
                 return _forbidden()
             suffix = Path(filepath).suffix.lower()
+            if in_trash_folder(filepath):
+                # Before any file access, so no preview is made for it (#543).
+                return _file_not_found()
             if suffix not in SERVABLE_EXTENSIONS and not (
                 # A JPEG XL preview is served from thumbnails/ only (#317).
                 suffix in thumbnail_settings.THUMBNAIL_EXTENSIONS
@@ -2763,6 +2806,8 @@ class ImageRoutesMixin:
                 # a fall-through to the next root.
                 if not _is_within(image_path, output_path):
                     return _forbidden()
+                if in_trash_below_roots(image_path, output_dirs):
+                    return _file_not_found()
                 if image_path.is_file():
                     return self._file_response(image_path)
 
@@ -2814,8 +2859,9 @@ class ImageRoutesMixin:
         if _is_thumbnail_cache_path(filepath):
             return _forbidden()
         image_path = root / filepath
-        if not _is_within(image_path, root):
-            return _forbidden()
+        denied = _media_access_error(image_path, [root])
+        if denied is not None:
+            return denied
         if not image_path.is_file():
             return web.json_response(
                 {"success": False, "error": "Image file not found"}, status=404
@@ -2929,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:
@@ -3075,7 +3122,9 @@ class ImageRoutesMixin:
         no symlinks), each preview folder is listed once, and at most
         MAX_THUMBNAIL_FILES targets are queued per batch (a job plans again
         after each batch); ``remaining`` counts the targets beyond that cap.
-        Sources in *exclude* (worked on by an earlier batch) are left out.
+        Sources in *exclude* (worked on by an earlier batch) are left out, and
+        so are files removed from the library (#501: a removal mark hides
+        them, and their previews went with their records).
         ``stop()`` true ends the walk before the next file.
         """
         settings = writer_settings(settings)
@@ -3084,11 +3133,14 @@ class ImageRoutesMixin:
         plan = {key: 0 for key in _PLAN_COUNTS}
         plan.update(targets=[], fresh_files=[])
         folders = {}
+        removed = self._removal_check_under(output_path)
         for src in _iter_media_files(output_path, MAX_GALLERY_FILES):
             if stop is not None and stop():
                 break
             if exclude and src in exclude:
                 continue
+            if removed(src):
+                continue
             rel_no_ext = src.relative_to(output_path).with_suffix("")
             stem = f"{rel_no_ext.name}{thumbnail_settings.THUMB_SUFFIX}"
             dst = thumbnails_dir / rel_no_ext.parent / (stem + settings.extension)
@@ -3656,12 +3708,12 @@ class ImageRoutesMixin:
     async def reveal_image(self, request):
         """POST ``{root_index, root_scope, path}``: show the file in its folder.
 
-        Local-only (a same-origin browser request from a loopback client, no
-        forwarding header) and JSON-only: the browser must run on the
-        computer whose file manager opens. One reveal at a time (429 busy).
+        Same-origin and JSON-only, directly or through a proxy: the folder
+        opens on the computer running ComfyUI, wherever the browser is
+        (maintainer decision, #533). One reveal at a time (429 busy).
         """
-        if not is_local_request(request.remote, request.headers, request.scheme):
-            return _not_local()
+        if not is_same_origin_request(request.headers, request.scheme):
+            return _not_same_origin()
         if request.content_type != "application/json":
             # A non-JSON POST is a CORS "simple" request with no preflight.
             return web.json_response(
@@ -3739,7 +3791,9 @@ class ImageRoutesMixin:
                 {
                     "success": False,
                     "code": "unavailable",
-                    "error": "No file manager is available on this computer.",
+                    "error": (
+                        "No file manager is available on the computer running ComfyUI."
+                    ),
                 },
                 status=501,
             )
@@ -3944,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(
                     {
@@ -4157,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)``.
 
@@ -4165,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.
@@ -4195,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

py/api/job_routes.py

@@ -250,7 +250,7 @@ class JobRoutesMixin:
         try:
             await response.write(_summary_event(job))
             async with contextlib.aclosing(job.events()) as events:
-                async for _payload in events:
+                async for _ in events:
                     await response.write(_summary_event(job))
             # The last event is published before the job takes its final state.
             await response.write(_summary_event(job))

py/api/jobs.py

@@ -123,6 +123,9 @@ MODEL_DOWNLOAD_JOB_KIND = "model_download"
 DUPLICATES_JOB_KIND = "duplicates"
 DUPLICATE_GROUP_NAMES = ("groups_found",)
 DUPLICATE_CLEANUP_GROUP_NAMES = ("removed", "kept", "errors")
+# The workspace Library's Delete… (#443, py/api/library_delete.py): a
+# ``records`` hold around its safety backup and write.
+PROMPT_DELETE_JOB_KIND = "prompt_delete"
 IMAGE_DELETE_JOB_KIND = "image_delete"  # #480
 IMAGE_DELETE_GROUP_NAMES = ("trashed", "removed", "not_changed", "errors")
 # The System health check (py/api/system.py): a read-only ``read`` job with
@@ -201,6 +204,8 @@ class Job:
         self.total = 0
         self.counters = dict.fromkeys(groups, 0)
         self.outcomes = {group: [] for group in groups}
+        # Groups whose report lists every row, beyond MAX_OUTCOME_ROWS (#550).
+        self.complete = {}
         self.log = []
         self._cancel_requested = False
         self.last_event = None
@@ -294,31 +299,36 @@ class Job:
             )
         self.log_line(log_line)
 
-    def record_counted(self, group, count, rows=()):
+    def record_counted(self, group, count, rows=(), *, complete=False):
         """Count ``count`` outcomes in ``group`` at once, listing ``rows``.
 
         For outcomes that need no log line each (a thumbnail job's already
         fresh previews): the counter is exact, ``rows`` (``record``'s row
         shape) are listed up to MAX_OUTCOME_ROWS, and nothing is logged.
+        ``complete`` also keeps every row for the report (#550: a record of
+        each change that must not be cut short; the caller bounds it).
         """
         if group not in self.counters:
             raise ValueError(f"Unknown outcome group {group!r} for {self.kind}")
         self.counters[group] += max(0, int(count))
         listed = self.outcomes[group]
+        kept = self.complete.setdefault(group, []) if complete else None
         for row in rows:
-            if len(listed) >= MAX_OUTCOME_ROWS:
+            if kept is None and len(listed) >= MAX_OUTCOME_ROWS:
                 break
             index = row.get("root_index")
             if isinstance(index, bool) or not isinstance(index, int) or index < 0:
                 index = None
-            listed.append(
-                {
-                    "path": _relative_path(row.get("path")),
-                    "root": str(row.get("root") or ""),
-                    "root_index": index,
-                    "detail": str(row.get("detail") or ""),
-                }
-            )
+            shaped = {
+                "path": _relative_path(row.get("path")),
+                "root": str(row.get("root") or ""),
+                "root_index": index,
+                "detail": str(row.get("detail") or ""),
+            }
+            if kept is not None:
+                kept.append(shaped)
+            if len(listed) < MAX_OUTCOME_ROWS:
+                listed.append(dict(shaped))
 
     def finish(self, state="finished"):
         """End the job in ``state`` and release every subscriber.
@@ -403,8 +413,9 @@ class Job:
         """The downloadable report: the summary, every outcome group and the log.
 
         Each group carries its exact count, its listed rows (at most
-        MAX_OUTCOME_ROWS) and whether the list is cut short. Rows and log lines
-        already hold root-relative paths only.
+        MAX_OUTCOME_ROWS, or every row of a ``complete`` group) and whether
+        the list is cut short. Rows and log lines already hold root-relative
+        paths only.
         """
         summary = self.summary()
         summary.pop("listed", None)
@@ -414,16 +425,21 @@ class Job:
             "generated_at": self._clock(),
             "job": summary,
             "outcomes": {
-                group: {
-                    "count": self.counters.get(group, len(rows)),
-                    "rows": [dict(row) for row in rows],
-                    "truncated": self.counters.get(group, len(rows)) > len(rows),
-                }
+                group: self._report_group(group, rows)
                 for group, rows in self.outcomes.items()
             },
             "log": list(self.log),
         }
 
+    def _report_group(self, group, listed):
+        rows = self.complete.get(group, listed)
+        count = self.counters.get(group, len(rows))
+        return {
+            "count": count,
+            "rows": [dict(row) for row in rows],
+            "truncated": count > len(rows),
+        }
+
     def report_filename(self):
         """The download name: prompt-manager-KIND-YYYYMMDD-HHMM.json (local time)."""
         stamp = time.strftime("%Y%m%d-%H%M", time.localtime(self._clock()))

py/api/library.py

@@ -157,6 +157,13 @@ def _archive_scope(request: web.Request) -> str:
     return value
 
 
+def _shelf(request: web.Request) -> str:
+    value = request.query.get("shelf", "active")
+    if value not in ("active", "archived"):
+        raise ValueError("shelf must be active or archived")
+    return value
+
+
 def _missing() -> web.Response:
     return web.json_response(
         {"success": False, "error": "Positive prompt not found"}, status=404
@@ -332,6 +339,9 @@ class LibraryRoutesMixin:
             "tags": sorted(set(request.query.getall("tag", []))),
             "tag_mode": request.query.get("tag_mode", "all"),
             "sort": request.query.get("sort", "newest"),
+            # #443: which shelf; part of the cursor scope, so a cursor never
+            # crosses shelves. Not the variant ``archive`` scope of /{id}.
+            "shelf": _shelf(request),
         }
         limit = _page_limit(request)
         cursor = request.query.get("cursor")

py/api/library_delete.py

@@ -0,0 +1,321 @@
+"""Delete or archive selected Library prompts, and restore archived ones (#443).
+
+The workspace Prompt library's Delete… (D152). The legacy ``POST
+/prompt_manager/bulk/delete`` and ``DELETE /prompt_manager/delete/{id}`` are
+separate and unchanged.
+
+- ``POST /prompt_manager/library/delete/preview {prompt_ids, expected_epoch}``
+  classifies the selection (reads only, under a lease): the ids to ``delete``
+  (in no saved pair), to ``archive`` (paired), the ``protected`` ones (tagged
+  ``__protected__``, kept), counts of ``already_archived`` and ``not_found``,
+  the ``image_links`` a delete removes, the safety ``backup`` policy and,
+  when a backup would be taken, the ``disk`` it needs.
+- ``POST /prompt_manager/library/delete {archive_ids, delete_ids,
+  expected_epoch, backup}`` executes that review under a ``records`` hold
+  (refused 409 ``busy`` beside a scan, restore, maintenance, move or
+  Optimize). When a delete remains and a backup was promised (the reviewed
+  ``backup`` was "on" or the policy is on now) it checks the disk (507
+  ``insufficient-space``) and takes a verified safety backup (500
+  ``backup-failed``); then one transaction deletes the reviewed deletes whose
+  fingerprint is unchanged since the backup and archives the paired ones.
+- ``POST /prompt_manager/library/restore {prompt_ids, expected_epoch}``
+  removes the archive mark (under a lease).
+
+A changed storage epoch answers 409 ``storage-changed``; nothing is written.
+"""
+
+import sqlite3
+
+from aiohttp import web
+
+try:
+    from ...database.library_commands import LibraryCommandError
+    from ...database.safety_backup import SafetyBackupFailed
+except ImportError:
+    from database.library_commands import LibraryCommandError
+    from database.safety_backup import SafetyBackupFailed
+
+from . import maintenance as _maintenance
+from .admin import _prune_safety_backups
+from .job_routes import busy_response
+from .jobs import PROMPT_DELETE_JOB_KIND, JobBusy
+from .prompts import parse_bulk_prompt_ids
+
+PREVIEW_URL = "/prompt_manager/library/delete/preview"
+DELETE_URL = "/prompt_manager/library/delete"
+RESTORE_URL = "/prompt_manager/library/restore"
+DELETE_TITLE = "Delete prompts (Library)"
+_SELECTION_KEYS = frozenset({"prompt_ids", "expected_epoch"})
+_DELETE_KEYS = frozenset({"archive_ids", "delete_ids", "expected_epoch", "backup"})
+# #482: Delete permanently, with its digests; both or neither (a #443 body).
+_PAIRED_KEYS = frozenset({"delete_paired_ids", "paired_digests"})
+# #492: the shelf the Delete… runs on, optional on its own (default "active").
+_SHELF_KEYS = frozenset({"shelf"})
+STORAGE_CHANGED = "Prompt storage was restored; refresh the library"
+
+
+def _failure(status, code, error, **extra):
+    return web.json_response(
+        {"success": False, "code": code, "error": error, **extra}, status=status
+    )
+
+
+def _invalid(message):
+    return _failure(400, "invalid-request", message)
+
+
+def _storage_changed(exc):
+    return _failure(409, exc.code, STORAGE_CHANGED, library_epoch=exc.epoch)
+
+
+async def _body(request, keys, *groups):
+    """The JSON object with exactly ``keys``, plus all or none of each group."""
+    try:
+        data = await request.json()
+    except ValueError:
+        raise ValueError("Expected a JSON object") from None
+    if not isinstance(data, dict):
+        raise ValueError("Expected a JSON object")
+    extra = set(data) - keys
+    partial = False
+    for group in groups:
+        if extra & group:
+            partial = partial or not group <= extra
+            extra -= group
+    if not keys <= set(data) or extra or partial:
+        raise ValueError("Request must contain exactly " + ", ".join(sorted(keys)))
+    return data
+
+
+def _shelf(data):
+    """#492: the shelf a Delete… was reviewed on."""
+    shelf = data.get("shelf", "active")
+    if shelf not in ("active", "archived"):
+        raise ValueError("shelf must be active or archived")
+    return shelf
+
+
+def _paired_delete(data, archive_ids):
+    """#482: the reviewed archive ids to delete with their pairs -> digest."""
+    if "delete_paired_ids" not in data:
+        return {}
+    ids = _reviewed_ids(data["delete_paired_ids"], "delete_paired_ids")
+    digests = data["paired_digests"]
+    if not isinstance(digests, dict):
+        raise ValueError("paired_digests must map prompt ids to digests")
+    if not set(ids) <= set(archive_ids):
+        raise ValueError("delete_paired_ids must be reviewed archive ids")
+    chosen = {}
+    for pid in ids:
+        digest = digests.get(str(pid))
+        if not isinstance(digest, str) or len(digest) != 64:
+            raise ValueError("Every delete_paired_ids entry needs its review digest")
+        chosen[pid] = digest
+    return chosen
+
+
+def _reviewed_ids(value, name):
+    """A reviewed id list: possibly empty, else as ``parse_bulk_prompt_ids``."""
+    if value == []:
+        return []
+    try:
+        return parse_bulk_prompt_ids(value)
+    except ValueError as exc:
+        raise ValueError(str(exc).replace("prompt_ids", name)) from None
+
+
+class LibraryDeleteRoutesMixin:
+    """Mixin providing the workspace Library's delete, archive and restore."""
+
+    def _register_library_delete_routes(self, routes):
+        routes.post(PREVIEW_URL)(self.preview_library_delete)
+        routes.post(DELETE_URL)(self.delete_library_prompts)
+        routes.post(RESTORE_URL)(self.restore_library_prompts)
+
+    async def preview_library_delete(self, request):
+        try:
+            data = await _body(request, _SELECTION_KEYS, _SHELF_KEYS)
+            ids = parse_bulk_prompt_ids(data["prompt_ids"])
+            shelf = _shelf(data)
+        except ValueError as exc:
+            return _invalid(str(exc))
+        db = self.db
+        policy = self._safety_backup_policy()
+        try:
+            async with self._jobs.lease("Library delete review"):
+                review = await self._run_in_executor(
+                    db.classify_library_delete,
+                    ids,
+                    expected_epoch=data["expected_epoch"],
+                    shelf=shelf,
+                )
+                disk = None
+                if (review["delete"] or review["deletable"]) and policy == "on":
+                    public, short, _size = await self._run_in_executor(
+                        _maintenance.disk_plan,
+                        db.model.db_path,
+                        backup=True,
+                        vacuum=False,
+                    )
+                    disk = {
+                        **public,
+                        "message": (
+                            None
+                            if short is None
+                            else _maintenance.short_space_message(short)
+                        ),
+                    }
+        except JobBusy as busy:
+            return busy_response(busy)
+        except LibraryCommandError as exc:
+            return _storage_changed(exc)
+        except ValueError as exc:
+            return _invalid(str(exc))
+        except (OSError, sqlite3.Error):
+            self.logger.exception("Library delete review failed")
+            return _failure(500, "review-failed", "The delete could not be reviewed")
+        return web.json_response(
+            {
+                "success": True,
+                "library_epoch": review["library_epoch"],
+                "delete": review["delete"],
+                "archive": review["archive"],
+                "protected": review["protected"],
+                "already_archived": len(review["already_archived"]),
+                "not_archived": len(review["not_archived"]),
+                "not_found": len(review["not_found"]),
+                "image_links": review["image_links"],
+                # #482: the archive ids Delete permanently may delete with pairs.
+                "deletable": review["deletable"],
+                "review_digests": {
+                    str(pid): digest for pid, digest in review["review_digests"].items()
+                },
+                "paired_counts": review["paired_counts"],
+                "paired_image_links": review["paired_image_links"],
+                "backup": policy,
+                "disk": disk,
+            }
+        )
+
+    async def delete_library_prompts(self, request):
+        try:
+            data = await _body(request, _DELETE_KEYS, _PAIRED_KEYS, _SHELF_KEYS)
+            shelf = _shelf(data)
+            archive_ids = _reviewed_ids(data["archive_ids"], "archive_ids")
+            delete_ids = _reviewed_ids(data["delete_ids"], "delete_ids")
+            if not archive_ids and not delete_ids:
+                raise ValueError("Nothing to delete or archive")
+            if len(archive_ids) + len(delete_ids) > 1000:
+                raise ValueError("At most 1000 prompts can be changed at once")
+            if set(archive_ids) & set(delete_ids):
+                raise ValueError("A prompt cannot be both archived and deleted")
+            if data["backup"] not in ("on", "off"):
+                raise ValueError("backup must be on or off")
+            paired = _paired_delete(data, archive_ids)
+        except ValueError as exc:
+            return _invalid(str(exc))
+        busy = self._jobs.blocker(PROMPT_DELETE_JOB_KIND, admission="records")
+        if busy is not None:
+            return busy_response(JobBusy(busy))
+        try:
+            async with self._jobs.hold(
+                PROMPT_DELETE_JOB_KIND, DELETE_TITLE, admission="records"
+            ) as job:
+                response = await self._held_library_delete(
+                    self.db, data, archive_ids, delete_ids, paired, shelf
+                )
+                if response.status >= 400:
+                    job.finish("failed")
+                return response
+        except JobBusy as busy:
+            return busy_response(busy)
+
+    async def _held_library_delete(
+        self, db, data, archive_ids, delete_ids, paired, shelf="active"
+    ):
+        epoch = data["expected_epoch"]
+        promised = data["backup"] == "on" or self._safety_backup_policy() == "on"
+        backup = None
+        try:
+            candidates = None
+            if delete_ids or paired:
+                review = await self._run_in_executor(
+                    db.classify_library_delete,
+                    [*delete_ids, *paired],
+                    expected_epoch=epoch,
+                    # #492: the same shelf, or archived ids would skip the backup.
+                    shelf=shelf,
+                )
+                candidates = review["fingerprints"]
+                # #482: a permanent delete of a prompt in a saved pair is backed up too.
+                if (candidates or review["deletable"]) and promised:
+                    _disk, short, _size = await self._run_in_executor(
+                        _maintenance.disk_plan,
+                        db.model.db_path,
+                        backup=True,
+                        vacuum=False,
+                    )
+                    if short is not None:
+                        return _failure(
+                            507,
+                            "insufficient-space",
+                            _maintenance.short_space_message(short),
+                        )
+                    backup = await self._run_in_executor(
+                        _maintenance.take_safety_backup, db.model
+                    )
+            result = await self._run_in_executor(
+                db.archive_or_delete_library_prompts,
+                archive_ids,
+                delete_ids,
+                expected_epoch=epoch,
+                candidates=candidates if candidates is not None else {},
+                delete_paired=paired,
+                shelf=shelf,
+            )
+        except LibraryCommandError as exc:
+            return _storage_changed(exc)
+        except SafetyBackupFailed as error:
+            self.logger.error("Safety backup before a Library delete failed")
+            return _failure(
+                500, "backup-failed", _maintenance.backup_failed_message(error)
+            )
+        except ValueError as exc:
+            return _invalid(str(exc))
+        except (OSError, sqlite3.Error):
+            self.logger.exception("Library delete failed")
+            return _failure(500, "delete-failed", "The prompts could not be changed")
+        finally:
+            if backup is not None:
+                await self._prune_after_delete(db)
+        return web.json_response({"success": True, **result, "backup": backup})
+
+    async def _prune_after_delete(self, db):
+        try:
+            await self._run_in_executor(_prune_safety_backups, db.model.db_path)
+        except Exception:  # pruning old copies never fails the delete
+            self.logger.exception("Could not prune old safety backups")
+
+    async def restore_library_prompts(self, request):
+        try:
+            data = await _body(request, _SELECTION_KEYS)
+            ids = parse_bulk_prompt_ids(data["prompt_ids"])
+        except ValueError as exc:
+            return _invalid(str(exc))
+        try:
+            async with self._jobs.lease("Library restore"):
+                result = await self._run_in_executor(
+                    self.db.restore_library_prompts,
+                    ids,
+                    expected_epoch=data["expected_epoch"],
+                )
+        except JobBusy as busy:
+            return busy_response(busy)
+        except LibraryCommandError as exc:
+            return _storage_changed(exc)
+        except ValueError as exc:
+            return _invalid(str(exc))
+        except (OSError, sqlite3.Error):
+            self.logger.exception("Library restore failed")
+            return _failure(500, "restore-failed", "The prompts could not be restored")
+        return web.json_response({"success": True, **result})

py/api/library_location.py

@@ -207,6 +207,7 @@ class LibraryLocationRoutesMixin:
         try:
             size = os.path.getsize(location.db_path)
         except OSError:
+            # no library file yet: report size 0
             pass
         return {
             "path": self._library_public_path(location.db_path),
@@ -373,6 +374,7 @@ class LibraryLocationRoutesMixin:
         line += ". Logs move at the next restart."
         job.log_line(line)
         job.update(progress=100, status_line=line)
+        return None
 
     @staticmethod
     def _fail_library_move(job, code, message):

py/api/log_cleanup.py

@@ -205,3 +205,4 @@ class LogCleanupMixin:
             line += f", {failed} could not be deleted"
         job.log_line(line)
         job.update(processed=len(files), progress=100, status_line=line)
+        return None

py/api/lora_import_routes.py

@@ -436,7 +436,7 @@ class LoraImportRoutesMixin:
             job.log_line(f"{CANCELLED} {CACHE_NOTE}")
             job.update(status_line=CANCELLED)
             job.finish("cancelled")
-            return
+            return None
         except Exception:
             self.logger.exception("LoRA import staging failed")
             return _fail_job(job, "import-failed", f"{FAILED} {CACHE_NOTE}")
@@ -455,10 +455,10 @@ class LoraImportRoutesMixin:
             job.log_line(f"{CANCELLED} {CACHE_NOTE}")
             job.update(status_line=CANCELLED)
             job.finish("cancelled")
-            return
+            return None
         ok, _backup = await self._safety_backup_step(job, db, backup_on)
         if not ok:
-            return
+            return None
         # I1: the files are listed and fingerprinted again right before the
         # replace, so a file changed, removed or added while the downloads ran
         # (LoraManager refreshing its CivitAI info) aborts with nothing deleted.
@@ -466,6 +466,7 @@ class LoraImportRoutesMixin:
             return _fail_job(job, "plan-changed", f"{SOURCE_CHANGED} {CACHE_NOTE}")
         job.update(status_line="Replacing the LoRA collection", progress=95)
         await self._replace_in_job(job, db, plan, staged)
+        return None
 
     def _count_downloads(self, job, staged):
         downloaded = staged.stats.get("downloaded", 0)
@@ -518,3 +519,4 @@ class LoraImportRoutesMixin:
             staged.stats.get("download_errors", 0),
         )
         job.update(progress=100, processed=job.total, status_line=line)
+        return None

py/api/lora_integration.py

@@ -44,6 +44,7 @@ _FILES_CHANGED = (
     "The LoRA metadata files changed during the scan; the previous imports "
     "were kept. Run the scan again."
 )
+_LORA_SCAN_FAILED = "The LoRA import failed. Check the server log for details."
 _IMPORTS_CHANGED = (
     "The imported LoRA prompts changed during the scan; the previous imports "
     "were kept. Run the scan again."
@@ -407,10 +408,21 @@ class LoraIntegrationMixin:
             return
         if isinstance(error, PlanChanged):
             message = _IMPORTS_CHANGED
-        elif isinstance(error, (LoraFilesChanged, LoraSettingsUnusable)):
+        elif isinstance(
+            error,
+            (
+                LoraFilesChanged,
+                LoraSettingsUnusable,
+                EmptyReplacement,
+                PairDependencyError,
+            ),
+        ):
             message = str(error)
-        else:
+        elif isinstance(error, OSError):
+            # Reduced to its reason and the file's basename.
             message = self._public_error(error)
+        else:
+            message = _LORA_SCAN_FAILED
         if isinstance(
             error,
             (
@@ -536,6 +548,7 @@ class LoraIntegrationMixin:
                 except asyncio.CancelledError:
                     interrupted = True
                 except Exception:
+                    # the await after this loop raises it for the caller
                     pass
             if interrupted:
                 raise asyncio.CancelledError

py/api/maintenance.py

@@ -1,8 +1,9 @@
 """Maintenance routes: reviewed, read-only plans (#249 System S6).
 
 ``POST /prompt_manager/maintenance/preview {operation, options:{}}`` reviews
-one operation (``duplicate_prompts``, ``optimize``, ``unused_components`` or
-``missing_links``; ``database/maintenance.py``; ``duplicate_files``,
+one operation (``duplicate_prompts``, ``optimize``, ``unused_components``,
+``missing_links`` or ``pair_metadata``, #550; ``database/maintenance.py``;
+``duplicate_files``,
 ``py/api/duplicate_files.py``; ``rotated_logs``, the rotated log copies,
 ``py/api/log_cleanup.py``, which has its own apply job) off the event loop and
 answers a ``Plan``: ``{success, token, operation, expires_at, scope:{count,
@@ -97,7 +98,11 @@ MAINTENANCE_TITLES = {
     "unused_components": "Unused prompt components",
     "missing_links": "Missing-file links",
     "workflow_storage": "Store each workflow once",
+    # #550, beyond the concept: fills old Pair prompts' empty category and tags.
+    "pair_metadata": "Pair category and tags",
 }
+# Operations whose job keeps every changed row for its report (#550).
+COMPLETE_RECORD = frozenset({"pair_metadata"})
 _OPERATION_ERROR = "operation must be one of: " + ", ".join(PREVIEW_OPERATIONS)
 MAINTENANCE_GROUPS = ("changed", "protected", "errors")
 MAX_TOKEN_LENGTH = 128
@@ -281,9 +286,27 @@ def short_space_message(short):
     )
 
 
-def _changed_line(operation, count):
+def _pair_metadata_line(result):
+    count = result["changed"]
+    noun = (
+        "prompt from its saved pairs"
+        if count == 1
+        else "prompts from their saved pairs"
+    )
+    categories = result.get("categories", 0)
+    tag_sets = result.get("tag_sets", 0)
+    return (
+        f"Filled {count:,} {noun} "
+        f"({categories:,} {'category' if categories == 1 else 'categories'}, "
+        f"{tag_sets:,} {'tag set' if tag_sets == 1 else 'tag sets'})"
+    )
+
+
+def _changed_line(operation, count, marks=0, result=None):
     if operation == "optimize":
         return "Optimized the database"
+    if operation == "pair_metadata":
+        return _pair_metadata_line(result or {"changed": count})
     if operation == "duplicate_prompts":
         noun = "duplicate prompt" if count == 1 else "duplicate prompts"
         return f"Merged {count:,} {noun}"
@@ -291,7 +314,11 @@ def _changed_line(operation, count):
         noun = "unused prompt component" if count == 1 else "unused prompt components"
         return f"Removed {count:,} {noun}"
     noun = "link to a missing file" if count == 1 else "links to missing files"
-    return f"Removed {count:,} {noun}"
+    line = f"Removed {count:,} {noun}"
+    if marks:
+        gone = "removed image" if marks == 1 else "removed images"
+        line += f"; forgot {marks:,} {gone} whose files are gone or changed"
+    return line
 
 
 def take_safety_backup(model, label="backup"):
@@ -419,6 +446,7 @@ class MaintenanceRoutesMixin:
             draft.public(),
             {
                 "targets": draft.targets,
+                "marks": draft.marks,
                 "roots": roots,
                 "protected": draft.protected["count"],
                 "rows": draft.scope["count"],
@@ -653,6 +681,7 @@ class MaintenanceRoutesMixin:
                 plan.fingerprint,
                 payload["targets"],
                 payload["roots"],
+                payload.get("marks", ()),
             )
         except PlanChanged:
             return _fail_job(job, "plan-changed", PLAN_CHANGED)
@@ -662,11 +691,31 @@ class MaintenanceRoutesMixin:
         rows = await self._run_in_executor(
             self._maintenance_rows, plan.operation, result["rows"]
         )
-        job.record_counted("changed", result["changed"], rows)
-        job.record_counted("protected", payload["protected"])
-        line = _changed_line(plan.operation, result["changed"])
+        job.record_counted(
+            "changed",
+            result["changed"],
+            rows,
+            complete=plan.operation in COMPLETE_RECORD,
+        )
+        if "kept" in result:
+            # #550: each kept record with its reason, like the changed ones.
+            kept = [
+                {
+                    "path": "",
+                    "detail": f"Prompt {item['prompt_id']}: kept, "
+                    + "; ".join(item["reasons"]),
+                }
+                for item in result["kept"]
+            ]
+            job.record_counted("protected", len(kept), kept, complete=True)
+        else:
+            job.record_counted("protected", payload["protected"])
+        line = _changed_line(
+            plan.operation, result["changed"], result.get("marks", 0), result
+        )
         job.log_line(line)
         job.update(progress=100, status_line=f"{line} · {backup_note(backup)}")
+        return None
 
     def _maintenance_rows(self, operation, described):
         """Outcome rows for the changed records (root-relative paths only)."""
@@ -684,6 +733,8 @@ class MaintenanceRoutesMixin:
                     }
                 )
             return rows
+        if operation == "pair_metadata":
+            return [{"path": "", "detail": _filled_detail(item)} for item in described]
         if operation == "duplicate_prompts":
             return [
                 {
@@ -701,6 +752,16 @@ class MaintenanceRoutesMixin:
         ]
 
 
+def _filled_detail(item):
+    """``Prompt 12: category “portrait”; tags a, b``: one filled prompt (#550)."""
+    parts = []
+    if "category" in item:
+        parts.append(f"category “{item['category']}”")
+    if item.get("tags"):
+        parts.append("tags " + ", ".join(item["tags"]))
+    return f"Prompt {item['prompt_id']}: " + "; ".join(parts)
+
+
 def backup_note(backup):
     """The status line's backup part: the file name, or that there is none."""
     return f"Safety backup {backup}" if backup else NO_BACKUP

py/api/media_target.py

@@ -9,13 +9,15 @@ by the caller.
 import os
 import re
 
-from .images import _forbidden, _has_traversal
+from .images import _file_not_found, _forbidden, _has_traversal
 from .prompts import bad_request
 
 try:
     from ...utils.media_capture import CAPTURE_EXTENSIONS
+    from ...utils.media_folders import in_trash_folder
 except ImportError:
     from utils.media_capture import CAPTURE_EXTENSIONS
+    from utils.media_folders import in_trash_folder
 
 MAX_TARGET_PATH = 1024
 TARGET_FIELDS = frozenset({"root_index", "root_scope", "path"})
@@ -29,7 +31,9 @@ def media_target(data: object, *, media_message: str = TAGGING_MEDIA_MESSAGE):
     Only the listing identity is accepted: an int root index, the root's
     64-hex scope and a bounded root-relative POSIX media path, never an
     absolute, drive-qualified, backslash or ``..`` path (a 400 for a malformed
-    shape, a 403 for an unsafe or non-media path, carrying *media_message*).
+    shape, a 403 for an unsafe or non-media path, carrying *media_message*, a
+    404 for a path in a trash folder, #543). A link or a short-name spelling
+    is refused later by the capture store's alias check.
     """
     if not isinstance(data, dict) or set(data) != TARGET_FIELDS:
         return None, bad_request("target must name root_index, root_scope and path")
@@ -51,4 +55,6 @@ def media_target(data: object, *, media_message: str = TAGGING_MEDIA_MESSAGE):
         return None, _forbidden()
     if os.path.splitext(parts[-1])[1].lower() not in CAPTURE_EXTENSIONS:
         return None, _forbidden(media_message)
+    if in_trash_folder(relative):
+        return None, _file_not_found()
     return (index, scope, relative), None

py/api/pairs.py

@@ -102,7 +102,12 @@ class PairRoutesMixin:
             validate_prompt_text(negative)
         fields = metadata_fields(data)
         pair_id = await self._run_in_executor(
-            self.db.save_prompt_pair, positive, negative, **fields
+            self.db.save_prompt_pair,
+            positive,
+            negative,
+            # The node's Save pair button: as a run, a new prompt gets the metadata.
+            fill_new_positive=True,
+            **fields,
         )
         pair = await self._run_in_executor(self.db.get_prompt_pair, pair_id)
         return web.json_response({"success": True, "pair": pair})

py/api/revalidation.py

@@ -0,0 +1,47 @@
+"""Revalidated responses for the pages and static files PromptManager serves.
+
+``Cache-Control: no-cache`` lets the browser keep a copy but makes it ask the
+server before every reuse, so a PromptManager update reaches the browser on a
+normal reload (#478). The ETag lets an unchanged file answer 304 without a
+body, which keeps that reload cheap.
+
+The tag hashes the bytes rather than the file's mtime and size: an update can
+keep both (an archive extract, a same-length edit within one clock tick), and
+the handlers read the whole file anyway. It is weak because the gzip
+middleware may send the same content in a different encoding.
+"""
+
+import hashlib
+
+from aiohttp import web
+
+REVALIDATE = "no-cache"
+
+
+def entity_tag(body: bytes) -> str:
+    """Weak ETag for ``body``."""
+    return f'W/"{hashlib.sha256(body).hexdigest()[:32]}"'
+
+
+def etag_matches(if_none_match, etag: str) -> bool:
+    """True when an If-None-Match header names ``etag`` (weak comparison)."""
+    if not if_none_match:
+        return False
+    if if_none_match.strip() == "*":
+        return True
+    opaque = etag.removeprefix("W/")
+    return any(
+        candidate.strip().removeprefix("W/") == opaque
+        for candidate in if_none_match.split(",")
+    )
+
+
+def revalidated_response(request, body: bytes, content_type: str, charset=None):
+    """200 with ``body``, or 304 when the request already holds this version."""
+    etag = entity_tag(body)
+    headers = {"Cache-Control": REVALIDATE, "ETag": etag}
+    if etag_matches(request.headers.get("If-None-Match"), etag):
+        return web.Response(status=304, headers=headers)
+    return web.Response(
+        body=body, content_type=content_type, charset=charset, headers=headers
+    )

py/api/review.py

@@ -9,7 +9,10 @@ candidates and the dependency fingerprint. ``POST .../items/{id}/decision``
 records one explicit decision: Apply links the chosen sampler's pair only
 when the capture is the item's bytes, the current file still hashes to them,
 the evidence they yield is unchanged and the ingester's compatible-only rules
-allow it; Leave unresolved links nothing.
+allow it; Leave unresolved links nothing. ``POST .../items/{id}/link`` (#536)
+links a Library prompt the reviewer chose, under the same capture and
+compatibility guards; it is their choice, so what the bytes resolve to is not
+asked.
 
 Targets are formatted from the recorded spelling against the current roots
 (no file access beyond the roots' real paths); a file under no current root
@@ -28,12 +31,13 @@ import urllib.parse
 
 from aiohttp import web
 
-from .annotations import _HEX64, _TOKEN, _UUID, _capture_error, _error
+from .annotations import _capture_error, _error
 from .images import media_thumbnail_url
 from .inspection_service import VIDEO_EXTENSIONS
 from .job_routes import busy_response
 from .jobs import JobBusy
 from .prompts import bad_request
+from .review_guards import UNRESOLVED, _decision_guards, _link_guards
 
 try:
     from ...database.library import display_title
@@ -59,17 +63,6 @@ MAX_CURSOR = 4096
 MAX_BODY_BYTES = 64 * 1024
 # The serialized detail stays under this; past it the detail is ``limited``.
 MAX_DETAIL_BYTES = 8 * 1024 * 1024
-UNRESOLVED = "unresolved"
-_DECISION_FIELDS = frozenset(
-    {
-        "expected_epoch",
-        "expected_revision",
-        "capture_token",
-        "sha256",
-        "dependency_fingerprint",
-        "choice",
-    }
-)
 _CURSOR_SECRET = os.urandom(32)
 # The real path of a spelling; module-level so a test can fake a junction.
 _realpath = os.path.realpath
@@ -84,6 +77,7 @@ _STATUS = {
     "storage-changed": 409,
     "source-changed": 409,
     "capture-expired": 409,
+    "prompt-missing": 409,
 }
 _MESSAGES = {
     "review-missing": "This review item no longer exists. Reload the queue.",
@@ -95,6 +89,9 @@ _MESSAGES = {
         "sampler path cannot be attached safely. The existing links were "
         "preserved."
     ),
+    "prompt-missing": (
+        "That prompt or recipe is no longer in the Library. Search again."
+    ),
     "dependencies-changed": (
         "The image's saved links changed since this item was opened. Reopen it "
         "to see the current association."
@@ -251,6 +248,7 @@ class ReviewRoutesMixin:
         routes.get(ITEMS_URL)(self.list_review_items)
         routes.get(ITEMS_URL + "/{item_id}")(self.get_review_item)
         routes.post(ITEMS_URL + "/{item_id}/decision")(self.decide_review_item)
+        routes.post(ITEMS_URL + "/{item_id}/link")(self.link_review_item)
 
     # -- shared ----------------------------------------------------------------
 
@@ -464,6 +462,43 @@ class ReviewRoutesMixin:
         except JobBusy as busy:
             return busy_response(busy)
 
+    async def link_review_item(self, request: web.Request) -> web.Response:
+        """POST the Library prompt the reviewer chose for one pending item.
+
+        Under the same lease as a decision. The capture must be the item's
+        current bytes; the choice and the file's links are checked in the
+        decision's transaction.
+        """
+        try:
+            async with self._jobs.lease(REVIEW_DECISION_LEASE):
+                return await self._link_review_item(request)
+        except JobBusy as busy:
+            return busy_response(busy)
+
+    async def _link_review_item(self, request: web.Request) -> web.Response:
+        data, error = await self._read_json_body(request, max_bytes=MAX_BODY_BYTES)
+        if error is not None:
+            return error
+        guards, error = _link_guards(data)
+        if error is not None:
+            return error
+        try:
+            item = await self._run_in_executor(
+                self.db.review_item, request.match_info["item_id"]
+            )
+        except Exception:
+            self.logger.exception("Review item read failed")
+            return _failed()
+        if item is None:
+            return _coded("review-missing")
+        verified, error = await self._verified_review_capture(
+            item, guards, need_metadata=True
+        )
+        if error is not None:
+            return error
+        manual = {"positive_id": guards["positive_id"], "pair_id": guards["pair_id"]}
+        return await self._record_decision(item, guards, None, verified, manual)
+
     async def _decide_review_item(self, request: web.Request) -> web.Response:
         data, error = await self._read_json_body(request, max_bytes=MAX_BODY_BYTES)
         if error is not None:
@@ -549,22 +584,25 @@ class ReviewRoutesMixin:
             return None, _coded("file-changed")
         metadata = None
         if need_metadata:
-            if captured is None or not _candidate_still_offered(
-                captured[1], guards["choice"]
+            # A manual link has no ``choice``: it rests on the bytes alone.
+            if captured is None or (
+                "choice" in guards
+                and not _candidate_still_offered(captured[1], guards["choice"])
             ):
                 return None, _coded("review-conflict", "evidence-changed")
             metadata = captured[0]
         image_path = os.path.join(str(root), *relative.split("/"))
         return (capture, facts, image_path, metadata), None
 
-    def _decide_sync(self, item, guards, candidate, verified):
+    def _decide_sync(self, item, guards, candidate, verified, manual=None):
         """Worker thread: resolve the file's spellings, decide, build the view.
 
         Returns ``(result, view)``; a refusal raises with ``exc.view`` set.
         """
         _capture, facts, image_path, metadata = verified or (None, None, None, None)
         aliases, known = (), None
-        if candidate is not None:
+        selected = candidate is not None or manual is not None
+        if selected:
             aliases, known = physical_spellings(self.db, image_path, item["locator"])
         try:
             result = self.db.decide_review_item(
@@ -573,8 +611,9 @@ class ReviewRoutesMixin:
                 expected_revision=guards["expected_revision"],
                 # The bytes the re-hash proved, or the body's for Leave unresolved.
                 sha256=facts.sha256 if facts is not None else guards["sha256"],
-                outcome="selected" if candidate is not None else UNRESOLVED,
+                outcome="selected" if selected else UNRESOLVED,
                 candidate=candidate,
+                manual=manual,
                 image_path=image_path,
                 metadata=metadata,
                 expected_fingerprint=guards["dependency_fingerprint"],
@@ -589,11 +628,11 @@ class ReviewRoutesMixin:
         roots, real_roots = self._review_roots()
         return result, self._review_view(result["item"], roots, real_roots)
 
-    async def _record_decision(self, item, guards, candidate, verified):
+    async def _record_decision(self, item, guards, candidate, verified, manual=None):
         capture = verified[0] if verified else None
         try:
             result, view = await self._run_in_executor(
-                self._decide_sync, item, guards, candidate, verified
+                self._decide_sync, item, guards, candidate, verified, manual
             )
         except ReviewCommandError as exc:
             return self._decision_refused(exc)
@@ -629,36 +668,6 @@ class ReviewRoutesMixin:
         return _coded(exc.code, message, **extra)
 
 
-def _decision_guards(data: dict):
-    """The validated decision fields, or ``(None, error)``."""
-    if set(data) != _DECISION_FIELDS:
-        return None, bad_request(
-            "Request must contain exactly " + ", ".join(sorted(_DECISION_FIELDS))
-        )
-    epoch, revision = data["expected_epoch"], data["expected_revision"]
-    if not isinstance(epoch, str) or _UUID.fullmatch(epoch) is None:
-        return None, bad_request("expected_epoch must be a canonical UUID")
-    if type(revision) is not int or revision < 0:
-        return None, bad_request("expected_revision must be a non-negative integer")
-    token = data["capture_token"]
-    if token is not None and (
-        not isinstance(token, str) or _TOKEN.fullmatch(token) is None
-    ):
-        return None, bad_request("Invalid capture token")
-    for name in ("sha256", "dependency_fingerprint"):
-        value = data[name]
-        if not isinstance(value, str) or _HEX64.fullmatch(value) is None:
-            return None, bad_request(
-                f"{name} must be 64 lowercase hexadecimal characters"
-            )
-    choice = data["choice"]
-    if not isinstance(choice, str) or (
-        choice != UNRESOLVED and _HEX64.fullmatch(choice) is None
-    ):
-        return None, bad_request("choice must be a candidate hash or unresolved")
-    return {**data}, None
-
-
 def _verify_and_read(store, capture, item, root, need_metadata: bool):
     """Worker thread: ``(relative, facts, captured)``, or None off the root.
 

py/api/review_guards.py

@@ -0,0 +1,66 @@
+"""Body validation of Needs review decisions and manual links (#249 E3, #536)."""
+
+from .annotations import _HEX64, _TOKEN, _UUID
+from .prompts import bad_request
+
+UNRESOLVED = "unresolved"
+_GUARD_FIELDS = frozenset(
+    {
+        "expected_epoch",
+        "expected_revision",
+        "capture_token",
+        "sha256",
+        "dependency_fingerprint",
+    }
+)
+_DECISION_FIELDS = _GUARD_FIELDS | {"choice"}
+_LINK_FIELDS = _GUARD_FIELDS | {"positive_id", "pair_id"}
+
+
+def _guard_error(data: dict, fields: frozenset):
+    """The first problem with a decision's shared fields, as a response."""
+    if set(data) != fields:
+        return bad_request("Request must contain exactly " + ", ".join(sorted(fields)))
+    epoch, revision = data["expected_epoch"], data["expected_revision"]
+    if not isinstance(epoch, str) or _UUID.fullmatch(epoch) is None:
+        return bad_request("expected_epoch must be a canonical UUID")
+    if type(revision) is not int or revision < 0:
+        return bad_request("expected_revision must be a non-negative integer")
+    token = data["capture_token"]
+    if token is not None and (
+        not isinstance(token, str) or _TOKEN.fullmatch(token) is None
+    ):
+        return bad_request("Invalid capture token")
+    for name in ("sha256", "dependency_fingerprint"):
+        value = data[name]
+        if not isinstance(value, str) or _HEX64.fullmatch(value) is None:
+            return bad_request(f"{name} must be 64 lowercase hexadecimal characters")
+    return None
+
+
+def _decision_guards(data: dict):
+    """The validated decision fields, or ``(None, error)``."""
+    error = _guard_error(data, _DECISION_FIELDS)
+    if error is not None:
+        return None, error
+    choice = data["choice"]
+    if not isinstance(choice, str) or (
+        choice != UNRESOLVED and _HEX64.fullmatch(choice) is None
+    ):
+        return None, bad_request("choice must be a candidate hash or unresolved")
+    return {**data}, None
+
+
+def _link_guards(data: dict):
+    """The validated manual link fields, or ``(None, error)``."""
+    error = _guard_error(data, _LINK_FIELDS)
+    if error is not None:
+        return None, error
+    if data["capture_token"] is None:
+        return None, bad_request("A link needs a capture of the file")
+    positive, pair = data["positive_id"], data["pair_id"]
+    if type(positive) is not int or positive < 1:
+        return None, bad_request("positive_id must be a positive integer")
+    if pair is not None and (type(pair) is not int or pair < 1):
+        return None, bad_request("pair_id must be a positive integer or null")
+    return {**data}, None

py/api/settings_routes.py

@@ -54,9 +54,25 @@ _SETTINGS_FILE_KEYS = {
     "AUTOTAG_EXISTING": ("autotag", "defaults", "existing"),
     "AUTOTAG_GENERAL_THRESHOLD": ("autotag", "defaults", "general_threshold"),
     "AUTOTAG_CHARACTER_THRESHOLD": ("autotag", "defaults", "character_threshold"),
-    "AUTOTAG_TEMPLATE": ("autotag", "defaults", "template"),
     "AUTOTAG_SUGGESTION_LIMIT": ("autotag", "defaults", "suggestion_limit"),
+    "AUTOTAG_SKIPPED_TAGS": ("autotag", "defaults", "skipped_tags"),
+    # #431: the saved instructions and each engine's default, always written
+    # together (server-side pairing).
+    "AUTOTAG_INSTRUCTIONS": ("autotag", "defaults", "instructions"),
+    "AUTOTAG_DEFAULT_INSTRUCTIONS": ("autotag", "defaults", "default_instructions"),
 }
+# The old single template: written only as the GGUF default's mirror for a
+# downgrade (#431 B1.1), deleted when that default is Built-in.
+_TEMPLATE_MIRROR_KEY = ("autotag", "defaults", "template")
+TEMPLATE_REPLACED_ERROR = (
+    "LLM tag template was replaced by Saved instructions. Reload the page."
+)
+# Whole-list settings (#431): a save without ``expected_revision`` would merge
+# a stale copy over a list changed elsewhere, so it is refused.
+_REVISION_REQUIRED = frozenset(
+    {"autotag_skipped_tags", "autotag_instructions", "autotag_default_instructions"}
+)
+REVISION_REQUIRED_ERROR = "Reload the page and try again."
 # The thumbnail settings a running Convert outdated job fixes (#317).
 _THUMBNAIL_SETTINGS = ("THUMBNAIL_FORMAT", "THUMBNAIL_QUALITY", "THUMBNAIL_EDGE")
 # Attributes that live on GalleryConfig; the others on PromptManagerConfig.
@@ -137,13 +153,20 @@ def _settings_mutation(updates, new_roots):
     """The config commit for a settings save: only the submitted keys change.
 
     Every other section and key (integrations and the CivitAI key, database,
-    unknown sections) is kept as it is on disk.
+    unknown sections) is kept as it is on disk. A save of the instructions
+    also writes the old ``template`` key as the GGUF default's mirror.
     """
 
     def mutate(config):
         for name, value in updates.items():
             *sections, key = _SETTINGS_FILE_KEYS[name]
-            _set_nested(config, sections, key, value)
+            _set_nested(config, sections, key, _jsonable(value))
+        if "AUTOTAG_INSTRUCTIONS" in updates:
+            *sections, key = _TEMPLATE_MIRROR_KEY
+            mirror = _mirror_text(
+                updates["AUTOTAG_INSTRUCTIONS"], updates["AUTOTAG_DEFAULT_INSTRUCTIONS"]
+            )
+            _set_nested(config, sections, key, _DELETE if mirror is None else mirror)
         if new_roots is not None:
             gallery = _section(config, "gallery")
             monitoring = _section(gallery, "monitoring")
@@ -155,10 +178,37 @@ def _settings_mutation(updates, new_roots):
     return mutate
 
 
+# ``_set_nested``'s value that removes the key instead of setting it.
+_DELETE = object()
+
+
+def _jsonable(value):
+    """A runtime value as config.json holds it (tuples become lists)."""
+    if isinstance(value, (list, tuple)):
+        return [_jsonable(item) for item in value]
+    if isinstance(value, dict):
+        return {key: _jsonable(item) for key, item in value.items()}
+    return value
+
+
+def _mirror_text(entries, defaults):
+    """The GGUF default's text, or None when it is Built-in (#431 B1.1)."""
+    from ..tagging_defaults import BUILTIN_INSTRUCTIONS_ID
+
+    for item in entries:
+        if item["id"] == defaults["gguf"] != BUILTIN_INSTRUCTIONS_ID:
+            return item["text"]
+    return None
+
+
 def _set_nested(config, sections, key, value):
-    """``config[s1][s2]...[key] = value``, each section a fresh dict."""
+    """``config[s1][s2]...[key] = value``, each section a fresh dict;
+    ``_DELETE`` removes the key."""
     if not sections:
-        config[key] = value
+        if value is _DELETE:
+            config.pop(key, None)
+        else:
+            config[key] = value
         return
     merged = _section(config, sections[0])
     _set_nested(merged, sections[1:], key, value)
@@ -176,7 +226,7 @@ def _validate_autotag_settings(data):
         MAX_SUGGESTION_LIMIT,
         MODEL_TYPES,
         validate_saved_threshold,
-        validate_template,
+        validate_skipped_tags,
     )
 
     updates = {}
@@ -205,12 +255,7 @@ def _validate_autotag_settings(data):
                 raise ValueError(f"{error}.") from None
             updates[key.upper()] = value
     if "autotag_template" in data:
-        try:
-            updates["AUTOTAG_TEMPLATE"] = validate_template(
-                data["autotag_template"], "LLM tag template"
-            )
-        except ValueError as error:
-            raise ValueError(f"{error}.") from None
+        raise ValueError(TEMPLATE_REPLACED_ERROR)
     if "autotag_suggestion_limit" in data:
         value = data["autotag_suggestion_limit"]
         if type(value) is not int or not 1 <= value <= MAX_SUGGESTION_LIMIT:
@@ -219,9 +264,41 @@ def _validate_autotag_settings(data):
                 f"{MAX_SUGGESTION_LIMIT}."
             )
         updates["AUTOTAG_SUGGESTION_LIMIT"] = value
+    if "autotag_skipped_tags" in data:
+        updates["AUTOTAG_SKIPPED_TAGS"] = tuple(
+            validate_skipped_tags(data["autotag_skipped_tags"])
+        )
+    updates.update(_validate_instruction_settings(data))
     return updates
 
 
+def _validate_instruction_settings(data):
+    """Server-side pairing (#431): a save that carries either instruction key
+    gets the other from the running configuration (which holds the migrated
+    list), the defaults are checked against the merged list, and both are
+    written. Empty when the save carries neither."""
+    from ..config import AutotagConfig
+    from ..tagging_defaults import (
+        validate_default_instructions,
+        validate_instructions,
+    )
+
+    if not {"autotag_instructions", "autotag_default_instructions"} & set(data):
+        return {}
+    if "autotag_instructions" in data:
+        entries = validate_instructions(data["autotag_instructions"])
+    else:
+        entries = [dict(item) for item in AutotagConfig.INSTRUCTIONS]
+    defaults = data.get(
+        "autotag_default_instructions", dict(AutotagConfig.DEFAULT_INSTRUCTIONS)
+    )
+    defaults = validate_default_instructions(defaults, entries)
+    return {
+        "AUTOTAG_INSTRUCTIONS": tuple(entries),
+        "AUTOTAG_DEFAULT_INSTRUCTIONS": defaults,
+    }
+
+
 async def _read_json(request):
     """Parse a JSON object body through the package's capped reader."""
     from . import _read_json_body
@@ -249,6 +326,7 @@ class SettingsRoutesMixin:
         """Get current settings."""
         try:
             from ..config import AutotagConfig, PromptManagerConfig, GalleryConfig
+            from ..tagging_defaults import DEFAULT_TAG_TEMPLATE
 
             root_paths = [
                 self._public_path(d) for d in GalleryConfig.MONITORING_DIRECTORIES
@@ -281,8 +359,21 @@ class SettingsRoutesMixin:
                         "autotag_character_threshold": (
                             AutotagConfig.CHARACTER_THRESHOLD
                         ),
-                        "autotag_template": AutotagConfig.TEMPLATE,
+                        # Read-only for one release (#431): a page cached
+                        # from before the upgrade still renders.
+                        "autotag_template": AutotagConfig.default_instructions_text(
+                            "gguf"
+                        ),
+                        "autotag_instructions": [
+                            dict(item) for item in AutotagConfig.INSTRUCTIONS
+                        ],
+                        "autotag_default_instructions": dict(
+                            AutotagConfig.DEFAULT_INSTRUCTIONS
+                        ),
+                        # Built-in's text, which System shows read-only.
+                        "autotag_builtin_instructions": DEFAULT_TAG_TEMPLATE,
                         "autotag_suggestion_limit": AutotagConfig.SUGGESTION_LIMIT,
+                        "autotag_skipped_tags": list(AutotagConfig.SKIPPED_TAGS),
                         "cpu_count": PromptManagerConfig.max_worker_threads(),
                         "gallery_root_paths": root_paths,
                         "gallery_root_path": root_paths[0] if root_paths else "",
@@ -383,6 +474,10 @@ class SettingsRoutesMixin:
                     {"success": False, "error": "expected_revision must be a string"},
                     status=400,
                 )
+            if expected_revision is None and _REVISION_REQUIRED & set(data):
+                return web.json_response(
+                    {"success": False, "error": REVISION_REQUIRED_ERROR}, status=400
+                )
             try:
                 updates = self._validate_settings(data)
             except ValueError as ve:
@@ -511,6 +606,12 @@ class SettingsRoutesMixin:
             if name in _AUTOTAG_SETTINGS:
                 setattr(AutotagConfig, name[len(_AUTOTAG_PREFIX) :], value)
                 continue
+        if "AUTOTAG_INSTRUCTIONS" in updates:
+            # What config.json now holds as the old key (the mirror).
+            AutotagConfig._LEGACY_TEMPLATE = AutotagConfig.mirror_template()
+        for name, value in updates.items():
+            if name in _AUTOTAG_SETTINGS:
+                continue
             owner = GalleryConfig if name in _GALLERY_SETTINGS else PromptManagerConfig
             setattr(owner, name, value)
         restart_fields = []

py/api/source_rescope.py

@@ -41,6 +41,13 @@ def _coded(status: int, code: str, message: str) -> web.Response:
     )
 
 
+def _missing_if_not_found(response: web.Response) -> web.Response:
+    """A file in a trash folder answers as a missing file does (#543)."""
+    if response.status == 404:
+        return _coded(404, "file-missing", "Image file not found")
+    return response
+
+
 def _no_store(response: web.Response) -> web.Response:
     response.headers["Cache-Control"] = "no-store"
     return response
@@ -71,7 +78,7 @@ class SourceRescopeMixin:
         path = root / relative
         denied = _media_access_error(path, [root])
         if denied is not None:
-            return denied
+            return _missing_if_not_found(denied)
         try:
             mtime = path.stat().st_mtime if path.is_file() else None
         except OSError:
@@ -89,7 +96,7 @@ class SourceRescopeMixin:
             return bad_request("Send exactly one target")
         target, error = media_target(data["target"])
         if error is not None:
-            return error
+            return _missing_if_not_found(error)
         old_index, scope, relative = target
         if old_index > MAX_ROOT_INDEX:
             return bad_request("Invalid gallery root")

py/api/thumbnail_auto.py

@@ -8,7 +8,8 @@ previews behind those URLs are made here, in three ways:
 - **On first view.** ``GET /prompt_manager/images/serve/thumbnails/<rel>_thumb
   <ext>`` for a preview that does not exist yet is admitted on the event loop
   (``_PreviewGate``): at most ``ON_DEMAND_LIMIT`` previews are written at once,
-  on this API's own small pool (never the default executor), one flight runs
+  on this API's interactive image pool (``IMAGE_POOL_SIZE`` threads, shared
+  with card facts and inspector reads; never the default executor), one flight runs
   per file, and the requests waiting are capped (``MAX_WAITING``); past the
   cap a request gets its fallback at once. Inside the admission the preview is
   mapped back to its media file under the same root (the request text is
@@ -24,7 +25,8 @@ previews behind those URLs are made here, in three ways:
 - **On capture.** The image monitor hands every new output it settles to
   ``queue_capture_thumbnail``, which queues its preview for one background
   worker thread (``AutoThumbnailer``). Each wake takes at most Worker threads
-  (``WORKER_THREADS``) files and writes them on that many threads; enqueuing
+  (``WORKER_THREADS``) files and writes them on the long-lived job pool
+  (``utils.parallel``, shared with scans and thumbnail jobs); enqueuing
   is O(1), so the monitor and ComfyUI's queue never wait. A file already
   queued is not queued again, the queue is bounded (``MAX_QUEUED``) and a
   preview left out is made on its first view instead.
@@ -49,7 +51,7 @@ write starts (after its slot) and again just before its rename.
 
 Shutdown: ``close_auto_thumbnails`` stops the worker at a file boundary,
 kills a running ffmpeg, waits a bounded time (and logs when the wait ran
-out) and shuts the view pool down; after the app's cleanup
+out) and shuts the image pool down; after the app's cleanup
 (``final=True``) nothing starts a worker again. Importing this module again
 (a package reload) stops the workers the previous import left.
 """
@@ -60,13 +62,14 @@ import functools
 import io
 import os
 import threading
-from concurrent.futures import ThreadPoolExecutor
+from concurrent.futures import CancelledError, ThreadPoolExecutor
 from pathlib import Path
 
 from aiohttp import web
 
 from . import thumbnail_settings
 from .images import (
+    in_trash_below_roots,
     MEDIA_EXTENSIONS,
     THUMBNAIL_CLEAR_TITLE,
     VIDEO_EXTENSIONS,
@@ -103,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.
@@ -123,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()
 
@@ -224,6 +230,8 @@ def _quick_source(root, parts):
         for name in (base + extension, base + extension.upper()):
             path = folder / name
             if _regular_file(path) and _is_within(path, root):
+                if in_trash_below_roots(path, [root]):
+                    return None  # also the fallback past the gate's cap (#543)
                 return path, extension in VIDEO_EXTENSIONS
     return None
 
@@ -248,6 +256,8 @@ def _thumbnail_source(root, parts):
         found = _scan_folder(root.joinpath(*folder_parts), base, set(wanted))
         if found is None or not _is_within(found, root):
             return None
+    if in_trash_below_roots(found, [root]):
+        return None
     rel = Path(*folder_parts, found.name)
     dst = root / media_thumbnail_rel(rel)
     if dst != _current_preview(root, parts):
@@ -289,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.
     """
 
@@ -486,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
@@ -528,6 +541,7 @@ class AutoThumbnailer:
             try:
                 process.kill()
             except OSError:
+                # the process has already exited
                 pass
         if worker is None:
             return True
@@ -561,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.
     """
 
@@ -616,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):
@@ -645,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
@@ -656,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:
@@ -700,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):
@@ -724,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):
@@ -768,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(
@@ -835,7 +904,7 @@ class AutoThumbnailMixin:
             if not folded.startswith(base) or not _is_within(path, root):
                 continue
             rel = Path(os.path.relpath(path, root))
-            if "thumbnails" in rel.parts[:-1]:
+            if "thumbnails" in rel.parts[:-1] or in_trash_below_roots(path, [root]):
                 return None
             dst = root / media_thumbnail_rel(rel)
             if not _is_within(root / "thumbnails", root) or not _is_within(

py/api/thumbnail_convert.py

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

py/api/thumbnail_encode.py

@@ -158,6 +158,7 @@ def _unlink_quietly(path):
     try:
         Path(path).unlink()
     except FileNotFoundError:
+        # already gone, which is what was wanted
         pass
     except OSError as exc:
         logger.debug(f"Could not remove temp file {Path(path).name}: {exc}")
@@ -221,6 +222,7 @@ def shutdown_encoders():
         try:
             process.kill()
         except OSError:
+            # the process has already exited
             pass
 
 
@@ -359,10 +361,12 @@ def thumbnail_frame(img, edge):
     try:
         img.seek(0)
     except (EOFError, AttributeError, ValueError):
+        # not seekable or a single frame: use the current one
         pass
     try:
         img.draft("RGB", (edge, edge))  # JPEG: decode at a reduced scale
     except (AttributeError, ValueError, OSError):
+        # draft is only a speedup; the full decode below still works
         pass
     frame = img.convert("RGB") if img.mode != "RGB" else img.copy()
     frame.thumbnail((edge, edge), Image.Resampling.LANCZOS)

py/api/thumbnail_jobs.py

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

py/api/workflow_storage.py

@@ -275,3 +275,4 @@ class WorkflowStorageRoutesMixin:
         job.record_counted("changed", 1)
         job.log_line(line)
         job.update(progress=100, status_line=line)
+        return None

py/autotag.py

@@ -30,9 +30,9 @@ except ImportError:
 # The tag template and thresholds live with the saved tagging defaults
 # (#328), so config.py never imports PIL.
 try:
-    from .tagging_defaults import DEFAULT_TAG_TEMPLATE
+    from .tagging_defaults import DEFAULT_TAG_TEMPLATE, read_wd14_label_rows
 except ImportError:
-    from py.tagging_defaults import DEFAULT_TAG_TEMPLATE
+    from py.tagging_defaults import DEFAULT_TAG_TEMPLATE, read_wd14_label_rows
 
 try:
     from ..utils import exit_signal
@@ -559,7 +559,6 @@ class AutoTagService:
         # A missing onnxruntime is refused by load_model before this runs
         # (#424); a ModuleNotFoundError here becomes the same refusal.
         import onnxruntime as ort
-        import csv
 
         config = MODELS[model_type]
         model_dir = self.models_dir / config["subdir"]
@@ -576,12 +575,8 @@ class AutoTagService:
 
         session = ort.InferenceSession(str(model_path), providers=providers)
 
-        # Read tag list from CSV
-        tags_list = []
-        with open(tags_path, "r", encoding="utf-8") as f:
-            reader = csv.DictReader(f)
-            for row in reader:
-                tags_list.append(row)
+        # Read tag list from CSV (the reader the Skipped tags preview shares)
+        tags_list = read_wd14_label_rows(tags_path)
 
         self.logger.info(
             f"WD14 model loaded ({len(tags_list)} tags, "
@@ -992,6 +987,7 @@ class AutoTagService:
         prompt: Optional[str] = None,
         general_threshold: Optional[float] = None,
         character_threshold: Optional[float] = None,
+        skip: Any = None,
     ) -> List[Tuple[str, Optional[float]]]:
         """The top ``limit`` suggestions as ``(name, confidence)`` pairs.
 
@@ -1000,7 +996,9 @@ class AutoTagService:
         ``None``. Names are distinct case-insensitively (first one wins).
         A name Save would reject (over 50 characters, a control character
         such as a newline, a comma) is skipped before the dedupe and the
-        ``limit`` count. ``prompt`` is used for this call only;
+        ``limit`` count. A name ``skip`` matches (a ``SkipMatcher``, the
+        Skipped tags compiled for this request or job, #431) is dropped next,
+        so it uses up no slot. ``prompt`` is used for this call only;
         ``custom_prompt`` is never changed.
         """
         if type(limit) is not int or not 1 <= limit <= MAX_SUGGESTION_LIMIT:
@@ -1017,6 +1015,8 @@ class AutoTagService:
                 name = validate_vocabulary_tag_name(raw)
             except ValueError:
                 continue
+            if skip is not None and skip.matches(name):
+                continue
             key = name.casefold()
             if key in seen:
                 continue

py/config.py

@@ -30,6 +30,7 @@ routes = server_instance.routes
 # Extension info
 extension_uri = None  # Will be set in __init__.py
 
+import hashlib  # noqa: E402
 import os  # noqa: E402
 from typing import Any, Dict, List, Optional, Tuple  # noqa: E402
 
@@ -587,8 +588,15 @@ try:
         MODEL_TYPES,
         WD14_MODEL_TYPES,
         validate_saved_threshold,
+        validate_skipped_tags,
         validate_template,
         validate_threshold,
+        BUILTIN_INSTRUCTIONS_ID,
+        INSTRUCTION_ENGINES,
+        MAX_SAVED_INSTRUCTIONS,
+        builtin_instructions,
+        validate_default_instructions,
+        validate_instructions,
     )
 except ImportError:
     from py.tagging_defaults import (  # noqa: F401 - re-exported
@@ -600,8 +608,15 @@ except ImportError:
         MODEL_TYPES,
         WD14_MODEL_TYPES,
         validate_saved_threshold,
+        validate_skipped_tags,
         validate_template,
         validate_threshold,
+        BUILTIN_INSTRUCTIONS_ID,
+        INSTRUCTION_ENGINES,
+        MAX_SAVED_INSTRUCTIONS,
+        builtin_instructions,
+        validate_default_instructions,
+        validate_instructions,
     )
 
 
@@ -617,6 +632,31 @@ def _valid_existing(value: Any) -> str:
     return value
 
 
+def _valid_skipped_tags(value: Any) -> Tuple[str, ...]:
+    return tuple(validate_skipped_tags(value))
+
+
+def _reason(error: ValueError) -> str:
+    """A settings sentence as a load warning's reason ("entry 2 is empty")."""
+    text = str(error).rstrip(".")
+    for prefix in ("Skipped tags", "Saved instructions"):
+        if text.startswith(prefix):
+            return text.removeprefix(prefix).lstrip(": ")
+    return text
+
+
+def _text_id(text: str) -> str:
+    """A migrated entry's id: deterministic, so every load gives the same."""
+    return hashlib.sha256(text.encode("utf-8")).hexdigest()[:12]
+
+
+def _fallback_warning(key: str, error: ValueError, default: Any) -> str:
+    """The one load warning of a saved tagging default that fails its check."""
+    if key == "skipped_tags":
+        return f"autotag.defaults.skipped_tags {_reason(error)}; using no skipped tags"
+    return f"autotag.defaults.{key} {error}; using {default!r}"
+
+
 def _valid_suggestion_limit(value: Any) -> int:
     if type(value) is not int or not 1 <= value <= MAX_SUGGESTION_LIMIT:
         raise ValueError(f"must be an integer from 1 to {MAX_SUGGESTION_LIMIT}")
@@ -637,8 +677,18 @@ class AutotagConfig:
     EXISTING = "skip"
     GENERAL_THRESHOLD = 0.35
     CHARACTER_THRESHOLD = 0.85
-    TEMPLATE = DEFAULT_TAG_TEMPLATE
     SUGGESTION_LIMIT = DEFAULT_SUGGESTION_LIMIT
+    SKIPPED_TAGS: Tuple[str, ...] = ()
+    # #431: the saved instructions (Built-in is never stored) and the default
+    # of each language-model engine, by id.
+    INSTRUCTIONS: Tuple[Dict[str, str], ...] = ()
+    DEFAULT_INSTRUCTIONS: Dict[str, str] = {
+        "gguf": BUILTIN_INSTRUCTIONS_ID,
+        "hf": BUILTIN_INSTRUCTIONS_ID,
+    }
+    # The old single template as read at load (None when absent or invalid);
+    # only the migration reads it. Saves keep it as the GGUF default's mirror.
+    _LEGACY_TEMPLATE: Optional[str] = None
 
     # config.json key -> (attribute, validator, default)
     _KEYS = {
@@ -654,12 +704,13 @@ class AutotagConfig:
             lambda v: validate_saved_threshold("value", v),
             0.85,
         ),
-        "template": ("TEMPLATE", validate_template, DEFAULT_TAG_TEMPLATE),
         "suggestion_limit": (
             "SUGGESTION_LIMIT",
             _valid_suggestion_limit,
             DEFAULT_SUGGESTION_LIMIT,
         ),
+        # #431: words whose suggested tags are never offered.
+        "skipped_tags": ("SKIPPED_TAGS", _valid_skipped_tags, ()),
     }
 
     @classmethod
@@ -672,17 +723,142 @@ class AutotagConfig:
             type(value) is int and 1 <= value <= MAX_SUGGESTION_LIMIT
         )  # bool, float and str are rejected
 
+    @classmethod
+    def all_instructions(cls) -> List[Dict[str, str]]:
+        """Built-in first, then the saved entries (copies)."""
+        return [builtin_instructions()] + [dict(item) for item in cls.INSTRUCTIONS]
+
+    @classmethod
+    def default_instructions_text(cls, engine: str) -> str:
+        """The text of ``engine``'s (``gguf`` or ``hf``) default instructions."""
+        if engine not in INSTRUCTION_ENGINES:
+            raise ValueError(f"{engine} takes no instructions")
+        wanted = cls.DEFAULT_INSTRUCTIONS.get(engine, BUILTIN_INSTRUCTIONS_ID)
+        for item in cls.INSTRUCTIONS:
+            if item["id"] == wanted:
+                return item["text"]
+        return DEFAULT_TAG_TEMPLATE
+
+    @classmethod
+    def mirror_template(cls) -> Optional[str]:
+        """What an older build reads as its one template (#431 B1.1): the
+        GGUF default's text, or None (no key) when that default is Built-in."""
+        if cls.DEFAULT_INSTRUCTIONS.get("gguf") == BUILTIN_INSTRUCTIONS_ID:
+            return None
+        return cls.default_instructions_text("gguf")
+
     @classmethod
     def get_config(cls) -> Dict[str, Any]:
-        """The saved defaults; the template only when it is not the built-in
-        one, so a later built-in template still reaches who never set one."""
+        """The saved defaults; ``template`` only as the GGUF default's mirror."""
         defaults = {
             key: getattr(cls, attr) for key, (attr, _v, _d) in cls._KEYS.items()
         }
-        if defaults["template"] == DEFAULT_TAG_TEMPLATE:
-            del defaults["template"]
+        defaults["skipped_tags"] = list(defaults["skipped_tags"])
+        defaults["instructions"] = [dict(item) for item in cls.INSTRUCTIONS]
+        defaults["default_instructions"] = dict(cls.DEFAULT_INSTRUCTIONS)
+        mirror = cls.mirror_template()
+        if mirror is not None:
+            defaults["template"] = mirror
         return {"defaults": defaults}
 
+    @classmethod
+    def _load_instructions(cls, defaults: Dict[str, Any]) -> None:
+        """Each instruction key alone, then the cross-key default check, then
+        the in-memory migration of the old template (#431). Nothing is
+        written here, so the revision never changes at load."""
+        cls._LEGACY_TEMPLATE = None
+        if "template" in defaults:
+            try:
+                cls._LEGACY_TEMPLATE = validate_template(defaults["template"])
+            except ValueError as error:
+                config_logger.warning(
+                    f"autotag.defaults.template {error}; using the built-in one"
+                )
+        entries: List[Dict[str, str]] = []
+        damaged = False
+        if "instructions" in defaults:
+            try:
+                entries = validate_instructions(defaults["instructions"])
+            except ValueError as error:
+                damaged = True
+                config_logger.warning(
+                    f"autotag.defaults.instructions {_reason(error)}; "
+                    "using Built-in only"
+                )
+        chosen = {engine: BUILTIN_INSTRUCTIONS_ID for engine in INSTRUCTION_ENGINES}
+        stored = defaults.get("default_instructions")
+        if damaged:
+            # Both defaults fall back with the list (its one warning says so);
+            # the old template is migrated as on a first upgrade.
+            pass
+        elif "default_instructions" in defaults and (
+            not isinstance(stored, dict) or set(stored) != set(INSTRUCTION_ENGINES)
+        ):
+            config_logger.warning(
+                "autotag.defaults.default_instructions must name gguf and hf; "
+                "using Built-in"
+            )
+        elif isinstance(stored, dict):
+            ids = {item["id"] for item in entries}
+            for engine in INSTRUCTION_ENGINES:
+                value = stored[engine]
+                if isinstance(value, str) and (
+                    value == BUILTIN_INSTRUCTIONS_ID or value in ids
+                ):
+                    chosen[engine] = value
+                else:
+                    config_logger.warning(
+                        f"autotag.defaults.default_instructions.{engine} names no "
+                        "saved instructions; using Built-in"
+                    )
+        cls._migrate_template(
+            entries,
+            chosen,
+            first=damaged or "instructions" not in defaults,
+            keep_defaults=not damaged and "default_instructions" in defaults,
+        )
+        cls.INSTRUCTIONS = tuple(entries)
+        cls.DEFAULT_INSTRUCTIONS = chosen
+
+    @classmethod
+    def _migrate_template(cls, entries, chosen, *, first, keep_defaults) -> None:
+        """Bring a template saved by an older build in as a named entry."""
+        legacy = cls._LEGACY_TEMPLATE
+        if legacy is None or legacy == DEFAULT_TAG_TEMPLATE:
+            return
+        if first:
+            # The first upgrade: it becomes "Custom template", the default for
+            # both engines unless defaults are stored already (20A).
+            entries.append(
+                {"id": _text_id(legacy), "name": "Custom template", "text": legacy}
+            )
+            if not keep_defaults:
+                for engine in INSTRUCTION_ENGINES:
+                    chosen[engine] = entries[-1]["id"]
+            return
+        if any(item["text"] == legacy for item in entries):
+            return  # the mirror of a saved default
+        # Edited on an older build after an upgrade (B1.2): import it.
+        if len(entries) >= MAX_SAVED_INSTRUCTIONS:
+            config_logger.warning(
+                "The tagging template edited on an older version was not imported: "
+                f"the saved instructions are full ({MAX_SAVED_INSTRUCTIONS})"
+            )
+            return
+        names = {item["name"].casefold() for item in entries}
+        number = 2
+        while f"custom template ({number})" in names:
+            number += 1
+        name = f"Custom template ({number})"
+        ids = {item["id"] for item in entries}
+        new_id, n = _text_id(legacy), 2
+        while new_id in ids:
+            new_id, n = _text_id(f"{legacy}\n{n}"), n + 1
+        entries.append({"id": new_id, "name": name, "text": legacy})
+        config_logger.info(
+            "Imported the tagging template edited on an older version as " f"'{name}'"
+        )
+
     @classmethod
     def update_config(cls, new_config: Any) -> None:
         if not isinstance(new_config, dict):
@@ -698,12 +874,9 @@ class AutotagConfig:
             try:
                 setattr(cls, attr, validate(defaults[key]))
             except ValueError as error:
-                config_logger.warning(
-                    f"autotag.defaults.{key} {error}; using {default!r}"
-                    if key != "template"
-                    else f"autotag.defaults.template {error}; using the built-in one"
-                )
+                config_logger.warning(_fallback_warning(key, error, default))
                 setattr(cls, attr, default)
+        cls._load_instructions(defaults)
 
 
 class PromptManagerConfig:

py/config_store.py

@@ -109,6 +109,7 @@ def _write_atomically(path: str, payload: bytes) -> None:
         try:
             os.unlink(temp_path)
         except OSError:
+            # nothing to clean up; the original error is raised below
             pass
         raise
 

py/lora_manager_layout.py

@@ -642,6 +642,7 @@ def layout_summary(layout: LoraManagerLayout, home: Path) -> Dict:
             relative = Path(settings_file).relative_to(home)
             settings_file = os.path.join("~", str(relative))
         except ValueError:
+            # outside the home folder: show the full path
             pass
     return {
         "source": layout.source,

py/lora_utils.py

@@ -585,6 +585,7 @@ def _read_capped(resp, cap: int) -> Optional[bytes]:
             if int(declared) > cap:
                 return None
         except ValueError:
+            # a bad header says nothing; the read below enforces the cap
             pass
 
     chunks = []
@@ -707,6 +708,7 @@ def _save_atomically(image, local_path: Path) -> None:
         try:
             os.remove(temp)
         except OSError:
+            # nothing to clean up; the original error is raised below
             pass
         raise
 

py/tagging_defaults.py

@@ -4,8 +4,13 @@ Server-free on purpose: ``py.api`` must import without ComfyUI's ``server``
 module, which ``py/config.py`` needs; config re-exports all of this.
 """
 
+import csv
 import math
-from typing import Any
+import os
+import re
+import threading
+import unicodedata
+from typing import Any, Dict, Iterable, List, Set, Tuple
 
 # The tagging engines (``py/autotag.py`` MODELS) and the options they take.
 WD14_MODEL_TYPES = frozenset({"wd14-swinv2", "wd14-vit"})
@@ -88,3 +93,230 @@ def validate_template(value: Any, name: str = "instructions") -> str:
     if len(encoded) > MAX_INSTRUCTION_BYTES:
         raise ValueError(f"{name} {'are' if name == 'instructions' else 'is'} too long")
     return value
+
+
+# Skipped tags (#431): words whose suggested tags are never offered.
+MAX_SKIPPED_TAGS = 200
+MAX_SKIPPED_TAG_CHARS = 50
+_WORD_SEPARATORS = re.compile(r"[ _-]+")
+_FORBIDDEN_IN_ENTRY = re.compile(r"[,\x00-\x1f\x7f-\x9f\u2028\u2029]")
+
+
+def tag_words(text: str) -> Tuple[str, ...]:
+    """The words of a tag or a skipped entry: NFC, case-folded, split on runs
+    of space, ``_`` and ``-``. Every other character belongs to a word."""
+    folded = unicodedata.normalize("NFC", text).casefold()
+    return tuple(word for word in _WORD_SEPARATORS.split(folded) if word)
+
+
+def validate_skipped_tags(value: Any) -> List[str]:
+    """The Skipped tags list, each entry trimmed, or ValueError with the
+    sentence the System page shows (entries are numbered from 1)."""
+    if not isinstance(value, list):
+        raise ValueError("Skipped tags must be a list of words.")
+    if len(value) > MAX_SKIPPED_TAGS:
+        raise ValueError(f"Skipped tags: at most {MAX_SKIPPED_TAGS} entries.")
+    entries: List[str] = []
+    seen: Dict[Tuple[str, ...], int] = {}
+    for number, raw in enumerate(value, 1):
+        if not isinstance(raw, str):
+            raise ValueError(f"Skipped tags: entry {number} must be text.")
+        entry = raw.strip()
+        if not entry:
+            raise ValueError(f"Skipped tags: entry {number} is empty.")
+        if len(entry) > MAX_SKIPPED_TAG_CHARS:
+            raise ValueError(
+                f"Skipped tags: entry {number} must be at most "
+                f"{MAX_SKIPPED_TAG_CHARS} characters."
+            )
+        if _FORBIDDEN_IN_ENTRY.search(entry):
+            raise ValueError(
+                f"Skipped tags: entry {number} must not contain a comma "
+                "or a line break."
+            )
+        words = tag_words(entry)
+        if not words:
+            raise ValueError(f"Skipped tags: entry {number} is empty.")
+        if words in seen:
+            raise ValueError(
+                f"Skipped tags: entry {number} repeats entry {seen[words]}."
+            )
+        seen[words] = number
+        entries.append(entry)
+    return entries
+
+
+class SkipMatcher:
+    """Whether a suggested tag contains a skipped entry's words, consecutive
+    and whole (#431, 17B). Compiled once per request or job; the cost of a
+    check depends on the tag's words, not on the number of entries."""
+
+    def __init__(self, entries: Iterable[str] = ()):
+        self._entries = frozenset(filter(None, (tag_words(e) for e in entries)))
+        self._longest = max((len(words) for words in self._entries), default=0)
+
+    def __bool__(self) -> bool:
+        return bool(self._entries)
+
+    def matches(self, name: str) -> bool:
+        if not self._entries:
+            return False
+        words = tag_words(name)
+        for start in range(len(words)):
+            stop = min(len(words), start + self._longest)
+            for end in range(start + 1, stop + 1):
+                if words[start:end] in self._entries:
+                    return True
+        return False
+
+    def counting(self) -> "CountingSkip":
+        """A per-request view that also remembers which names it hid."""
+        return CountingSkip(self)
+
+
+class CountingSkip:
+    """A :class:`SkipMatcher` that records the distinct names it hid, so the
+    Tag image dialog can say how many skipped tags were not suggested."""
+
+    def __init__(self, matcher: SkipMatcher):
+        self._matcher = matcher
+        self.hidden: Set[str] = set()
+
+    def matches(self, name: str) -> bool:
+        if self._matcher.matches(name):
+            self.hidden.add(name.casefold())
+            return True
+        return False
+
+
+# A WD14 label list's tags the engine can suggest: general (0) and character (4).
+WD14_SUGGESTED_CATEGORIES = frozenset({"0", "4"})
+_label_cache: Dict[str, Tuple[Tuple[int, int], Tuple[str, ...]]] = {}
+_label_cache_lock = threading.Lock()
+
+
+def read_wd14_label_rows(path) -> List[Dict[str, str]]:
+    """Every row of a WD14 ``selected_tags.csv``, read as UTF-8 and closed
+    before returning (the handle never outlives the call, for Windows)."""
+    with open(path, "r", encoding="utf-8", newline="") as handle:
+        return list(csv.DictReader(handle))
+
+
+def wd14_label_names(path) -> Tuple[str, ...]:
+    """The names a WD14 label list can suggest (categories 0 and 4), cached
+    by ``(mtime_ns, size)``. OSError (missing) and ValueError (unreadable)
+    propagate; the caller skips that list."""
+    key = os.fspath(path)
+    stat = os.stat(key)
+    stamp = (stat.st_mtime_ns, stat.st_size)
+    with _label_cache_lock:
+        cached = _label_cache.get(key)
+    if cached is not None and cached[0] == stamp:
+        return cached[1]
+    try:
+        rows = read_wd14_label_rows(key)
+    except csv.Error as exc:  # a damaged list (a NUL byte, an oversized field)
+        raise ValueError(f"unreadable label list: {exc}") from None
+    names = tuple(
+        name
+        for row in rows
+        if row.get("category", "0") in WD14_SUGGESTED_CATEGORIES
+        and (name := (row.get("name") or "").strip())
+    )
+    with _label_cache_lock:
+        _label_cache[key] = (stamp, names)
+    return names
+
+
+# Saved instructions (#431): named language-model instructions with a default
+# per engine. Built-in is never stored; readers put it first.
+BUILTIN_INSTRUCTIONS_ID = "builtin"
+BUILTIN_INSTRUCTIONS_NAME = "Built-in"
+MAX_SAVED_INSTRUCTIONS = 20
+MAX_INSTRUCTION_NAME_CHARS = 60
+INSTRUCTION_ENGINES = ("gguf", "hf")
+ENGINE_LABELS = {"gguf": "GGUF JoyCaption", "hf": "HuggingFace JoyCaption"}
+_INSTRUCTION_ID = re.compile(r"[0-9a-f]{12}")
+_CONTROL = re.compile(r"[\x00-\x1f\x7f-\x9f

]")
+
+
+def builtin_instructions() -> Dict[str, str]:
+    """The read-only Built-in entry, always the shipped template."""
+    return {
+        "id": BUILTIN_INSTRUCTIONS_ID,
+        "name": BUILTIN_INSTRUCTIONS_NAME,
+        "text": DEFAULT_TAG_TEMPLATE,
+    }
+
+
+def _instruction_entry(raw: Any, number: int) -> Dict[str, str]:
+    prefix = f"Saved instructions: entry {number}"
+    if (
+        not isinstance(raw, dict)
+        or set(raw) != {"id", "name", "text"}
+        or not isinstance(raw["id"], str)
+        or _INSTRUCTION_ID.fullmatch(raw["id"]) is None
+    ):
+        raise ValueError(f"{prefix} has an invalid id.")
+    name = raw["name"].strip() if isinstance(raw["name"], str) else ""
+    if not name:
+        raise ValueError(f"{prefix} needs a name.")
+    if len(name) > MAX_INSTRUCTION_NAME_CHARS:
+        raise ValueError(
+            f"{prefix} name must be at most {MAX_INSTRUCTION_NAME_CHARS} characters."
+        )
+    if _CONTROL.search(name):
+        raise ValueError(f"{prefix} name must not contain a line break.")
+    if name.casefold() == BUILTIN_INSTRUCTIONS_NAME.casefold():
+        raise ValueError("Saved instructions: the name “Built-in” is reserved.")
+    try:
+        text = validate_template(raw["text"], f"{prefix} text")
+    except ValueError as error:
+        raise ValueError(f"{error}.") from None
+    return {"id": raw["id"], "name": name, "text": text}
+
+
+def validate_instructions(value: Any) -> List[Dict[str, str]]:
+    """The saved entries ``[{id, name, text}]`` in order (names trimmed), or
+    ValueError with the sentence System shows (Built-in is not counted)."""
+    if not isinstance(value, list):
+        raise ValueError("Saved instructions must be a list.")
+    if len(value) > MAX_SAVED_INSTRUCTIONS:
+        raise ValueError(
+            f"Saved instructions: at most {MAX_SAVED_INSTRUCTIONS} entries besides "
+            "Built-in."
+        )
+    entries: List[Dict[str, str]] = []
+    ids: Dict[str, int] = {}
+    names: Dict[str, int] = {}
+    for number, raw in enumerate(value, 1):
+        item = _instruction_entry(raw, number)
+        if item["id"] in ids:
+            raise ValueError(
+                f"Saved instructions: entry {number} repeats the id of entry "
+                f"{ids[item['id']]}."
+            )
+        key = item["name"].casefold()
+        if key in names:
+            raise ValueError(
+                f"Saved instructions: entry {number} name “{item['name']}” is "
+                "already used."
+            )
+        ids[item["id"]] = number
+        names[key] = number
+        entries.append(item)
+    return entries
+
+
+def validate_default_instructions(value: Any, entries) -> Dict[str, str]:
+    """``{gguf: id, hf: id}``, each Built-in or an id of ``entries``."""
+    if not isinstance(value, dict) or set(value) != set(INSTRUCTION_ENGINES):
+        raise ValueError("Default instructions must name GGUF and HuggingFace.")
+    ids = {BUILTIN_INSTRUCTIONS_ID} | {item["id"] for item in entries}
+    for engine in INSTRUCTION_ENGINES:
+        if not isinstance(value[engine], str) or value[engine] not in ids:
+            raise ValueError(
+                f"Default for {ENGINE_LABELS[engine]} must be one of the saved "
+                "instructions."
+            )
+    return {engine: value[engine] for engine in INSTRUCTION_ENGINES}

pyproject.toml

@@ -1,7 +1,7 @@
 [project]
 name = "promptmanager"
 description = "A powerful ComfyUI custom node that extends the standard text encoder with persistent prompt storage, advanced search capabilities, and an automatic image gallery system using SQLite."
-version = "3.3.1"
+version = "3.4.1"
 license = {file = "LICENSE"}
 requires-python = ">=3.12"
 dependencies = ["watchdog>=2.1.0", "Pillow>=8.0.0"]

requirements.txt

@@ -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              # Output directory monitoring
+watchdog>=2.1.0               # Output directory monitoring
 Pillow>=8.0.0                # Image metadata extraction and thumbnails

tests/browser/activity-concept.spec.mjs

@@ -6,7 +6,6 @@ import { THEMES, seedConcept, openConcept, openProduction, diffInventory, mutate
 // app.js 1428–1432) and its backup box with the concept's System `.notice`; the differences are D55. The job view
 // has no concept counterpart: it is compared with the approved mockup `activity-scan.html` (a local, ignored file),
 // whose structure and CSS values are restated below (D56). Each comparison is shown to notice a mutation.
-const workspace = '/prompt_manager/workspace';
 
 test.beforeEach(async ({ page, request, browserName }) => {
     await startCoverage(page, browserName);

tests/browser/browser-logs.spec.mjs

@@ -0,0 +1,26 @@
+// #513: a worker records its browser's own log, so a browser that dies between two tests leaves its exit and stderr
+// behind instead of only "Target page, context or browser has been closed" on the next test.
+import { test, expect } from './fixtures.mjs';
+import { isTargetGone } from '../../tools/vendor/browser-logs.mjs';
+
+test('the worker records the launch of the browser its tests use', async ({ browser, browserLogs }) => {
+    expect(browser.isConnected()).toBe(true);
+    expect(browserLogs.launches().length).toBeGreaterThan(0);
+});
+
+test('a killed browser leaves its signal and stderr in the worker log', async ({ playwright, browserName, browserLogs }) => {
+    const before = browserLogs.launches();
+    const victim = await playwright[browserName].launch();
+    const pid = browserLogs.launches().find(item => !before.includes(item));
+    expect(pid, 'the new browser\'s launch was recorded').toBeGreaterThan(0);
+    const disconnected = new Promise(resolve => victim.once('disconnected', resolve));
+    // Playwright starts a browser in a process group of its own; WebKit's launched pid is a wrapper script.
+    process.kill(-pid, 'SIGKILL');
+    await disconnected;
+    const error = await victim.newContext().then(() => null, item => item);
+    expect(error, 'a dead browser refuses a new context').not.toBeNull();
+    expect(isTargetGone(error.message)).toBe(true);
+    // The pipe can close before the launcher has logged the exit.
+    await expect.poll(() => browserLogs.exits()).toContainEqual({ pid, exitCode: null, signal: 'SIGKILL' });
+    expect(browserLogs.tail()).toContain(`[pid=${pid}] <process did exit: exitCode=null, signal=SIGKILL>`);
+});

tests/browser/classic-node.spec.mjs

@@ -7,7 +7,7 @@ test.beforeEach(async ({page, browserName}) => {
     page.classicErrors = [];
     page.on('pageerror', error => page.classicErrors.push(error.message));
     await page.route('**/scripts/app.js', route => route.fulfill({contentType:'text/javascript', body:'export const app = window.fixtureApp;'}));
-    await page.route('**/scripts/api.js', route => route.fulfill({contentType:'text/javascript', body:'export const api = {fileURL: path => path, queuePrompt: async (...args) => args};'}));
+    await page.route('**/scripts/api.js', route => route.fulfill({contentType:'text/javascript', body:'export const api = {fileURL: path => (window.fixtureApiBase ?? "") + path, queuePrompt: async (...args) => args};'}));
     await page.route('**/extensions/promptmanager/prompt_manager.js', route => route.fulfill({contentType:'text/javascript', body:readFileSync(new URL('../../web/comfy/prompt_manager.js', import.meta.url),'utf8')}));
     await page.goto('/prompt_manager/test');
     await page.evaluate(() => {
@@ -64,6 +64,28 @@ test('Classic keeps immediate selection, search query and metadata serialization
     await expect(page.getByRole('button',{name:'🌐 Web UI',exact:true})).toBeVisible();
 });
 
+test('Classic Web UI opens the workspace Library in the saved display mode, through the mount prefix (#525)',async({page})=>{
+    await page.evaluate(()=>{window.fixtureApiBase='/mounted';window.open=(...args)=>{window.openArgs=args;return {};};});
+    const webUi=page.getByRole('button',{name:'🌐 Web UI',exact:true});
+    for (const [mode,args] of [
+        ['newtab',['/mounted/prompt_manager/workspace#/library','_blank']],
+        ['popup',['/mounted/prompt_manager/workspace#/library','_blank','width=1200,height=800,scrollbars=yes,resizable=yes']],
+    ]) {
+        await page.evaluate(value=>{node.properties.webuiDisplayMode=value;},mode);
+        await webUi.click();
+        expect(await page.evaluate(()=>window.openArgs)).toEqual(args);
+    }
+});
+
+test('Classic Web UI reports a blocked popup and keeps the canvas open (#525)',async({page})=>{
+    const before=page.url();
+    await page.evaluate(()=>{window.open=()=>null;});
+    await page.getByRole('button',{name:'🌐 Web UI',exact:true}).click();
+    await expect(page.getByText('Allow popups to open Web UI, then try again.')).toBeVisible();
+    expect(page.url()).toBe(before);
+    await expect(page.locator('.pm-classic')).toBeVisible();
+});
+
 test('Classic detaches native text synchronization on removal',async({page})=>{
     // The page installs the text sync (serializeValue and its listeners) in a 50 ms timer; wait for it, not for time.
     await expect.poll(()=>page.evaluate(()=>typeof node.widgets.find(item=>item.name==='text').serializeValue)).toBe('function');

tests/browser/concept-inventory.mjs

@@ -111,9 +111,23 @@ async function selectTheme(page,theme) {
 }
 
 /** The concept at `#<route>` with its own Theme select. */
+/**
+ * The recorded layout deviations (#530), applied to the opened concept page and never to its files, so every box that
+ * follows from them (wider columns, wrapped text, row heights) is compared rather than excused path by path.
+ * Every other measurement still differs if production moves. workspace-layout-tags-system.spec.mjs pins the real values
+ * and that the concept itself still has the old ones.
+ * D182: the second `.two-col` card starts level, and a surface after a row is 18px below it. D184: no 1000px limit on System. D183: the sticky tag header's background and drawn line.
+ */
+export const CONCEPT_LAYOUT_DEVIATIONS=`.two-col > .surface + .surface { margin-top: 0; }
+.two-col + .surface { margin-top: 18px; }
+.settings-layout { max-width: none; }
+.two-col > .surface > .table-wrap thead th { background: var(--ca-surface); border-bottom-color: transparent; }`;
+export const mapConceptLayout=page=>page.addStyleTag({content:CONCEPT_LAYOUT_DEVIATIONS});
+
 export async function openConcept(page,route,theme,motion='no-preference') {
     await page.emulateMedia({reducedMotion:motion});
     await page.goto('about:blank');await page.goto(`${CONCEPT_URL}#${route}`);
+    await mapConceptLayout(page);
     await selectTheme(page,theme);
     await expect(page.locator('#main #page-title')).toBeVisible();
 }
@@ -421,7 +435,8 @@ export function measureInventory(scope) {
             description:when(description,()=>({className:description.className,text:description.textContent,...font(description),...edges(description)})),
             actions:when(actions,()=>({...layout(actions),...edges(actions),
                 fullWidth:px(rect(actions).width)===px(rect(heading).width),stacked:copy?rect(actions).top>=rect(copy).bottom-0.5:null,
-                buttons:[...actions.querySelectorAll('button')].map(node=>({text:node.textContent.trim(),primary:node.classList.contains('ca-primary'),
+                // D163: a closed menu's items are not heading actions.
+                buttons:[...actions.querySelectorAll('button')].filter(node=>!node.closest('[role="menu"]')).map(node=>({text:node.textContent.trim(),primary:node.classList.contains('ca-primary'),
                     ...font(node),...edges(node),minHeight:getComputedStyle(node).minHeight,height:px(rect(node).height)}))}))};
     }
     // ---- Images (#249 C7). Every box is relative to its section, every section to `main` (vertical rhythm). ----