chore: release 3.4.0

# Conflicts:
#	README.md

.github/workflows/portability.yml

@@ -21,15 +21,22 @@ 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, macos 4m39s-6m14s)
+#   native-trash (#499) has no green main runs yet: its probe, run 40949370943 (2026-10-29), took
+#     about 1 minute per macOS leg, so ceil(1.5 x 2) + 3 gives it 6 minutes (step 4). Recalibrate
+#     after three green runs.
+#   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:
@@ -51,5 +58,88 @@ jobs:
           python -m pip install --upgrade pip
           pip install -r requirements-dev.txt
 
+      # 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
+      # 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
+      # and paths still have to be resolved. D: makes no short names by default
+      # (run 40922502209), so they are switched on for it first.
+      - name: Keep test files on the work disk (Windows)
+        if: runner.os == 'Windows'
+        shell: pwsh
+        run: |
+          fsutil 8dot3name set D: 0
+          $folder = New-Item -ItemType Directory -Force -Path (Join-Path $env:RUNNER_TEMP "PromptManagerTests")
+          $short = (New-Object -ComObject Scripting.FileSystemObject).GetFolder($folder.FullName).ShortPath
+          Write-Output "test TEMP: $short (long name $($folder.FullName))"
+          if ($short -eq $folder.FullName) { Write-Output "::warning::D: gave the test TEMP no 8.3 short name" }
+          "TEMP=$short" >> $env:GITHUB_ENV
+          "TMP=$short" >> $env:GITHUB_ENV
+
       - name: Run tests
+        env:
+          PM_TEST_SLOWEST: "40"
         run: python -m tests.run_suite -v
+
+  # The real Trash on both macOS CPUs (#499 design §10.2). It runs apart from the suite, so a red
+  # suite never skips the native proof.
+  native-trash:
+    name: native trash on ${{ matrix.os }}
+    runs-on: ${{ matrix.os }}
+    timeout-minutes: 6
+    strategy:
+      fail-fast: false
+      matrix:
+        os: [macos-latest, macos-15-intel]
+
+    steps:
+      - uses: actions/checkout@v7
+
+      - name: Set up Python
+        uses: actions/setup-python@v7
+        with:
+          python-version: "3.12"
+
+      - name: Native trash proof
+        timeout-minutes: 4
+        env:
+          PM_NATIVE_TRASH: macos
+        run: python -m unittest tests.test_trash_macos_native tests.test_trash_macos -v
+
+  # The same tests as the fake mount table, on a real ext4 mount used as the output root (#543). A
+  # runner can mount with sudo; the tests run as the normal user, so owner and mode checks are real.
+  native-trash-linux:
+    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 and macOS moved to portability.yml (#420), 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,

.github/workflows/viewerjs-check.yml

@@ -36,6 +36,11 @@ concurrency:
 # 12m53s, node units and the viewer suite before 11.8m of workflows), so coverage is the tight one
 # in its 18 minutes. This holds only when the npm cache hits: a cold cache adds up to 5m to npm ci,
 # and then the job limit can still end the job before the step limit does.
+# #473: firefox/webkit shard 1/2 reached its 18-minute step limit (queue run 40866252975: webkit 622 tests
+# passed in 17m56s, 48.5 worker-minutes vs 28.0 in shard 2/2; recent shard 1/2 jobs 13.7m-18.9m, shard 2/2
+# 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.
 jobs:
   frontend-coverage:
     name: frontend coverage (shard ${{ matrix.shard }}/3)
@@ -175,7 +180,7 @@ jobs:
           retention-days: 14
 
   browser-workflows:
-    name: page workflows (${{ matrix.browser }}, shard ${{ matrix.shard }}/2)
+    name: page workflows (${{ matrix.browser }}, shard ${{ matrix.shard }}/3)
     # Merge queue, main and manual runs only (#420); pull requests run the Chromium/touch coverage shards.
     if: github.event_name != 'pull_request'
     runs-on: ubuntu-latest
@@ -194,7 +199,7 @@ jobs:
       fail-fast: false
       matrix:
         browser: [firefox, webkit]
-        shard: [1, 2]
+        shard: [1, 2, 3]
     steps:
       - uses: actions/checkout@v7
       # The image's own Python 3.12 and Node 24: setup-python/setup-node install the runner's toolcache
@@ -221,9 +226,9 @@ jobs:
           path: ~/.npm
           key: npm-${{ runner.os }}-playwright-v1.64.1-${{ hashFiles('tools/vendor/package-lock.json') }}
       - run: npm ci --prefix tools/vendor --ignore-scripts
-      - name: Page workflows in ${{ matrix.browser }}, shard ${{ matrix.shard }}/2
+      - name: Page workflows in ${{ matrix.browser }}, shard ${{ matrix.shard }}/3
         timeout-minutes: 18
-        run: npm run --prefix tools/vendor test:workflows -- --project ${{ matrix.browser }} --shard ${{ matrix.shard }}/2
+        run: npm run --prefix tools/vendor test:workflows -- --project ${{ matrix.browser }} --shard ${{ matrix.shard }}/3
       # Traces and the blob report (test-results/blob/) of a failed, timed-out or cancelled step.
       - name: Upload browser traces
         if: always()

database/annotations.py

@@ -40,6 +40,7 @@ from .annotation_schema import (
     METADATA_TABLE,
 )
 from .autotag_schema import AUTOTAG_TABLE
+from .image_removal_schema import clear_marks
 from .library_commands import _UUID, LibraryCommandError, _validate_epoch
 from .models import utc_now_iso
 from .vocabulary import validate_vocabulary_tag_name
@@ -236,6 +237,9 @@ class AnnotationOperations:
 
     @staticmethod
     def _annotation_owner(conn: sqlite3.Connection, locator: str, now: str) -> str:
+        # A file with image tags or Needs review state is in the library: a
+        # removal mark never hides it (#480).
+        clear_marks(conn, [locator])
         row = conn.execute(_OWNER, (locator,)).fetchone()
         if row is not None:
             return row[0]

database/execution_contexts.py

@@ -7,6 +7,11 @@ from typing import Callable, ContextManager, Iterator
 
 try:
     from ..utils.execution_history_events import RunFinish, encode_command
+    from ..utils.hashing import (
+        generate_prompt_hash,
+        stored_prompt_text,
+        stored_prompt_text_sql,
+    )
     from ..utils.execution_contexts import (
         CAPTURE_REASONS,
         ContextBundle,
@@ -15,6 +20,11 @@ try:
     )
 except ImportError:
     from utils.execution_history_events import RunFinish, encode_command
+    from utils.hashing import (
+        generate_prompt_hash,
+        stored_prompt_text,
+        stored_prompt_text_sql,
+    )
     from utils.execution_contexts import (
         CAPTURE_REASONS,
         ContextBundle,
@@ -134,7 +144,8 @@ def _insert_exact(
 
 def _role_hints(role: str) -> str:
     return (
-        f"SELECT r.{role}_prompt_id, tx.text_value "
+        f"SELECT r.{role}_prompt_id, tx.text_value, "
+        f"{stored_prompt_text_sql('tx.text_value')} "
         "FROM execution_run_contexts r "
         "JOIN execution_contexts c ON c.context_id = r.context_id "
         f"JOIN execution_prompt_snapshots s ON s.snapshot_id = c.{role}_snapshot_id "
@@ -146,57 +157,78 @@ def _role_hints(role: str) -> str:
 
 # Set form of the read-time prompt-hint rule in get_execution_contexts():
 # a stored role hint is live only in the current storage epoch, and only when
-# its exact raw snapshot text matches exactly one prompt, which is the hinted
-# one (_prompt_hint). Evaluated once per statement, never per prompt row.
+# the hinted prompt is the one match of _prompt_hint: by text (the raw
+# snapshot text or its stored form, outer whitespace stripped, #438), or, only
+# when no prompt has either text, by dedup hash (case folded too, #460).
+# The hash is a fallback, so a hint #438 kept is never made ambiguous.
+# Needs register_prompt_functions (pm_prompt_hash). The hint rows are the
+# outer loop (CROSS JOIN), so each prompts lookup is by index.
 # Defines the CTE ``live_hint_prompts(prompt_id)``; prefix it with WITH.
 LIVE_HINT_CTES = (
-    "stored_hints(prompt_id, raw) AS ("
+    "stored_hints(prompt_id, raw, stored) AS ("
     + _role_hints("positive")
     + " UNION "
     + _role_hints("negative")
-    + "), unique_hint_texts(prompt_id, raw) AS ("
-    "SELECT MIN(id), text COLLATE BINARY FROM prompts "
-    "WHERE text COLLATE BINARY IN (SELECT raw FROM stored_hints) "
-    "GROUP BY text COLLATE BINARY HAVING COUNT(*) = 1"
+    + "), hint_matches(prompt_id, raw, stored, match_id) AS ("
+    "SELECT h.prompt_id, h.raw, h.stored, p.id FROM stored_hints h "
+    "CROSS JOIN prompts p ON p.text COLLATE BINARY IN (h.raw, h.stored) "
+    "UNION ALL "
+    "SELECT h.prompt_id, h.raw, h.stored, p.id FROM stored_hints h "
+    "CROSS JOIN prompts p ON p.hash = pm_prompt_hash(h.raw) "
+    "WHERE NOT EXISTS (SELECT 1 FROM prompts q "
+    "WHERE q.text COLLATE BINARY IN (h.raw, h.stored))"
     "), live_hint_prompts(prompt_id) AS ("
-    "SELECT h.prompt_id FROM stored_hints h JOIN unique_hint_texts u "
-    "ON u.prompt_id = h.prompt_id AND u.raw = h.raw COLLATE BINARY)"
+    "SELECT prompt_id FROM hint_matches GROUP BY prompt_id, raw, stored "
+    "HAVING COUNT(*) = 1 AND MIN(match_id) = prompt_id)"
 )
 LIVE_HINT_PROMPT_IDS_SQL = (
     "WITH " + LIVE_HINT_CTES + " SELECT DISTINCT prompt_id FROM live_hint_prompts"
 )
 
+_BY_TEXT = "SELECT id FROM prompts WHERE text COLLATE BINARY IN (?,?) LIMIT 2"
+
+
+def _text_matches(conn: sqlite3.Connection, raw: str) -> list:
+    return [row[0] for row in conn.execute(_BY_TEXT, (raw, stored_prompt_text(raw)))]
+
 
 def _prompt_hint(conn: sqlite3.Connection, raw: str | None) -> int | None:
+    """The one prompt that stores ``raw`` (#438), else the one it counts toward.
+
+    By text first: ``raw`` or ``raw.strip()`` (the exact form keeps hints of
+    older rows that kept outer whitespace). Only when no prompt has either
+    text, by the dedup hash, which also folds case (#460).
+    """
     if raw is None:
         return None
-    rows = conn.execute(
-        "SELECT id FROM prompts WHERE text=? COLLATE BINARY LIMIT 2", (raw,)
-    ).fetchall()
-    return rows[0][0] if len(rows) == 1 else None
+    matches = _text_matches(conn, raw)
+    if not matches:
+        matches = [
+            row[0]
+            for row in conn.execute(
+                "SELECT id FROM prompts WHERE hash=? LIMIT 2",
+                (generate_prompt_hash(raw),),
+            )
+        ]
+    return matches[0] if len(matches) == 1 else None
 
 
 def _hints(
     conn: sqlite3.Connection, positive: str | None, negative: str | None
 ) -> tuple:
     pos_id, neg_id = _prompt_hint(conn, positive), _prompt_hint(conn, negative)
+    stored = None if negative is None else stored_prompt_text(negative)
     pair_id = None
+    # A known negative that names no prompt because several share its text
     negative_ambiguous = (
-        negative is not None
-        and neg_id is None
-        and conn.execute(
-            "SELECT 1 FROM prompts WHERE text=? COLLATE BINARY LIMIT 1", (negative,)
-        ).fetchone()
-        is not None
+        stored is not None and neg_id is None and bool(_text_matches(conn, negative))
     )
-    if pos_id is not None and negative is not None and not negative_ambiguous:
+    if pos_id is not None and stored is not None and not negative_ambiguous:
         rows = conn.execute(
-            "SELECT pp.id FROM prompt_pairs pp LEFT JOIN prompts n ON "
-            "n.id=pp.negative_prompt_id "
+            "SELECT pp.id FROM prompt_pairs pp "
             "WHERE pp.positive_prompt_id=? AND ((pp.negative_prompt_id IS NULL AND "
-            "?='') "
-            "OR (pp.negative_prompt_id=? AND n.text=? COLLATE BINARY)) LIMIT 2",
-            (pos_id, negative, neg_id, negative),
+            "?='') OR pp.negative_prompt_id=?) LIMIT 2",
+            (pos_id, stored, neg_id),
         ).fetchall()
         if len(rows) == 1:
             pair_id = rows[0][0]

database/health.py

@@ -23,10 +23,10 @@ from pathlib import PurePosixPath, PureWindowsPath
 from urllib.parse import quote
 
 try:
-    from ..utils.hashing import generate_prompt_hash
+    from ..utils.hashing import register_prompt_functions
     from .models import normalize_image_path
 except ImportError:
-    from utils.hashing import generate_prompt_hash
+    from utils.hashing import register_prompt_functions
     from database.models import normalize_image_path
 
 logger = logging.getLogger("prompt_manager.database.health")
@@ -125,12 +125,8 @@ def _foreign_keys(conn, _roots):
     )
 
 
-def _prompt_hash(text):
-    return generate_prompt_hash(text) if isinstance(text, str) else None
-
-
 def _canonical_hashes(conn, _roots):
-    conn.create_function("pm_prompt_hash", 1, _prompt_hash, deterministic=True)
+    register_prompt_functions(conn)  # pm_prompt_hash, shared with history (#460)
     return _counted(
         f"Prompt {row[0]}: the stored hash does not match its text"
         for row in conn.execute(

database/history_reads.py

@@ -15,6 +15,11 @@ from .execution_history import HistoryEpochChanged
 from .execution_outputs import _capture, _REFERENCE_COLUMNS
 from .library import _ELIGIBLE
 
+try:
+    from ..utils.hashing import prompt_key, stored_prompt_text
+except ImportError:
+    from utils.hashing import prompt_key, stored_prompt_text
+
 RunKey = tuple[int, str, int, str]
 
 
@@ -100,14 +105,43 @@ _ORDER = (
 )
 # Stale-hint rows a newest-pair lookup inspects before it reports none.
 _PAIR_RUN_CANDIDATES = 16
+# The recorded texts of one scope, as a JSON array parameter (_scope_text_ids).
+_IN_SCOPE = "IN (SELECT value FROM json_each(?))"
 _MATCHING = (
     "SELECT DISTINCT rc.run_id FROM execution_prompt_snapshots s "
     "JOIN execution_contexts c ON c.positive_snapshot_id=s.snapshot_id "
     "JOIN execution_run_contexts rc ON rc.context_id=c.context_id "
-    "WHERE s.raw_text_id=?"
+    f"WHERE s.raw_text_id {_IN_SCOPE}"
 )
 
 
+def _scope_text_ids(conn: sqlite3.Connection, key: str) -> str:
+    """Every recorded raw text whose prompt key is ``key``, as a JSON array.
+
+    A run keeps its exact raw text, while the Library saves and counts it as
+    the prompt with its key (outer whitespace #438, case #460). The values are
+    compared, never only their digests; ``pm_prompt_key`` is Python's key.
+    """
+    return json.dumps(
+        [
+            row[0]
+            for row in conn.execute(
+                # Only snapshot raw texts: effective and affix texts are skipped
+                "SELECT t.text_id FROM (SELECT DISTINCT raw_text_id AS text_id "
+                "FROM execution_prompt_snapshots) s CROSS JOIN "
+                "execution_context_texts t WHERE t.text_id=s.text_id AND "
+                "pm_prompt_key(t.text_value)=? COLLATE BINARY",
+                (key,),
+            )
+        ]
+    )
+
+
+def _scope_text_id(raw: str | None) -> str | None:
+    """The scope digest of a raw text: that of its prompt key."""
+    return None if raw is None else content_digest("text", (prompt_key(raw),))
+
+
 def _limit(limit: int) -> None:
     if type(limit) is not int or not 1 <= limit <= 100:
         raise ValueError("History limit must be an integer from 1 to 100")
@@ -255,8 +289,15 @@ class HistoryReadOperations:
         if saved is None:
             return None
         text = saved["text"]
-        raw_id = None if text is None else content_digest("text", (text,))
-        if after is not None and after.raw_text_id != raw_id:
+        raw_id = _scope_text_id(text)
+        # A position from before #460 names the scope by the stripped text
+        # alone; it is the same scope, so a page open across the upgrade resumes.
+        legacy_id = (
+            None
+            if text is None
+            else content_digest("text", (stored_prompt_text(text),))
+        )
+        if after is not None and after.raw_text_id not in (raw_id, legacy_id):
             raise ValueError("History position text scope changed")
         group = {
             "positive_id": positive_id,
@@ -273,18 +314,11 @@ class HistoryReadOperations:
             next_position=None,
             resume_position=None,
         )
-        if (
-            raw_id is None
-            or conn.execute(
-                "SELECT 1 FROM execution_context_texts WHERE text_id=? "
-                "AND text_value=? COLLATE BINARY",
-                (raw_id, text),
-            ).fetchone()
-            is None
-        ):
+        if raw_id is None:
             return result
+        scope = _scope_text_ids(conn, prompt_key(text))
         where = f"r.run_id IN ({_MATCHING})"
-        params = [raw_id]
+        params = [scope]
         if after is not None:
             where += f" AND {_KEY}<=(?,?,?,?)"
             params.extend(after.upper)
@@ -338,9 +372,9 @@ class HistoryReadOperations:
             "LEFT JOIN execution_prompt_snapshots n "
             "ON n.snapshot_id=c.negative_snapshot_id "
             "LEFT JOIN execution_context_texts t ON t.text_id=n.raw_text_id "
-            f"WHERE rc.run_id IN ({placeholders}) AND p.raw_text_id=? "
+            f"WHERE rc.run_id IN ({placeholders}) AND p.raw_text_id {_IN_SCOPE} "
             "ORDER BY rc.anchor_id COLLATE BINARY,rc.run_id COLLATE BINARY",
-            (*by_run, raw_id),
+            (*by_run, scope),
         )
         summaries = {}
         evidence_by_context = {}
@@ -438,6 +472,7 @@ class HistoryReadOperations:
         if saved["text"] is None:
             return {"run": None}
         epoch = self.get_execution_epoch()
+        key = prompt_key(saved["text"])
         candidates = conn.execute(
             "SELECT r.run_id,rc.anchor_id,rc.context_id,"
             "c.positive_snapshot_id,c.negative_snapshot_id "
@@ -445,10 +480,11 @@ class HistoryReadOperations:
             "ON rc.run_id=r.run_id JOIN execution_contexts c "
             "ON c.context_id=rc.context_id JOIN execution_prompt_snapshots s "
             "ON s.snapshot_id=c.positive_snapshot_id "
-            "WHERE s.raw_text_id=? AND rc.pair_id=? AND rc.reference_epoch=? "
+            f"WHERE s.raw_text_id {_IN_SCOPE} AND rc.pair_id=? "
+            "AND rc.reference_epoch=? "
             f"ORDER BY {_ORDER},rc.anchor_id COLLATE BINARY LIMIT ?",
             (
-                content_digest("text", (saved["text"],)),
+                _scope_text_ids(conn, key),
                 pair_id,
                 epoch,
                 _PAIR_RUN_CANDIDATES,
@@ -457,7 +493,11 @@ class HistoryReadOperations:
         for row in candidates:
             positive = _values(conn, row["positive_snapshot_id"])
             negative = _values(conn, row["negative_snapshot_id"])
-            if positive is None or positive["raw"] != saved["text"]:
+            if (
+                positive is None
+                or positive["raw"] is None
+                or prompt_key(positive["raw"]) != key
+            ):
                 continue
             hints = _hints(conn, positive["raw"], negative["raw"] if negative else None)
             if hints[2] == pair_id:
@@ -477,11 +517,9 @@ class HistoryReadOperations:
         snapshots = {
             row["anchor_id"]: row
             for row in conn.execute(
-                "SELECT rc.anchor_id,c.positive_snapshot_id,c.negative_snapshot_id,"
-                "p.raw_text_id AS positive_raw_text_id "
+                "SELECT rc.anchor_id,c.positive_snapshot_id,c.negative_snapshot_id "
                 "FROM execution_run_contexts rc JOIN execution_contexts c "
-                "ON c.context_id=rc.context_id LEFT JOIN execution_prompt_snapshots p "
-                "ON p.snapshot_id=c.positive_snapshot_id WHERE rc.run_id=?",
+                "ON c.context_id=rc.context_id WHERE rc.run_id=?",
                 (run_id,),
             )
         }
@@ -491,12 +529,12 @@ class HistoryReadOperations:
             context.update(
                 {
                     key: row[key]
-                    for key in (
-                        "positive_snapshot_id",
-                        "negative_snapshot_id",
-                        "positive_raw_text_id",
-                    )
-                }
+                    for key in ("positive_snapshot_id", "negative_snapshot_id")
+                },
+                # The history scope this context belongs to (#438)
+                positive_raw_text_id=_scope_text_id(
+                    (context["positive"] or {}).get("raw")
+                ),
             )
             pair = context["pair_id"]
             context["navigation"] = {

database/image_categories.py

@@ -221,6 +221,7 @@ class ImageCategoryOperations:
         sources. Returns ``{updated, unchanged}``; a file that already had
         exactly this value is ``unchanged`` and keeps its ``updated_at``.
         """
+        from .image_removal_schema import clear_marks_of_keys
         from .models import utc_now_iso  # models imports this module's schema
 
         keys = list(dict.fromkeys(tuple(key) for key in file_keys))
@@ -234,6 +235,7 @@ class ImageCategoryOperations:
                     continue
                 conn.execute(UPSERT_SQL, (*key, category, now))
                 counts["updated"] += 1
+            clear_marks_of_keys(conn, keys)  # #480: a categorized file is listed
         return counts
 
     def clear_image_categories(self, file_keys) -> int:

database/image_collections.py

@@ -402,6 +402,7 @@ class ImageCollectionOperations:
         is created only when at least one item is inserted. Returns
         ``{collection, created, added, already}``.
         """
+        from .image_removal_schema import clear_marks_of_keys
         from .models import utc_now_iso  # models imports this module's schema
 
         if (collection_id is None) == (name is None):
@@ -438,6 +439,7 @@ class ImageCollectionOperations:
                 ).rowcount
                 for root, path in pairs
             )
+            clear_marks_of_keys(conn, pairs)  # #480: a collected file is listed
         return {
             "collection": collection,
             "created": created,

database/image_favorites.py

@@ -104,11 +104,14 @@ class ImageFavoriteOperations:
         """Mark one file as a favorite; repeating it is a no-op. Returns True."""
         from .models import utc_now_iso  # models imports this module's schema
 
+        from .image_removal_schema import clear_marks_of_keys
+
         with self.model.get_connection() as conn:
             conn.execute(
                 "INSERT OR IGNORE INTO image_favorites VALUES (?, ?, ?)",
                 (root_key, relative_path, utc_now_iso()),
             )
+            clear_marks_of_keys(conn, [(root_key, relative_path)])  # #480
         return True
 
     def clear_image_favorite(self, root_key: str, relative_path: str) -> bool:

database/image_removal.py

@@ -0,0 +1,348 @@
+"""Remove a file's records from the library, and its removal mark (#480).
+
+A *removal* describes one physical file the way the API resolved it from the
+walked listing (``py/api/image_delete.py``):
+
+- ``spellings``: every path string records may name it by
+  (``duplicate_spellings.Spellings.of``);
+- ``locator``: its private canonical locator;
+- ``favorites``: ``(root key, relative key)`` under every listed root holding
+  it, the keys of its favorite and its collection memberships;
+- ``category``: its own category's file key, or None.
+
+``remove_image_records`` deletes, in one ``BEGIN IMMEDIATE`` transaction:
+
+- its image records (``generated_images``) under every spelling, plus legacy
+  rows matched by ``image_path``; their saved-pair links go by cascade and the
+  workflow documents they held are released;
+- the image tags and auto-tag marks of its byte records. The annotation file
+  and byte rows go too, unless a Needs review item still cites them
+  (``media_review_items.byte_id`` is ``ON DELETE RESTRICT``): a decided or
+  revised item keeps its decision;
+- its pending, undecided Needs review items (settled as duplicate removal
+  settles them);
+- its own category, favorite and collection memberships.
+
+With ``mark`` it also writes the file's removal mark (library-only removal):
+the file stays on disk, hidden until its bytes change. Prompts, recorded runs
+and the media file itself are never touched here.
+"""
+
+import os
+
+from .annotation_schema import BYTES_TABLE, FILES_TABLE
+from .annotations import _BUMP_INDEX
+from .autotag_schema import AUTOTAG_TABLE
+from .duplicate_copies import (
+    DELETE_FAVORITE_SQL,
+    DELETE_LINK_SQL,
+    DELETE_MEMBERSHIPS_SQL,
+    _file_link_rows,
+)
+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,
+    write_mark as _write_mark,
+)
+from .library import display_title
+from .models import utc_now_iso
+from .review import _BUMP_REVIEW_INDEX, _PENDING_OF_LOCATOR, _SETTLE_ITEM
+from .review_changes import notify_review_change
+from .workflow_documents import documents_of_rows, release_documents
+
+BYTES_OF_LOCATOR_SQL = (
+    f"SELECT b.id FROM {FILES_TABLE} f JOIN {BYTES_TABLE} b "  # nosec B608
+    "ON b.file_id=f.id WHERE f.locator=?"
+)
+TAG_COUNT_SQL = "SELECT COUNT(*) FROM media_image_tags WHERE byte_id=?"
+AUTOTAG_COUNT_SQL = (
+    f"SELECT COUNT(*) FROM {AUTOTAG_TABLE} WHERE byte_id=?"  # nosec B608
+)
+CLEAR_TAGS_SQL = "DELETE FROM media_image_tags WHERE byte_id=?"
+CLEAR_AUTOTAG_SQL = f"DELETE FROM {AUTOTAG_TABLE} WHERE byte_id=?"  # nosec B608
+BUMP_TAG_REVISION_SQL = (
+    f"UPDATE {BYTES_TABLE} SET tag_revision=tag_revision+1 WHERE id=?"  # nosec B608
+)
+REVIEW_REFS_SQL = (
+    f"SELECT COUNT(*) FROM {FILES_TABLE} f "  # nosec B608
+    f"JOIN {BYTES_TABLE} b ON b.file_id=f.id "
+    "JOIN media_review_items i ON i.byte_id=b.id WHERE f.locator=?"
+)
+DELETE_FILE_SQL = f"DELETE FROM {FILES_TABLE} WHERE locator=?"  # nosec B608
+FILE_EXISTS_SQL = f"SELECT 1 FROM {FILES_TABLE} WHERE locator=?"  # nosec B608
+FAVORITE_EXISTS_SQL = (
+    "SELECT COUNT(*) FROM image_favorites WHERE root_key=? AND relative_path=?"
+)
+MEMBERSHIP_COUNT_SQL = (
+    "SELECT COUNT(*) FROM image_collection_items WHERE root_key=? AND relative_path=?"
+)
+CATEGORY_EXISTS_SQL = (
+    "SELECT COUNT(*) FROM image_categories WHERE root_key=? AND relative_path=?"
+)
+PAIR_LINK_COUNT_SQL = "SELECT COUNT(*) FROM image_prompt_pairs WHERE image_id=?"
+PROMPT_LINKS_SQL = "SELECT id FROM generated_images WHERE prompt_id=?"
+PROMPT_TITLE_SQL = (
+    "SELECT p.text, s.title FROM prompts p "
+    "LEFT JOIN prompt_library_state s ON s.prompt_id=p.id WHERE p.id=?"
+)
+MARKS_SQL = f"SELECT locator, size, mtime_ns FROM {TABLE}"  # nosec B608
+REVISION_SQL = f"SELECT revision FROM {META_TABLE} WHERE singleton_id=1"  # nosec B608
+MAX_ORPHAN_TITLES = 3
+RECORD_KINDS = (
+    "prompt_links",
+    "pair_links",
+    "image_tags",
+    "auto_tagged",
+    "categories",
+    "collections",
+    "favorites",
+    "review_items",
+)
+
+
+def _link_ids(conn, removal):
+    """``{image id: prompt id}`` of the records naming the file, any spelling."""
+    found = {}
+    for spelled in dict.fromkeys(removal["spellings"]):
+        for image_id, prompt_id in _file_link_rows(conn, spelled):
+            found[image_id] = prompt_id
+    return found
+
+
+def _byte_ids(conn, locator):
+    return [row[0] for row in conn.execute(BYTES_OF_LOCATOR_SQL, (locator,))]
+
+
+def file_locator(path) -> str:
+    """The locator of an absolute path (``normcase(realpath)``)."""
+    return os.path.normcase(os.path.realpath(str(path)))
+
+
+def removed_from_library(conn, path) -> bool:
+    """Whether *path* is hidden by a removal mark whose bytes still match.
+
+    A mark whose file changed (other size or ``mtime_ns``) or is gone is
+    stale: it is deleted on *conn* (the caller's write transaction), and the
+    file is not hidden.
+    """
+    locator = file_locator(path)
+    mark = mark_of(conn, locator)
+    if mark is None:
+        return False
+    try:
+        info = os.lstat(path)
+    except OSError:
+        clear_marks(conn, [locator])
+        return False
+    if (info.st_size, info.st_mtime_ns) == tuple(mark):
+        return True
+    clear_marks(conn, [locator])
+    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())
+
+
+class ImageRemovalOperations:
+    """PromptDatabase mixin: remove a file's records; removal marks."""
+
+    def count_removal_records(self, removals) -> dict:
+        """Counts of what removing *removals* deletes, in one read snapshot.
+
+        ``{records: {kind: n}, orphaned_prompts: {count, titles}}``: prompts
+        whose every image record is among these files lose their last image
+        (they stay in the Library).
+        """
+        from .maintenance import _read_snapshot
+
+        def read(conn):
+            totals = dict.fromkeys(RECORD_KINDS, 0)
+            links = {}
+            for removal in removals:
+                found = _link_ids(conn, removal)
+                links.update(found)
+                for image_id in found:
+                    totals["pair_links"] += conn.execute(
+                        PAIR_LINK_COUNT_SQL, (image_id,)
+                    ).fetchone()[0]
+                tagged = autotagged = False
+                for byte_id in _byte_ids(conn, removal["locator"]):
+                    tagged |= bool(
+                        conn.execute(TAG_COUNT_SQL, (byte_id,)).fetchone()[0]
+                    )
+                    autotagged |= bool(
+                        conn.execute(AUTOTAG_COUNT_SQL, (byte_id,)).fetchone()[0]
+                    )
+                totals["image_tags"] += int(tagged)
+                totals["auto_tagged"] += int(autotagged)
+                totals["review_items"] += len(
+                    conn.execute(_PENDING_OF_LOCATOR, (removal["locator"],)).fetchall()
+                )
+                totals["favorites"] += int(
+                    any(
+                        conn.execute(FAVORITE_EXISTS_SQL, key).fetchone()[0]
+                        for key in removal["favorites"]
+                    )
+                )
+                totals["collections"] += sum(
+                    conn.execute(MEMBERSHIP_COUNT_SQL, key).fetchone()[0]
+                    for key in removal["favorites"]
+                )
+                if removal.get("category"):
+                    totals["categories"] += conn.execute(
+                        CATEGORY_EXISTS_SQL, removal["category"]
+                    ).fetchone()[0]
+            totals["prompt_links"] = len(links)
+            return {
+                "records": totals,
+                "orphaned_prompts": self._orphaned_prompts(conn, links),
+            }
+
+        return _read_snapshot(self.model.db_path, read)
+
+    @staticmethod
+    def _orphaned_prompts(conn, links):
+        removed = set(links)
+        orphans = []
+        for prompt_id in dict.fromkeys(links.values()):
+            remaining = [
+                row[0]
+                for row in conn.execute(PROMPT_LINKS_SQL, (prompt_id,))
+                if row[0] not in removed
+            ]
+            if not remaining:
+                orphans.append(prompt_id)
+        titles = []
+        for prompt_id in orphans[:MAX_ORPHAN_TITLES]:
+            row = conn.execute(PROMPT_TITLE_SQL, (prompt_id,)).fetchone()
+            if row is not None:
+                titles.append(display_title((row[1],), row[0]) or "")
+        return {"count": len(orphans), "titles": titles}
+
+    def remove_image_records(self, removal, *, mark=None) -> dict:
+        """Delete every record of one file (one transaction); see the module.
+
+        ``mark``: ``(size, mtime_ns)`` to write the file's removal mark, or
+        None (the file was moved to the trash). Returns the counts deleted.
+        """
+        counts = dict.fromkeys(RECORD_KINDS, 0)
+        locator = removal["locator"]
+        with self._pair_transaction() as conn:
+            links = _link_ids(conn, removal)
+            if links:
+                documents = documents_of_rows(conn, "id", list(links))
+                for image_id in links:
+                    counts["pair_links"] += conn.execute(
+                        PAIR_LINK_COUNT_SQL, (image_id,)
+                    ).fetchone()[0]
+                    conn.execute(DELETE_LINK_SQL, (image_id,))
+                release_documents(conn, documents)
+            counts["prompt_links"] = len(links)
+            counts.update(self._remove_annotations(conn, locator))
+            for key in removal["favorites"]:
+                counts["favorites"] += conn.execute(DELETE_FAVORITE_SQL, key).rowcount
+                counts["collections"] += conn.execute(
+                    DELETE_MEMBERSHIPS_SQL, key
+                ).rowcount
+            if removal.get("category"):
+                counts["categories"] += conn.execute(
+                    DELETE_CATEGORY_SQL, tuple(removal["category"])
+                ).rowcount
+            if mark is not None:
+                write_mark(conn, locator, *mark)
+        if counts["review_items"]:
+            notify_review_change(self.model.db_path)
+        return counts
+
+    def _remove_annotations(self, conn, locator) -> dict:
+        """Tags, auto-tag marks, pending review items and, when free, the rows."""
+        counts = {"image_tags": 0, "auto_tagged": 0, "review_items": 0}
+        bytes_ = _byte_ids(conn, locator)
+        for byte_id in bytes_:
+            tags = conn.execute(CLEAR_TAGS_SQL, (byte_id,)).rowcount
+            autotags = conn.execute(CLEAR_AUTOTAG_SQL, (byte_id,)).rowcount
+            if tags:
+                conn.execute(BUMP_TAG_REVISION_SQL, (byte_id,))
+            counts["image_tags"] |= bool(tags)
+            counts["auto_tagged"] |= bool(autotags)
+        for (item_id,) in conn.execute(_PENDING_OF_LOCATOR, (locator,)).fetchall():
+            counts["review_items"] += conn.execute(_SETTLE_ITEM, (item_id,)).rowcount
+        if counts["review_items"]:
+            conn.execute(_BUMP_REVIEW_INDEX)
+        if conn.execute(FILE_EXISTS_SQL, (locator,)).fetchone():
+            if not conn.execute(REVIEW_REFS_SQL, (locator,)).fetchone()[0]:
+                conn.execute(DELETE_FILE_SQL, (locator,))
+            conn.execute(_BUMP_INDEX)
+        counts["image_tags"] = int(counts["image_tags"])
+        counts["auto_tagged"] = int(counts["auto_tagged"])
+        return counts
+
+    def removal_marks(self) -> dict:
+        """``{locator: (size, mtime_ns)}`` of every removal mark (one read)."""
+        conn = self.model.get_connection()
+        if not has_marks_table(conn):
+            return {}
+        return {row[0]: (row[1], row[2]) for row in conn.execute(MARKS_SQL)}
+
+    def removal_revision(self) -> int:
+        conn = self.model.get_connection()
+        if not has_marks_table(conn):
+            return 0
+        row = conn.execute(REVISION_SQL).fetchone()
+        return row[0] if row else 0
+
+    def clear_removal_marks(self, locators) -> int:
+        """Show these files again (their marks go); the count removed."""
+        locators = list(locators)
+        if not locators:
+            return 0
+        with self._pair_transaction() as conn:
+            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

@@ -0,0 +1,213 @@
+"""Removal marks (#480): files removed from the library that stay on disk.
+
+``image_removals`` holds one row per hidden file: its private canonical
+locator (the ``media_annotation_files.locator`` form) and the ``size`` and
+``mtime_ns`` it had when it was removed. A file at that locator with other
+bytes is a new file and shows again. ``image_removals_meta`` is one row, the
+marks revision, bumped in the same transaction as every mark change so the
+workspace listing caches can tell marks changed.
+
+The group installs like ``image_favorites``: optional in old backups,
+required once a database is current. Older builds ignore it.
+"""
+
+import os
+import re
+import sqlite3
+
+from .annotation_schema import MAX_LOCATOR_LENGTH
+
+TABLE = "image_removals"
+META_TABLE = "image_removals_meta"
+TABLES = (TABLE, META_TABLE)
+_TABLE_DDL = {
+    TABLE: (
+        f"CREATE TABLE {TABLE} (locator TEXT PRIMARY KEY, "
+        "size INTEGER NOT NULL CHECK(size>=0), mtime_ns INTEGER NOT NULL, "
+        "removed_at TEXT NOT NULL) WITHOUT ROWID"
+    ),
+    META_TABLE: (
+        f"CREATE TABLE {META_TABLE} (singleton_id INTEGER PRIMARY KEY "
+        "CHECK(singleton_id=1), revision INTEGER NOT NULL DEFAULT 0 "
+        "CHECK(revision>=0))"
+    ),
+}
+_SEED_META = (
+    f"INSERT OR IGNORE INTO {META_TABLE} (singleton_id, revision) VALUES (1, 0)"
+)
+_COLUMNS = {
+    TABLE: (
+        ("locator", "TEXT", 1, 1),
+        ("size", "INTEGER", 1, 0),
+        ("mtime_ns", "INTEGER", 1, 0),
+        ("removed_at", "TEXT", 1, 0),
+    ),
+    META_TABLE: (
+        ("singleton_id", "INTEGER", 0, 1),
+        ("revision", "INTEGER", 1, 0),
+    ),
+}
+_AUTOINDEX = f"sqlite_autoindex_{TABLE}_1"
+
+
+def _compact(sql: str) -> str:
+    return re.sub(r"\s+", "", sql or "").lower()
+
+
+def _structure_error(conn: sqlite3.Connection, allow_missing: bool) -> str | None:
+    rows = conn.execute(
+        "SELECT type,name,tbl_name,sql FROM sqlite_master WHERE name IN (?,?) "
+        "OR tbl_name IN (?,?)",
+        TABLES + TABLES,
+    ).fetchall()
+    if not rows:
+        return None if allow_missing else "Missing image removal marks schema"
+    tables = {name: sql for kind, name, _tbl, sql in rows if kind == "table"}
+    others = {(kind, name) for kind, name, _tbl, _sql in rows if kind != "table"}
+    if set(tables) != set(TABLES) or others - {("index", _AUTOINDEX)}:
+        return "Image removal marks name is reserved by an unsupported schema object"
+    for table in TABLES:
+        if _compact(tables[table]) != _compact(_TABLE_DDL[table]):
+            return "Image removal marks have 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[table]:
+            return "Image removal marks have invalid columns"
+    return None
+
+
+def _values_error(conn: sqlite3.Connection) -> str | None:
+    limit = MAX_LOCATOR_LENGTH * 4
+    if conn.execute(
+        f"SELECT 1 FROM {TABLE} WHERE typeof(locator) != 'text' "  # nosec B608
+        f"OR length(CAST(locator AS BLOB)) > {limit} OR locator = '' "
+        "OR typeof(size) != 'integer' OR typeof(mtime_ns) != 'integer' "
+        "OR typeof(removed_at) != 'text' OR length(CAST(removed_at AS BLOB)) > 64 "
+        "LIMIT 1"
+    ).fetchone():
+        return "Image removal marks contain invalid values"
+    meta = conn.execute(
+        f"SELECT singleton_id, typeof(revision) FROM {META_TABLE}"  # nosec B608
+    ).fetchall()
+    if meta != [(1, "integer")]:
+        return "Image removal marks have an invalid revision"
+    return None
+
+
+def image_removals_schema_error(
+    conn: sqlite3.Connection, *, allow_missing: bool = True
+) -> str | None:
+    """Reject unsupported schemas and invalid persisted values (reads only)."""
+    error = _structure_error(conn, allow_missing)
+    if error:
+        return error
+    exists = conn.execute(
+        "SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", (TABLE,)
+    ).fetchone()
+    return _values_error(conn) if exists else None
+
+
+def migrate_image_removals(conn: sqlite3.Connection) -> None:
+    """Install both tables atomically; validate before and after."""
+    conn.execute("SAVEPOINT image_removals_schema")
+    try:
+        error = image_removals_schema_error(conn)
+        if error:
+            raise sqlite3.DatabaseError(error)
+        for sql in _TABLE_DDL.values():
+            conn.execute(sql.replace("CREATE TABLE", "CREATE TABLE IF NOT EXISTS", 1))
+        conn.execute(_SEED_META)
+        error = image_removals_schema_error(conn, allow_missing=False)
+        if error:
+            raise sqlite3.DatabaseError(error)
+        conn.execute("RELEASE image_removals_schema")
+    except BaseException:
+        conn.execute("ROLLBACK TO image_removals_schema")
+        conn.execute("RELEASE image_removals_schema")
+        raise
+
+
+_UNMARK_SQL = f"DELETE FROM {TABLE} WHERE locator=?"  # nosec B608
+_MARK_SQL = (
+    f"INSERT OR REPLACE INTO {TABLE} "  # nosec B608
+    "(locator, size, mtime_ns, removed_at) VALUES (?, ?, ?, ?)"
+)
+_MARK_OF_SQL = f"SELECT size, mtime_ns FROM {TABLE} WHERE locator=?"  # nosec B608
+BUMP_MARKS_SQL = (
+    f"UPDATE {META_TABLE} SET revision=revision+1 WHERE singleton_id=1"  # nosec B608
+)
+
+
+# Bumped whenever this process writes or clears a mark (even in a transaction
+# that later rolls back): readers that cache the marks reload when it moves.
+_generation = 0
+
+
+def generation() -> int:
+    """The marks generation of this process (see ``_generation``)."""
+    return _generation
+
+
+def _bump() -> None:
+    global _generation
+    _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(
+            "SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", (TABLE,)
+        ).fetchone()
+        is not None
+    )
+
+
+def locator_of_key(root_key: str, relative_key: str) -> str:
+    """The locator of a favorite-style file key (root key + relative key)."""
+    return os.path.normcase(os.path.join(root_key, *relative_key.split("/")))
+
+
+def mark_of(conn: sqlite3.Connection, locator: str):
+    """``(size, mtime_ns)`` of the file's removal mark, or None."""
+    if not has_marks_table(conn):
+        return None
+    row = conn.execute(_MARK_OF_SQL, (locator,)).fetchone()
+    return (row[0], row[1]) if row else None
+
+
+def write_mark(
+    conn: sqlite3.Connection, locator: str, size: int, mtime_ns: int, now: str
+):
+    conn.execute(_MARK_SQL, (locator, int(size), int(mtime_ns), now))
+    conn.execute(BUMP_MARKS_SQL)
+    _bump()
+
+
+def clear_marks(conn: sqlite3.Connection, locators) -> int:
+    """Delete the marks of *locators* on *conn* (no commit); the count removed.
+
+    Every record write for a file calls this (#480): a file with records is
+    in the library, so it is never hidden.
+    """
+    locators = list(dict.fromkeys(locators))
+    if not locators or not has_marks_table(conn):
+        return 0
+    removed = 0
+    for locator in locators:
+        removed += conn.execute(_UNMARK_SQL, (locator,)).rowcount
+    if removed:
+        conn.execute(BUMP_MARKS_SQL)
+        _bump()
+    return removed
+
+
+def clear_marks_of_keys(conn: sqlite3.Connection, file_keys) -> int:
+    """``clear_marks`` for favorite-style ``(root key, relative key)`` pairs."""
+    return clear_marks(conn, [locator_of_key(*key) for key in file_keys])

database/library.py

@@ -5,7 +5,9 @@ import sqlite3
 from collections.abc import Iterator
 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)
@@ -105,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("_", "\\_")
 
@@ -121,7 +129,54 @@ def _archive(value: str) -> str:
     return ARCHIVE_SCOPES[value]
 
 
-def _filter(text: str, category: str | None, tags: list[str]) -> tuple[str, list]:
+# How the Tags filter matches (#444): every tag (AND), any of them (OR), or
+# parents without any tag.
+TAG_MODES = ("all", "any", "untagged")
+# A parent that no tag's Browse returns (#444 "Untagged prompts"): no tag of its
+# own and no active pair carrying one. The vocabulary's untagged count uses it too.
+UNTAGGED = (
+    "NOT EXISTS(SELECT 1 FROM prompt_tags ut WHERE ut.prompt_id=p.id) "
+    "AND NOT EXISTS(SELECT 1 FROM prompt_pairs up JOIN prompt_pair_tags upt "
+    "ON upt.pair_id=up.id WHERE up.positive_prompt_id=p.id AND up.archived=0)"
+)
+
+
+def _tag_mode(tag_mode: object, tags: list[str]) -> str:
+    if not isinstance(tag_mode, str) or tag_mode not in TAG_MODES:
+        raise ValueError("tag_mode must be one of " + ", ".join(TAG_MODES))
+    if tag_mode == "untagged" and tags:
+        raise ValueError("Untagged prompts cannot also filter by tags")
+    return tag_mode
+
+
+def _tag_conditions(tags: list[str], any_tag: bool) -> tuple[list, list, list]:
+    """Parent-arm and pair-arm tag conditions and the tags' parameters."""
+    parent_exists = (
+        "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{test})"
+    )
+    pair_exists = (
+        "EXISTS(SELECT 1 FROM prompt_pair_tags pt JOIN tags t ON t.id=pt.tag_id "
+        "WHERE pt.pair_id=pp.id AND t.name{test})"
+    )
+    if any_tag and tags:
+        # One condition per arm: the card, or one active pair, carries any tag.
+        test = " IN (" + ",".join("?" for _ in tags) + ")"
+        return [parent_exists.format(test=test)], [pair_exists.format(test=test)], tags
+    return (
+        [parent_exists.format(test="=?") for _ in tags],
+        [pair_exists.format(test="=?") for _ in tags],
+        tags,
+    )
+
+
+def _filter(
+    text: str,
+    category: str | None,
+    tags: list[str],
+    tag_mode: str = "all",
+    folder_prefixes: list[str] | None = None,
+) -> tuple[str, list]:
     if not isinstance(text, str) or len(text) > 500:
         raise ValueError("Search text must be a string of at most 500 characters")
     if category is not None and (not isinstance(category, str) or len(category) > 200):
@@ -132,6 +187,8 @@ def _filter(text: str, category: str | None, tags: list[str]) -> tuple[str, list
         or any(not isinstance(tag, str) or not tag or len(tag) > 200 for tag in tags)
     ):
         raise ValueError("Invalid library tags")
+    mode = _tag_mode(tag_mode, tags)
+    untagged = mode == "untagged"
     parent, pair, parent_params, pair_params = [], ["pp.archived=0"], [], []
     if text:
         parent.append("(p.text LIKE ? ESCAPE '\\' OR ls.title LIKE ? ESCAPE '\\')")
@@ -146,17 +203,11 @@ def _filter(text: str, category: str | None, tags: list[str]) -> tuple[str, list
         pair.append("pp.category=?")
         parent_params.append(category)
         pair_params.append(category)
-    for tag in tags:
-        parent.append(
-            "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=?)"
-        )
-        pair.append(
-            "EXISTS(SELECT 1 FROM prompt_pair_tags pt JOIN tags t ON t.id=pt.tag_id "
-            "WHERE pt.pair_id=pp.id AND t.name=?)"
-        )
-        parent_params.append(tag)
-        pair_params.append(tag)
+    parent_tags, pair_tags, tag_params = _tag_conditions(tags, mode == "any")
+    parent.extend(parent_tags)
+    pair.extend(pair_tags)
+    parent_params.extend(tag_params)
+    pair_params.extend(tag_params)
     condition = (
         "("
         + (" AND ".join(parent) or "1=1")
@@ -168,7 +219,17 @@ def _filter(text: str, category: str | None, tags: list[str]) -> tuple[str, list
         + " AND ".join(pair)
         + ")"
     )
-    return f"({_ELIGIBLE}) AND ({condition})", [*parent_params, *pair_params]
+    if untagged:
+        # A whole-card condition: inside the parent-or-pair OR an untagged pair
+        # would let a tagged parent through.
+        condition = f"({condition}) AND {UNTAGGED}"
+    params = [*parent_params, *pair_params]
+    if folder_prefixes:
+        # A whole-card condition around both arms: images link to the parent.
+        folder, folder_params = folder_condition(folder_prefixes)
+        condition = f"({condition}) AND {folder}"
+        params.extend(folder_params)
+    return f"({_ELIGIBLE}) AND ({condition})", params
 
 
 class LibraryOperations:
@@ -216,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,"
@@ -239,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
 
@@ -251,14 +315,21 @@ class LibraryOperations:
         text: str = "",
         category: str | None = None,
         tags: list[str] | None = None,
+        tag_mode: str = "all",
+        folder_prefixes: list[str] | None = None,
         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")
-        where, params = _filter(text, category, [] if tags is None else tags)
+        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
+        )
         if after is not None:
             if (
                 not isinstance(after, dict)
@@ -270,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 "<"
@@ -311,6 +385,9 @@ 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,
             "next_key": next_key,
         }
@@ -321,9 +398,12 @@ class LibraryOperations:
         text: str = "",
         category: str | None = None,
         tags: list[str] | None = None,
+        tag_mode: str = "all",
+        folder_prefixes: list[str] | None = None,
         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(
@@ -333,9 +413,12 @@ class LibraryOperations:
                 text=text,
                 category=category,
                 tags=tags,
+                tag_mode=tag_mode,
+                folder_prefixes=folder_prefixes,
                 sort=sort,
                 limit=limit,
                 after=after,
+                shelf=shelf,
             )
 
     def _is_library_positive(self, conn: sqlite3.Connection, positive_id: int) -> bool:
@@ -463,9 +546,13 @@ class LibraryOperations:
             "categories": [row["category"] for row in rows],
         }
 
-    def get_library_options(self) -> dict:
+    def get_library_options(self, folder_roots=None) -> dict:
+        """Categories; with *folder_roots* (root spellings) also ``folders``."""
         with self.library_snapshot() as (conn, epoch, _ceiling):
-            return self._get_library_options(conn, epoch)
+            result = self._get_library_options(conn, epoch)
+            if folder_roots is not None:
+                result["folders"] = list_library_folders(conn, folder_roots, _ELIGIBLE)
+            return result
 
     def get_library_previews(self, positive_ids: list[int]) -> dict[int, list[dict]]:
         """Read at most two links per parent in one bounded page query."""
@@ -576,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
@@ -589,6 +678,7 @@ class LibraryOperations:
                     "tags": [],
                     "rating": row["rating"],
                     "notes": row["notes"],
+                    "archived": bool(row["archived"]),
                     "variants": [],
                 }
                 order.append(row["id"])
@@ -633,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
@@ -642,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_folders.py

@@ -0,0 +1,114 @@
+"""Library output-folder filter (#444): folder names, root prefixes and listing.
+
+A folder is a ``/``-joined path relative to an output root, as Images' folder
+filter and the legacy ``/prompt_manager/subfolders`` name them. A parent
+matches a folder when one of its linked images (``generated_images.prompt_id``,
+the link its previews and image count read) lies in that folder or below it,
+under any spelling of any output root.
+
+Matching compares ``generated_images.file_path``, which is always
+``normcase(normpath(abspath(path)))`` (``models.normalize_image_path``), with
+prefixes built by the same normalization here, so case folding is the
+platform's own on both sides (Windows folds non-ASCII too, unlike SQLite's
+``lower()``) and the comparison is an exact ``substr`` with no wildcards.
+"""
+
+import os
+import sqlite3
+
+_MAX_FOLDER = 1024
+_MAX_PREFIXES = 64
+
+
+def normalize_folder(value: object, pathmod=os.path) -> str:
+    """The folder as ``a/b``, outer slashes dropped; empty means no filter.
+
+    ``/`` always separates. ``\\`` separates only on Windows (``ntpath``): on
+    POSIX it is an ordinary character of a folder's name, and the listing
+    (``relative_folder``) names such a folder with it, so both agree.
+    ``.``/``..``/empty segments, control characters and (on Windows) a drive
+    are refused: the value only ever names a folder below a root.
+    """
+    if not isinstance(value, str) or len(value) > _MAX_FOLDER:
+        raise ValueError(f"folder must be text of at most {_MAX_FOLDER} characters")
+    windows = pathmod.sep == "\\"
+    folder = (value.replace("\\", "/") if windows else value).strip("/")
+    if not folder:
+        return ""
+    parts = folder.split("/")
+    if (
+        any(part in ("", ".", "..") for part in parts)
+        or (windows and ":" in parts[0])
+        or any(ord(char) < 32 or ord(char) == 127 for char in folder)
+    ):
+        raise ValueError("folder must be a folder below an output root")
+    return folder
+
+
+def _base(root: object, pathmod) -> str:
+    return pathmod.normcase(pathmod.normpath(str(root)))
+
+
+def folder_prefixes(roots, folder: str, pathmod=os.path) -> list[str]:
+    """One ``file_path`` prefix per root spelling for a normalized folder."""
+    prefixes = []
+    for root in roots:
+        joined = pathmod.join(_base(root, pathmod), *folder.split("/"))
+        prefix = pathmod.normcase(pathmod.normpath(joined)) + pathmod.sep
+        if prefix not in prefixes:
+            prefixes.append(prefix)
+    if not prefixes or len(prefixes) > _MAX_PREFIXES:
+        raise ValueError("No output folder to filter by")
+    return prefixes
+
+
+def folder_condition(prefixes: list[str]) -> tuple[str, list]:
+    """A condition on alias ``p``: one linked image under any prefix."""
+    tests = " OR ".join("substr(fg.file_path,1,?)=?" for _ in prefixes)
+    params = [value for prefix in prefixes for value in (len(prefix), prefix)]
+    return (
+        "EXISTS(SELECT 1 FROM generated_images fg WHERE fg.prompt_id=p.id "
+        f"AND ({tests}))",
+        params,
+    )
+
+
+def relative_folder(image_path: str, roots, pathmod=os.path) -> str | None:
+    """The folder of *image_path* below the first root spelling holding it.
+
+    The spelling of the stored path is kept (its case on Windows); ``None``
+    for an image directly in a root or outside every root.
+    """
+    directory = pathmod.normpath(pathmod.dirname(image_path))
+    folded = pathmod.normcase(directory)
+    for root in roots:
+        base = _base(root, pathmod)
+        start = base if base.endswith(pathmod.sep) else base + pathmod.sep
+        if not folded.startswith(start):
+            continue
+        # normcase keeps the length for ordinary names; else show the folded form.
+        rest = (directory if len(directory) == len(folded) else folded)[len(start) :]
+        parts = rest.split(pathmod.sep)
+        if any(part in ("", ".", "..") for part in parts):
+            return None
+        return "/".join(parts)
+    return None
+
+
+def list_library_folders(
+    conn: sqlite3.Connection, roots, eligible: str, pathmod=os.path
+) -> list[str]:
+    """Every folder (and ancestor) holding an image of an eligible parent."""
+    folders = set()
+    for (image_path,) in conn.execute(
+        "SELECT DISTINCT g.image_path FROM generated_images g "
+        "JOIN prompts p ON p.id=g.prompt_id WHERE " + eligible
+    ):
+        if not isinstance(image_path, str) or not image_path:
+            continue
+        folder = relative_folder(image_path, roots, pathmod)
+        if folder is None:
+            continue
+        parts = folder.split("/")
+        folders.update("/".join(parts[:index]) for index in range(1, len(parts) + 1))
+    return sorted(folders, key=lambda value: (value.casefold(), value))

database/maintenance.py

@@ -15,6 +15,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,9 +43,16 @@ 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
 
+try:
+    from ..utils.hashing import register_prompt_functions
+except ImportError:
+    from utils.hashing import register_prompt_functions
+
 MAINTENANCE_OPERATIONS = (
     "duplicate_prompts",
     "optimize",
@@ -80,6 +89,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",
@@ -88,7 +103,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
@@ -182,6 +198,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)."""
@@ -284,11 +302,96 @@ 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)
     try:
         conn.row_factory = sqlite3.Row
+        register_prompt_functions(conn)  # LIVE_HINT_CTES calls them (#460)
         conn.execute("BEGIN")  # one read snapshot for the whole plan
         return read(conn)
     finally:
@@ -425,8 +528,11 @@ 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))
         reader = (
             self._plan_duplicate_prompts
             if operation == "duplicate_prompts"
@@ -557,13 +663,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):
@@ -591,29 +699,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):
@@ -663,14 +789,18 @@ 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.
         """
         if operation not in MAINTENANCE_OPERATIONS or operation == "workflow_storage":
             raise ValueError(f"Unknown maintenance operation {operation!r}")
@@ -683,9 +813,12 @@ 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 == "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):
@@ -714,7 +847,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(
@@ -731,10 +864,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/models.py

@@ -38,7 +38,14 @@ from .review_schema import (
     review_without_annotations_error,
 )
 from .saved_views import migrate_saved_views, saved_views_schema_error
+
+try:
+    from ..utils.hashing import register_prompt_functions
+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,
@@ -153,12 +160,15 @@ def configure_connection(conn: sqlite3.Connection) -> None:
     (common on Windows) let other writers' polling busy handlers starve past
     BUSY_TIMEOUT_SECONDS and fail with "database is locked" (#242). Without
     WAL (for example an in-memory database) SQLite's FULL default is kept.
+    It also registers the prompt-key SQL functions (register_prompt_functions).
     """
     journal_mode = conn.execute("PRAGMA journal_mode = WAL").fetchone()[0]
     conn.execute("PRAGMA foreign_keys = ON")
     conn.execute(f"PRAGMA busy_timeout = {BUSY_TIMEOUT_SECONDS * 1000}")
     if str(journal_mode).lower() == "wal":
         conn.execute("PRAGMA synchronous = NORMAL")
+    # History reads and the live-hint rule call Python's prompt key (#460)
+    register_prompt_functions(conn)
 
 
 class PromptModel:
@@ -358,6 +368,10 @@ 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.
         migrate_workflow_documents(conn)
 
@@ -1234,6 +1248,16 @@ 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
+            )
+            if removals_error:
+                return False, removals_error
             views_error = saved_views_schema_error(
                 conn, allow_missing=not require_library_state
             )

database/operations.py

@@ -18,12 +18,14 @@ from .models import PromptModel, normalize_image_path, utc_now_iso
 from .pairs import PairDependencyError, PairOperations  # noqa: F401 (re-exported)
 from .library import LibraryOperations, titled
 from .image_favorites import ImageFavoriteOperations
+from .image_removal import ImageRemovalOperations, removed_from_library
 from .image_collections import ImageCollectionOperations
 from .annotations import AnnotationOperations
 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
@@ -448,6 +450,7 @@ class PromptDatabase(
     PairOperations,
     LibraryOperations,
     LibraryCommandOperations,
+    LibraryArchiveOperations,
     VocabularyOperations,
     ExecutionHistoryOperations,
     ExecutionContextOperations,
@@ -465,6 +468,7 @@ class PromptDatabase(
     WorkflowStorageOperations,
     CategoryImportOperations,
     ImageCategoryOperations,
+    ImageRemovalOperations,
 ):
     """Database operations class for managing prompts."""
 
@@ -1806,6 +1810,11 @@ class PromptDatabase(
         """
         if not isinstance(metadata, dict):
             metadata = {}
+        if removed_from_library(conn, image_path):
+            # #480: the user removed this exact file from the library; its
+            # bytes are unchanged, so nothing links it again.
+            self.logger.debug("Not linking a file removed from the library")
+            return 0
         filename = os.path.basename(image_path)
         file_path = normalize_image_path(image_path)
         file_size, width, height, image_format = _coerce_file_info(metadata)
@@ -2344,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/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/saved_views.py

@@ -33,6 +33,9 @@ MAX_TAG_LENGTH = 50
 _IMAGE_SORTS = ("newest", "oldest", "name", "name-desc", "rating")
 _LIBRARY_SORTS = ("newest", "oldest", "name", "rating")
 _MEDIA = ("all", "image", "video")
+# How a Library view's Tags match (#444): every tag, any of them, or parents
+# without tags ("untagged" never goes with tags; the listing refuses both).
+_TAG_MODES = ("all", "any", "untagged")
 _CONTROL = re.compile(r"[\x00-\x1f\x7f]")
 _SOURCE = re.compile(r"all|0|[1-9][0-9]{0,8}")
 # A collection id as the listing's ``collection`` query takes it (#331): none
@@ -67,6 +70,9 @@ _STATE_FIELDS = {
         "category": ("text", 100, ""),
         "sort": ("choice", _LIBRARY_SORTS, "newest"),
         "tags": ("tags", None, ()),
+        # #444: the Tags match and the output folder (relative, "/"-joined).
+        "tag_mode": ("choice", _TAG_MODES, "all"),
+        "folder": ("text", 1024, ""),
     },
 }
 
@@ -152,6 +158,8 @@ def encode_state(page: object, state: object) -> str:
         _validate_value(key, kind, limit, state[key])
     if state.get("favorites") is True and state.get("collection"):
         raise SavedViewInvalid("A view cannot filter by favorites and a collection")
+    if state.get("tag_mode") == "untagged" and state.get("tags"):
+        raise SavedViewInvalid("A view cannot list untagged prompts and filter tags")
     encoded = json.dumps(
         {key: state[key] for key in fields}, ensure_ascii=False, separators=(",", ":")
     )
@@ -195,6 +203,9 @@ def normalize_state(page: str, raw: object) -> dict:
     # as the listing does for a URL with both (#331 D140).
     if state.get("favorites") is True and state.get("collection"):
         state["favorites"] = False
+    # Never written by this build either: the tags win.
+    if state.get("tag_mode") == "untagged" and state.get("tags"):
+        state["tag_mode"] = "all"
     return state
 
 

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, _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
@@ -38,30 +39,76 @@ _IMAGE_FILES = (
 )
 
 
-def _tag_rows(page: str) -> str:
+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,"
         "COALESCE(MAX(images.n),0) AS image_count "
         "FROM page LEFT JOIN owners ON owners.tag_id=page.id "
         "LEFT JOIN images ON images.tag_id=page.id "
-        "GROUP BY page.id ORDER BY page.id"
+        "GROUP BY page.id ORDER BY " + order
     )
 
 
 _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 = {
+    "created": (None, "ASC"),
+    "name": ("name COLLATE NOCASE", "ASC"),
+    "name-desc": ("name COLLATE NOCASE", "DESC"),
+    "prompts": ("prompt_count", "DESC"),
+    "prompts-asc": ("prompt_count", "ASC"),
+}
+_MAX_SEARCH = 200
+_TAG_SEARCH = "name LIKE ? ESCAPE '\\'"
+# Every matching tag ranked by the prompt count its row shows (the owners rule
+# of ``_tag_rows``), then one keyset page of it with its image counts.
+_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 "
+    "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),"
+    "page AS (SELECT id,name,prompt_count FROM counted WHERE {after} "
+    "ORDER BY prompt_count {direction},id LIMIT ?),"
+    "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,page.prompt_count,"
+    "COALESCE(images.n,0) AS image_count FROM page "
+    "LEFT JOIN images ON images.tag_id=page.id "
+    "ORDER BY page.prompt_count {direction},page.id"
+)
+_UNTAGGED_COUNT = (
+    "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.
 _TAG_IMPACT = (
@@ -100,7 +147,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
@@ -109,15 +158,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,"
@@ -210,6 +261,28 @@ def _tag_after(after: object) -> int:
     return after
 
 
+def _tag_search(search: object) -> str:
+    if not isinstance(search, str) or len(search) > _MAX_SEARCH:
+        raise ValueError(f"Tag search must be text of at most {_MAX_SEARCH} characters")
+    return search.strip()
+
+
+def _sorted_tag_after(after: object, sort: str) -> tuple | None:
+    """The ``[key, id]`` cursor of a name or prompt-count sort."""
+    if after is None:
+        return None
+    key_type = str if sort.startswith("name") else int
+    if (
+        not isinstance(after, list)
+        or len(after) != 2
+        or type(after[0]) is not key_type
+        or type(after[1]) is not int
+        or after[1] <= 0
+    ):
+        raise ValueError("Invalid vocabulary cursor")
+    return after[0], after[1]
+
+
 def _category_after(after: object) -> tuple[int, str] | None:
     if after is None:
         return None
@@ -431,22 +504,80 @@ class VocabularyOperations:
         epoch: str,
         *,
         limit: int = 100,
-        after: int | None = None,
+        after: object = None,
+        search: str = "",
+        sort: str = "created",
     ) -> dict:
+        """One page of tags matching ``search`` (a literal, case-insensitive
+        substring) in ``sort`` order, with the untagged parents' count."""
         _limit(limit)
-        start = _tag_after(after)
-        total = conn.execute("SELECT COUNT(*) FROM tags").fetchone()[0]
-        rows = [dict(row) for row in conn.execute(_TAG_PAGE, (start, limit + 1))]
+        search = _tag_search(search)
+        if not isinstance(sort, str) or sort not in TAG_SORTS:
+            raise ValueError("sort must be one of " + ", ".join(TAG_SORTS))
+        where, params = "1=1", []
+        if search:
+            where, params = _TAG_SEARCH, [f"%{_escape(search)}%"]
+        total, untagged = conn.execute(
+            f"SELECT (SELECT COUNT(*) FROM tags WHERE {where}),({_UNTAGGED_COUNT})",
+            params,
+        ).fetchone()
+        rows = [
+            dict(row) for row in self._tag_page(conn, where, params, sort, after, limit)
+        ]
         has_more = len(rows) > limit
         rows = rows[:limit]
+        key, _direction = TAG_SORTS[sort]
+        next_key = None
+        if has_more:
+            last = rows[-1]
+            next_key = (
+                last["id"]
+                if key is None
+                else [
+                    last["prompt_count" if key == "prompt_count" else "name"],
+                    last["id"],
+                ]
+            )
         return {
             "library_epoch": epoch,
             "tags": rows,
             "total": total,
+            "untagged_count": untagged,
+            # Echoed so a client can tell this read applied its search and sort.
+            "search": search,
+            "sort": sort,
             "has_more": has_more,
-            "next_key": rows[-1]["id"] if has_more else None,
+            "next_key": next_key,
         }
 
+    @staticmethod
+    def _tag_page(conn, where, params, sort, after, limit):
+        key, direction = TAG_SORTS[sort]
+        if key is None:
+            page = (
+                f"SELECT id,name FROM tags WHERE {where} AND id>? ORDER BY id LIMIT ?"
+            )
+            return conn.execute(
+                _tag_rows(page), [*params, _tag_after(after), limit + 1]
+            )
+        cursor = _sorted_tag_after(after, sort)
+        operator = ">" if direction == "ASC" else "<"
+        keyset, keys = "1=1", []
+        if cursor is not None:
+            keyset = f"({key}{operator}? OR ({key}=? AND id>?))"
+            keys = [cursor[0], cursor[0], cursor[1]]
+        if key == "prompt_count":
+            sql = _RANKED_TAG_PAGE.format(
+                search=where, after=keyset, direction=direction
+            )
+            return conn.execute(sql, [*params, *keys, limit + 1])
+        page = (
+            f"SELECT id,name FROM tags WHERE {where} AND {keyset} "
+            f"ORDER BY {key} {direction},id LIMIT ?"
+        )
+        order = f"page.{key} {direction},page.id"
+        return conn.execute(_tag_rows(page, order), [*params, *keys, limit + 1])
+
     def _vocabulary_categories(
         self,
         conn: sqlite3.Connection,
@@ -490,9 +621,18 @@ class VocabularyOperations:
             ),
         }
 
-    def list_vocabulary_tags(self, *, limit: int = 100, after: int | None = None):
+    def list_vocabulary_tags(
+        self,
+        *,
+        limit: int = 100,
+        after: object = None,
+        search: str = "",
+        sort: str = "created",
+    ):
         with self.library_snapshot() as (conn, epoch, _ceiling):
-            return self._vocabulary_tags(conn, epoch, limit=limit, after=after)
+            return self._vocabulary_tags(
+                conn, epoch, limit=limit, after=after, search=search, sort=sort
+            )
 
     def list_vocabulary_categories(
         self, *, limit: int = 100, after: list | None = None

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
@@ -74,6 +77,8 @@ from .autotag_routes import AutotagRoutesMixin
 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:
@@ -313,6 +318,7 @@ async def _gzip_middleware(request, handler):
 class PromptManagerAPI(
     PairRoutesMixin,
     LibraryRoutesMixin,
+    LibraryDeleteRoutesMixin,
     VocabularyRoutesMixin,
     AnnotationRoutesMixin,
     ReviewRoutesMixin,
@@ -343,6 +349,8 @@ class PromptManagerAPI(
     LoraIntegrationMixin,
     LoraImportRoutesMixin,
     ImageCategoryRoutesMixin,
+    ImageDeleteMixin,
+    ImageReturnMixin,
     SourceRescopeMixin,
 ):
     """REST API handler for PromptManager operations and web interface.
@@ -423,6 +431,12 @@ class PromptManagerAPI(
         # Reviewed maintenance plans under single-use tokens; a restart
         # forgets them (py/api/maintenance.py).
         self._maintenance_plans = PlanStore()
+        # #480: the removal marks the workspace listing hides, read once now so
+        # a listing page reads no table until a mark or the database changes.
+        try:
+            self._removal_marks()
+        except Exception:
+            self.logger.warning("Removal marks could not be read", exc_info=True)
 
         # A safety backup interrupted by a crash leaves its temp copy behind.
         try:
@@ -653,40 +667,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)
@@ -710,6 +729,8 @@ class PromptManagerAPI(
         self._register_lora_routes(routes)
         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)
@@ -726,8 +747,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.
@@ -742,10 +766,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:
@@ -762,7 +789,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(

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))
             )
@@ -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).
@@ -883,6 +901,8 @@ class AdminRoutesMixin:
         extensions = IMAGE_EXTENSIONS + VIDEO_EXTENSIONS
         media_files = []
         seen = set()
+        # #480: a file removed from the library is never a copy or a keeper.
+        marks = self._removal_marks()
         for slot, root in enumerate(roots):
             for path in _collect_media_files([root], extensions):
                 try:
@@ -891,6 +911,9 @@ class AdminRoutesMixin:
                     stat = os.stat(path)
                 except OSError:
                     continue
+                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:
                     seen.add(key)
@@ -1074,13 +1097,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}"
@@ -1089,15 +1120,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):
@@ -1371,13 +1406,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}")
@@ -1385,20 +1424,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")
@@ -1436,7 +1481,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"]``
@@ -1474,9 +1519,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),
@@ -1551,8 +1603,13 @@ class AdminRoutesMixin:
             if job.cancelled:
                 return
             batch = media_files[batch_start : batch_start + SCAN_BATCH_SIZE]
-            skipped = [f for f in batch if str(f) in known]
-            to_read = [f for f in batch if str(f) not in known]
+            # #480: a file removed from the library is not read or linked
+            # again while its bytes are unchanged.
+            removed = await self._run_in_executor(self._removed_paths, batch)
+            skipped = [f for f in batch if str(f) in known or str(f) in removed]
+            to_read = [
+                f for f in batch if str(f) not in known and str(f) not in removed
+            ]
             outcomes = [_scan_outcome(f, roots, "skipped") for f in skipped]
             counts["processed"] += len(skipped)
             counts["unchanged"] += len(skipped)

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."
@@ -844,9 +849,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
 
@@ -1319,8 +1327,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 +1339,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 +1366,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/diagnostics_report.py

@@ -213,6 +213,8 @@ def public_config():
             "max_image_age_days": GalleryConfig.MAX_IMAGE_AGE_DAYS,
             "max_concurrent_processing": GalleryConfig.MAX_CONCURRENT_PROCESSING,
             "metadata_extraction_timeout": GalleryConfig.METADATA_EXTRACTION_TIMEOUT,
+            # config.json gallery.monitoring.save_foreign_runs (#247)
+            "save_foreign_runs": GalleryConfig.SAVE_FOREIGN_RUNS,
         },
         "integrations": {
             "lora_manager_enabled": bool(IntegrationConfig.LORA_MANAGER_ENABLED),

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/history.py

@@ -27,7 +27,11 @@ from .history_current import (
     resolve_current_candidate,
     snapshot_current_output_scope,
 )
-from .inspection_service import CurrentFileChanged, inspect_current_file
+from .inspection_service import (
+    CurrentFileChanged,
+    image_tags_reader,
+    inspect_current_file,
+)
 from .images import image_category_key
 
 try:
@@ -778,6 +782,14 @@ class HistoryRoutesMixin:
                             ),
                             own,
                         ),
+                        # The file's own image tags, keyed as the Tag dialog
+                        # keys them: the normcased realpath (#457).
+                        image_tags=image_tags_reader(
+                            db,
+                            str(resolved.canonical_target),
+                            self.logger,
+                            close=True,
+                        ),
                         recorded=recorded_output_lookup(
                             db, close=True, reference=receipt["reference"]
                         ),

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

@@ -0,0 +1,521 @@
+"""Delete selected images: remove them from the library, or trash them (#480).
+
+``POST /prompt_manager/images/delete/preview {items | filter}`` reviews the
+selection for both modes at once and answers a reviewed plan (the System ›
+Maintenance contract, ``py/api/maintenance.py``): the token, the counts of
+the records each file loses, the refused files with a coded reason (both
+modes, and the trash-only ones), the prompts that lose their last image and
+the safety backup. It writes nothing. ``items`` are Images listing identities
+(at most 500); ``filter`` is "Select all N matching". Each target is matched
+to its entry in the current listing (``not-listed`` otherwise), so every
+later step uses the walked spelling, never the request's text.
+
+``POST /prompt_manager/maintenance/apply {token, mode}`` runs the plan as the
+``image_delete`` job (admission ``records``, not cancellable): the sources
+are checked, the promised safety backup is taken, then each file is checked
+again (containment and fingerprint) and
+
+- ``mode="library"``: its records are deleted and its removal mark written
+  in one transaction; the file stays on disk, hidden;
+- ``mode="trash"``: it is moved to the trash first (``utils/trash.py``); a
+  refused move leaves the file and its records alone; after a move its
+  records are deleted.
+
+No path here unlinks a media file. A failed records write stops the job; the
+files after it are not attempted.
+"""
+
+import hashlib
+import json
+import os
+import stat as stat_module
+from pathlib import Path
+
+from .admin import DELETABLE_MEDIA_EXTENSIONS, _remove_thumbnail
+from .duplicate_review import (
+    CHANGED_CONTENT,
+    GONE,
+    LINK,
+    OUTSIDE_SOURCE,
+    UNREADABLE,
+    file_problem,
+)
+from .images import _favorite_relative_key, _favorite_root_key
+from .jobs import IMAGE_DELETE_GROUP_NAMES, IMAGE_DELETE_JOB_KIND, JobBusy
+from .job_routes import busy_response
+from .maintenance import MAINTENANCE_PREVIEW_LEASE, _fail_job, _failure, backup_note
+
+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")
+TITLES = {
+    "library": "Delete images (remove from the library)",
+    "trash": "Delete images (move to the trash)",
+}
+MAX_DETAIL_ROWS = 50
+# 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 = {
+    "not-listed": "No longer listed in Images.",
+    "outside": "Outside the output folders.",
+    "link": "A link, not a file.",
+    "gone": "The file no longer exists.",
+    "unreadable": "The file can't be read.",
+    "thumbnail": "A generated preview, not an image.",
+    "not-media": "Not a media file.",
+    "trash-folder": "Inside a trash folder.",
+    "path-too-long": "The path is too long.",
+    "changed": "The file changed after the review.",
+    trash.NO_TRASH: "This drive has no trash.",
+    trash.NETWORK: "Network drives can't use the trash.",
+    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 = {
+    OUTSIDE_SOURCE: "outside",
+    LINK: "link",
+    GONE: "gone",
+    UNREADABLE: "unreadable",
+    CHANGED_CONTENT: "changed",
+}
+SOURCES_CHANGED = "The sources changed since the review; nothing was changed."
+RECORDS_FAILED = "The image records could not be removed; the job stopped."
+
+
+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]):
+        return "trash-folder"
+    if os.path.splitext(parts[-1])[1].lower() not in DELETABLE_MEDIA_EXTENSIONS:
+        return "not-media"
+    return None
+
+
+def file_check(path, root, relative, *, expected=None):
+    """``(code, lstat)``: why *path* may not be deleted, or ``(None, lstat)``.
+
+    Text first (inside the root, not a preview or trash folder, a media
+    file), then ``duplicate_review.file_problem``: a regular file, not a
+    link, whose resolved path is still inside the resolved root. With
+    *expected* (``{size, mtime_ns, inode}`` from the review) the file must
+    still be those bytes. Every handle is closed when this returns.
+    """
+    code = lexical_problem(relative)
+    if code:
+        return code, None
+    try:
+        info = os.lstat(path)
+    except FileNotFoundError:
+        return "gone", None
+    except OSError:
+        return "unreadable", None
+    if getattr(info, "st_file_attributes", 0) & getattr(
+        stat_module, "FILE_ATTRIBUTE_REPARSE_POINT", 0
+    ):
+        return "link", info
+    if expected is None:
+        problem = file_problem(path, root, None, info.st_size, hash_check=False)
+    else:
+        problem = file_problem(
+            path,
+            root,
+            None,
+            expected["size"],
+            hash_check=False,
+            mtime_ns=expected["mtime_ns"],
+            inode=expected["inode"],
+        )
+    if problem:
+        return _PROBLEM_CODES.get(problem, "outside"), info
+    if len(os.path.normcase(os.path.realpath(path))) > MAX_LOCATOR_LENGTH:
+        return "path-too-long", info
+    return None, info
+
+
+def _row(target, code):
+    return {
+        "path": target["rel"],
+        "root_label": target["root_label"],
+        "code": code,
+        "detail": REASONS[code],
+    }
+
+
+def _summary(rows):
+    counts = {}
+    for row in rows:
+        counts[row["code"]] = counts.get(row["code"], 0) + 1
+    return {"count": len(rows), "counts": counts, "rows": rows[:MAX_DETAIL_ROWS]}
+
+
+def _fingerprint(targets) -> str:
+    material = [
+        (t["root_index"], t["rel"], t["size"], t["mtime_ns"], t["inode"])
+        for t in targets
+    ]
+    return hashlib.sha256(json.dumps(material).encode("utf-8")).hexdigest()
+
+
+class ImageDeleteMixin:
+    """The Images Delete… review and its apply job (#480)."""
+
+    def _register_image_delete_routes(self, routes):
+        # A fixed path beside the dynamic /images/{image_id} routes: aiohttp
+        # (3.10+) looks fixed paths up first (see image_categories.py).
+        @routes.post("/prompt_manager/images/delete/preview")
+        async def preview_image_delete_route(request):
+            return await self.preview_image_delete(request)
+
+    # ── review ────────────────────────────────────────────────────────
+
+    async def preview_image_delete(self, request):
+        """Review the selection under the maintenance preview lease."""
+        try:
+            async with self._jobs.lease(MAINTENANCE_PREVIEW_LEASE):
+                return await self._preview_image_delete(request)
+        except JobBusy as busy:
+            return busy_response(busy)
+
+    async def _preview_image_delete(self, request):
+        _data, targets, roots, error = await self._category_request(
+            request, frozenset()
+        )
+        if error is not None:
+            return error
+        db = self.db
+        identity = self._database_identity(db)
+        try:
+            files = {}
+            for index in dict.fromkeys(t[0] for t in targets):
+                files[index] = await self._get_gallery_files(roots[index])
+            review = await self._run_in_executor(
+                self._review_image_delete, db, roots, targets, files
+            )
+        except Exception:
+            self.logger.exception("Image delete review failed")
+            return _failure("The review failed. Check server logs for details.", 500)
+        return await self._plan_response(
+            db,
+            identity,
+            IMAGE_DELETE,
+            _fingerprint(review["targets"]),
+            review["public"],
+            {"targets": review["targets"]},
+        )
+
+    def _review_image_delete(self, db, roots, targets, files):
+        """Every target checked now (blocking); ``{targets, public}``."""
+        listings, seen = {}, set()
+        acted, refused, trash_refused = [], [], []
+        spellings = self._source_spellings()
+        root_texts = [str(root) for root in roots]
+        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))
+            target = {
+                "root_index": index,
+                "root_scope": _scope,
+                "root": str(root),
+                "root_label": self._public_path(root),
+                "rel": spelled or relative,
+            }
+            if spelled is None:
+                refused.append(_row(target, "not-listed"))
+                continue
+            absolute = os.path.join(str(root), *spelled.split("/"))
+            code, info = file_check(absolute, str(root), spelled)
+            if code:
+                refused.append(_row(target, code))
+                continue
+            identity = (info.st_dev, info.st_ino)
+            if identity in seen:
+                continue  # one physical file named twice (nested roots, hardlinks)
+            seen.add(identity)
+            target.update(self._removal_of(absolute, info, spellings))
+            target["trash"] = trash.trash_available(absolute, root_texts)
+            if target["trash"]:
+                trash_refused.append(_row(target, target["trash"]))
+            acted.append(target)
+        counted = db.count_removal_records([t["removal"] for t in acted])
+        library_count = len(acted)
+        public = {
+            "operation": IMAGE_DELETE,
+            "scope": {"count": library_count, "label": "images"},
+            "library": {"count": library_count},
+            "trash": {"count": library_count - len(trash_refused)},
+            "protected": {"count": len(refused), "reasons": []},
+            "refused": _summary(refused),
+            "trash_refused": _summary(trash_refused),
+            "records": counted["records"],
+            "orphaned_prompts": counted["orphaned_prompts"],
+            "media": {"deletes_files": False, "label": "Chosen in the review"},
+            "details": [],
+        }
+        return {"targets": acted, "public": public}
+
+    @staticmethod
+    def _removal_of(absolute, info, spellings):
+        favorites = [
+            (_favorite_root_key(root), _favorite_relative_key(relative))
+            for root, relative in spellings.favorites(absolute).items()
+        ]
+        return {
+            "abs": absolute,
+            "size": info.st_size,
+            "mtime_ns": info.st_mtime_ns,
+            "inode": [info.st_dev, info.st_ino],
+            "removal": {
+                "spellings": list(spellings.of(absolute).values()),
+                "locator": os.path.normcase(os.path.realpath(absolute)),
+                "favorites": favorites,
+                "category": spellings.category_key(absolute),
+            },
+        }
+
+    # ── apply ─────────────────────────────────────────────────────────
+
+    def _start_image_delete(self, db, plan, mode, backup_on):
+        """Register the delete job for a reviewed plan (on the loop)."""
+        return self._jobs.start(
+            IMAGE_DELETE_JOB_KIND,
+            TITLES[mode],
+            admission="records",
+            cancellable=False,
+            scope={"operation": IMAGE_DELETE, "mode": mode},
+            groups=IMAGE_DELETE_GROUP_NAMES,
+            runner=lambda job: self._run_image_delete(job, db, plan, mode, backup_on),
+        )
+
+    async def _run_image_delete(self, job, db, plan, mode, backup_on):
+        targets = plan.payload["targets"]
+        unchanged = await self._run_in_executor(self._sources_unchanged, targets)
+        if not unchanged:
+            return _fail_job(job, "sources-changed", SOURCES_CHANGED)
+        ok, backup = await self._safety_backup_step(job, db, backup_on)
+        if not ok:
+            return
+        try:
+            await self._delete_images(job, db, targets, mode, backup)
+        finally:
+            if backup:
+                await self._prune_backup_pool(job, db)
+
+    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, None
+        for done, target in enumerate(targets, start=1):
+            if failed:
+                self._record_delete(job, "not_changed", target, "not-attempted")
+                continue
+            group, code, label = await self._run_in_executor(
+                self._delete_one_image, db, target, mode, roots
+            )
+            if group == "errors":
+                failed = code or RECORDS_FAILED_CODE
+                text = REASONS[code] if code else RECORDS_FAILED
+                job.record(
+                    "errors",
+                    {
+                        "path": target["rel"],
+                        "root": target["root_label"],
+                        "root_index": target["root_index"],
+                        "root_scope": target["root_scope"],
+                        "detail": text,
+                    },
+                    f"! {target['root_label']}/{target['rel']} · {text}",
+                )
+            elif group == done_group:
+                changed += 1
+                detail = (
+                    f"Moved to the {label}" if label else "Removed from the library"
+                )
+                self._record_delete(job, group, target, None, detail)
+            else:
+                self._record_delete(job, group, target, code)
+            job.update(processed=done, progress=int(done / len(targets) * 100))
+        if changed:
+            self.invalidate_gallery_cache()
+        verb = "Moved" if mode == "trash" else "Removed"
+        where = "to the trash" if mode == "trash" else "from the library"
+        line = f"{verb} {changed:,} of {len(targets):,} images {where}"
+        job.log_line(line)
+        job.update(progress=100, status_line=f"{line} · {backup_note(backup)}")
+        if failed:
+            # "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
+    def _record_delete(job, group, target, code, detail=None):
+        text = detail or REASONS[code]
+        row = {
+            "path": target["rel"],
+            "root": target["root_label"],
+            "root_index": target["root_index"],
+            "root_scope": target["root_scope"],
+            "code": code,
+            "detail": text,
+        }
+        mark = {"removed": "-", "trashed": "-"}.get(group, "=")
+        job.record(
+            group, row, f"{mark} {target['root_label']}/{target['rel']} · {text}"
+        )
+
+    def _delete_one_image(self, db, target, mode, roots):
+        """``(group, code, trash label)`` for one reviewed file (blocking)."""
+        code, _info = file_check(
+            target["abs"],
+            target["root"],
+            target["rel"],
+            expected={
+                "size": target["size"],
+                "mtime_ns": target["mtime_ns"],
+                "inode": target["inode"],
+            },
+        )
+        if code:
+            return "not_changed", code, None
+        label = None
+        if mode == "trash":
+            try:
+                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"])
+        try:
+            db.remove_image_records(target["removal"], mark=mark)
+        except Exception:
+            self.logger.exception("Removing an image's records failed")
+            return "errors", None, label
+        _remove_thumbnail(Path(target["root"]), Path(target["rel"]), self.logger)
+        return ("trashed" if mode == "trash" else "removed"), None, label
+
+    # ── the workspace listing ─────────────────────────────────────────
+
+    def _removal_marks(self) -> dict:
+        """The removal marks, cached until a mark changes or the database does.
+
+        A listing page reads no table while nothing changed: the cache follows
+        this process's marks generation and the database identity (a restore).
+        """
+        key = (self._database_identity(self.db), image_removal_schema.generation())
+        cached = getattr(self, "_marks_cache", None)
+        if cached is None or cached[0] != key:
+            cached = (key, self.db.removal_marks())
+            self._marks_cache = cached
+        return cached[1]
+
+    def _hidden_entries(self, listings) -> set:
+        """Normalized paths of listed files a removal mark hides (blocking).
+
+        Only marked locators are looked at, and only those files are
+        ``lstat``'ed: a mark hides a file while its size and ``mtime_ns`` are
+        still the ones it was removed with.
+        """
+        marks = self._removal_marks()
+        if not marks:
+            return set()
+        located = self._workspace_locators(listings)
+        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()
+        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."""
+    return value if isinstance(value, str) and value in MODES else None

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

@@ -488,7 +488,11 @@ class ImageTagJobsMixin:
                     _still_targets, self, index, root, files, seen
                 )
             )
-        return found
+        # #480: a file removed from the library is never tagged by a batch.
+        hidden = await self._run_in_executor(
+            self._removed_paths, [target.path for target in found]
+        )
+        return [target for target in found if target.path not in hidden]
 
     async def _start_autotag(
         self, *, model_type, existing, options, keep_in_memory, title, owner_kind

py/api/images.py

@@ -55,6 +55,7 @@ from .inspection_service import (
     IMAGE_EXTENSIONS as _IMAGE_EXTENSIONS,
     SERVABLE_EXTENSIONS,
     VIDEO_EXTENSIONS,
+    image_tags_reader,
     inspect_current_file,
 )
 
@@ -65,6 +66,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 (
@@ -78,7 +85,7 @@ try:
         OpenerUnavailable,
         RevealGate,
         RevealTargetChanged,
-        is_local_request,
+        is_same_origin_request,
         reveal_executor,
         reveal_in_file_manager,
     )
@@ -90,6 +97,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 (
@@ -103,7 +116,7 @@ except ImportError:
         OpenerUnavailable,
         RevealGate,
         RevealTargetChanged,
-        is_local_request,
+        is_same_origin_request,
         reveal_executor,
         reveal_in_file_manager,
     )
@@ -274,10 +287,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")
@@ -285,6 +308,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
 
 
@@ -792,12 +817,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,
     )
@@ -914,7 +939,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
@@ -926,8 +952,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:
@@ -951,7 +975,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,)))
 
 
@@ -975,8 +999,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
@@ -1158,7 +1184,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()
@@ -1173,6 +1202,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)
@@ -1200,6 +1235,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",
     }
 
 
@@ -1502,7 +1538,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)
@@ -2071,6 +2107,7 @@ class ImageRoutesMixin:
                         filters["tags"],
                         filters["favorites"],
                         filters["collection"],
+                        filters["removed"],
                     ]
                 ).encode()
             ).hexdigest()
@@ -2100,6 +2137,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:
@@ -2126,6 +2168,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"]
@@ -2449,6 +2492,13 @@ class ImageRoutesMixin:
             def key(entry):
                 return _workspace_key(entry, order)
 
+        # #480: files removed from the library stay on disk, hidden here; the
+        # removed view (#498) lists only them.
+        hidden = self._hidden_entries(listings)
+        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,
             key,
@@ -2735,6 +2785,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
@@ -2758,6 +2811,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)
 
@@ -2809,8 +2864,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
@@ -3070,7 +3126,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)
@@ -3079,11 +3137,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)
@@ -3651,12 +3712,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(
@@ -3734,7 +3795,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,
             )
@@ -3962,28 +4025,9 @@ class ImageRoutesMixin:
             )
 
     def _inspection_image_tags(self, root: Path, parts: tuple) -> Callable:
-        """The inspector's ``image_tags`` reader for one rooted file (S1).
-
-        The returned function takes the inspection's own stable stat and
-        answers ``current`` only for a byte record verified at exactly that
-        ``(size, mtime_ns)``; a failed read is ``unavailable``, never an error.
-        """
+        """The inspector's ``image_tags`` reader for one rooted file (S1)."""
         locator = os.path.normcase(os.path.join(os.path.realpath(root), *parts))
-
-        def read(size: int, mtime_ns: int) -> dict:
-            try:
-                found = self.db.current_image_tags({locator: (size, mtime_ns)})
-            except Exception:
-                self.logger.warning(
-                    "Inspection image tags could not be read", exc_info=True
-                )
-                return {"state": "unavailable", "tags": []}
-            tags = found.get(locator)
-            if tags is None:
-                return {"state": "none", "tags": []}
-            return {"state": "current", "tags": list(tags)}
-
-        return read
+        return image_tags_reader(self.db, locator, self.logger)
 
     def _read_own_category(self, sources, path):
         """One file's own category, None, or OWN_CATEGORY_UNREAD (blocking).

py/api/inspection_service.py

@@ -93,6 +93,35 @@ def _in_host_temp(path: Path) -> bool:
         return False
 
 
+def image_tags_reader(
+    db: Any, locator: str, logger: Any, *, close: bool = False
+) -> Callable[[int, int], dict]:
+    """The inspector's ``image_tags`` reader for one file *locator* (S1).
+
+    The returned function takes the inspection's own stable stat and answers
+    ``current`` only for a byte record verified at exactly that
+    ``(size, mtime_ns)``; a failed read is ``unavailable``, never an error.
+    ``close`` closes the calling thread's connection afterwards, for routes
+    that close theirs after reading (History).
+    """
+
+    def read(size: int, mtime_ns: int) -> dict:
+        try:
+            found = db.current_image_tags({locator: (size, mtime_ns)})
+        except Exception:
+            logger.warning("Inspection image tags could not be read", exc_info=True)
+            return {"state": "unavailable", "tags": []}
+        finally:
+            if close:
+                db.model.close()
+        tags = found.get(locator)
+        if tags is None:
+            return {"state": "none", "tags": []}
+        return {"state": "current", "tags": list(tags)}
+
+    return read
+
+
 def inspect_current_file(
     path: Path,
     *,

py/api/jobs.py

@@ -123,6 +123,11 @@ 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
 # no file outcomes, so it may run beside a scan.
 HEALTH_JOB_KIND = "health"

py/api/library.py

@@ -21,9 +21,11 @@ from .pairs import pair_response
 try:
     from ...database.library import ARCHIVE_SCOPES
     from ...database.library_commands import LibraryCommandError
+    from ...database.library_folders import folder_prefixes, normalize_folder
 except ImportError:
     from database.library import ARCHIVE_SCOPES
     from database.library_commands import LibraryCommandError
+    from database.library_folders import folder_prefixes, normalize_folder
 
 _SECRET = os.urandom(32)
 
@@ -155,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
@@ -283,36 +292,87 @@ class LibraryRoutesMixin:
         outputs = list(self._get_all_output_dirs())
         return self._media_roots(outputs), len(outputs)
 
+    def _library_root_spellings(self) -> list[str]:
+        """Every spelling of each output root (#444 folder filter): the
+        validated, resolved root and each configured or detected path that
+        resolves to it, so images saved through a symlinked root match."""
+        from ..config import GalleryConfig
+
+        roots = [str(root) for root in self._get_all_output_dirs()]
+        candidates = [
+            *GalleryConfig.MONITORING_DIRECTORIES,
+            self._find_comfyui_output_dir(),
+            *self._comfyui_output_dir_list(),
+        ]
+        spellings = []
+        for root in roots:
+            for value in (root, *candidates):
+                if not value or not isinstance(value, (str, os.PathLike)):
+                    continue
+                value = str(value)
+                same = value == root or os.path.normcase(
+                    os.path.realpath(value)
+                ) == os.path.normcase(root)
+                if same and value not in spellings:
+                    spellings.append(value)
+        return spellings
+
     @pair_response
     async def get_library_options(self, request: web.Request) -> web.Response:
-        result = await self._run_in_executor(self.db.get_library_options)
+        include = request.query.get("include", "")
+        if include not in ("", "folders"):
+            raise ValueError("include must be empty or 'folders'")
+        if include:
+            # Folders are read only when Refine opens (a walk of linked paths).
+            roots = await self._run_in_executor(self._library_root_spellings)
+            result = await self._run_in_executor(self.db.get_library_options, roots)
+        else:
+            result = await self._run_in_executor(self.db.get_library_options)
         return web.json_response({"success": True, **result})
 
     @pair_response
     async def list_library(self, request: web.Request) -> web.Response:
+        folder = normalize_folder(request.query.get("folder", ""))
         options = {
             "text": request.query.get("text", ""),
             "category": request.query.get("category"),
             "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")
+        roots = (
+            await self._run_in_executor(self._library_root_spellings) if folder else []
+        )
+        prefixes = folder_prefixes(roots, folder) if folder else None
 
         def read() -> tuple[dict, dict]:
             with self.db.library_snapshot(include_ceiling=True) as snapshot:
                 conn, epoch, ceiling = snapshot
+                # The folder and the roots it was resolved under join the scope.
                 scope = _scope(
-                    {"kind": "parents", "projection": 2, "epoch": epoch, **options}
+                    {
+                        "kind": "parents",
+                        "projection": 2,
+                        "epoch": epoch,
+                        **options,
+                        **({"folder": folder, "roots": roots} if folder else {}),
+                    }
                 )
                 result = self.db._list_library(
                     conn,
                     epoch,
                     ceiling,
                     **options,
+                    folder_prefixes=prefixes,
                     limit=limit,
                     after=_decode(cursor, scope),
                 )
+                result["folder"] = folder
                 records = self.db.get_library_previews(
                     [parent["id"] for parent in result["items"]]
                 )

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/maintenance.py

@@ -281,7 +281,7 @@ def short_space_message(short):
     )
 
 
-def _changed_line(operation, count):
+def _changed_line(operation, count, marks=0):
     if operation == "optimize":
         return "Optimized the database"
     if operation == "duplicate_prompts":
@@ -291,7 +291,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 +423,7 @@ class MaintenanceRoutesMixin:
             draft.public(),
             {
                 "targets": draft.targets,
+                "marks": draft.marks,
                 "roots": roots,
                 "protected": draft.protected["count"],
                 "rows": draft.scope["count"],
@@ -488,7 +493,9 @@ class MaintenanceRoutesMixin:
     # ── apply ─────────────────────────────────────────────────────────
 
     async def apply_maintenance_plan(self, request):
-        data, error = await self._read_maintenance_body(request, ("token",))
+        from .image_delete import IMAGE_DELETE, parse_mode
+
+        data, error = await self._read_maintenance_body(request, ("token", "mode"))
         if error is not None:
             return error
         token = data.get("token")
@@ -497,6 +504,13 @@ class MaintenanceRoutesMixin:
         plan, refused = self._peek_plan(token)
         if refused is not None:
             return refused
+        mode = None
+        if plan.operation == IMAGE_DELETE:
+            mode = parse_mode(data.get("mode"))
+            if mode is None:
+                return _invalid('mode must be "library" or "trash"')
+        elif "mode" in data:
+            return _invalid("Unknown field: mode")
         store = self._maintenance_plans
         if plan.operation == ROTATED_LOGS:
             # Log files only: no database identity, backup or disk check.
@@ -544,6 +558,8 @@ class MaintenanceRoutesMixin:
         try:
             if plan.operation == DUPLICATE_FILES:
                 job = self._start_duplicate_cleanup(db, plan, backup)
+            elif plan.operation == IMAGE_DELETE:
+                job = self._start_image_delete(db, plan, mode, backup)
             elif plan.operation == WORKFLOW_STORAGE:
                 job = self._start_workflow_storage(db, backup)
             else:
@@ -642,6 +658,7 @@ class MaintenanceRoutesMixin:
                 plan.fingerprint,
                 payload["targets"],
                 payload["roots"],
+                payload.get("marks", ()),
             )
         except PlanChanged:
             return _fail_job(job, "plan-changed", PLAN_CHANGED)
@@ -653,7 +670,7 @@ class MaintenanceRoutesMixin:
         )
         job.record_counted("changed", result["changed"], rows)
         job.record_counted("protected", payload["protected"])
-        line = _changed_line(plan.operation, result["changed"])
+        line = _changed_line(plan.operation, result["changed"], result.get("marks", 0))
         job.log_line(line)
         job.update(progress=100, status_line=f"{line} · {backup_note(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

@@ -8,7 +8,8 @@ is shared with the LoRA integration's save.
 GET also reports the image sources and the monitor (``sources.py``). A save
 that changes the sources runs as a ``source_change`` registry hold, so it is
 refused while a record-changing job (a scan) runs. ``monitoring_enabled``
-takes effect at the next restart. ``POST /prompt_manager/settings/validate-source``
+takes effect at the next restart; ``save_foreign_runs`` (#247) at once, since
+the monitor reads it for every file. ``POST /prompt_manager/settings/validate-source``
 checks one path and writes nothing.
 """
 
@@ -43,6 +44,7 @@ _SETTINGS_FILE_KEYS = {
     "INFINITE_SCROLL": ("web_ui", "infinite_scroll"),
     "WORKER_THREADS": ("performance", "worker_threads"),
     "MONITORING_ENABLED": ("gallery", "monitoring", "enabled"),
+    "SAVE_FOREIGN_RUNS": ("gallery", "monitoring", "save_foreign_runs"),
     "SAFETY_BACKUP": ("maintenance", "safety_backup"),
     "THUMBNAIL_FORMAT": ("thumbnails", "format"),
     "THUMBNAIL_QUALITY": ("thumbnails", "quality"),
@@ -58,7 +60,7 @@ _SETTINGS_FILE_KEYS = {
 # 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.
-_GALLERY_SETTINGS = frozenset({"MONITORING_ENABLED"})
+_GALLERY_SETTINGS = frozenset({"MONITORING_ENABLED", "SAVE_FOREIGN_RUNS"})
 # Attributes that live on AutotagConfig, named without the AUTOTAG_ prefix.
 _AUTOTAG_SETTINGS = frozenset(
     name for name in _SETTINGS_FILE_KEYS if name.startswith("AUTOTAG_")
@@ -264,6 +266,7 @@ class SettingsRoutesMixin:
                     "thumbnail_encoders": encoders,
                     "settings": {
                         "monitoring_enabled": GalleryConfig.MONITORING_ENABLED,
+                        "save_foreign_runs": GalleryConfig.SAVE_FOREIGN_RUNS,
                         "result_timeout": PromptManagerConfig.RESULT_TIMEOUT,
                         "webui_display_mode": PromptManagerConfig.WEBUI_DISPLAY_MODE,
                         "infinite_scroll": PromptManagerConfig.INFINITE_SCROLL,
@@ -626,6 +629,10 @@ class SettingsRoutesMixin:
             if not isinstance(data["monitoring_enabled"], bool):
                 raise ValueError("monitoring_enabled must be true or false")
             updates["MONITORING_ENABLED"] = data["monitoring_enabled"]
+        if "save_foreign_runs" in data:
+            if not isinstance(data["save_foreign_runs"], bool):
+                raise ValueError("save_foreign_runs must be true or false")
+            updates["SAVE_FOREIGN_RUNS"] = data["save_foreign_runs"]
         if "safety_backup" in data:
             if not isinstance(data["safety_backup"], bool):
                 raise ValueError("safety_backup must be true or false")

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

@@ -67,6 +67,7 @@ from aiohttp import web
 
 from . import thumbnail_settings
 from .images import (
+    in_trash_below_roots,
     MEDIA_EXTENSIONS,
     THUMBNAIL_CLEAR_TITLE,
     VIDEO_EXTENSIONS,
@@ -224,6 +225,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 +251,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):
@@ -835,7 +840,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/vocabulary.py

@@ -61,18 +61,25 @@ class VocabularyRoutesMixin:
         )
 
     async def _vocabulary_page(
-        self, request: web.Request, kind: str, reader_name: str, name: bool
+        self,
+        request: web.Request,
+        kind: str,
+        reader_name: str,
+        name: bool,
+        options: dict | None = None,
     ) -> web.Response:
         limit = _vocabulary_limit(request)
         cursor = request.query.get("cursor")
+        options = options or {}
 
         def read() -> dict:
             # The snapshot stays open while the cursor is decoded and encoded,
-            # so its epoch scope is the epoch the page was read from.
+            # so its epoch scope is the epoch the page was read from. A tag
+            # search or sort joins the scope: its cursor fits no other order.
             with self.db.library_snapshot() as (conn, epoch, _ceiling):
-                scope = _scope({"kind": kind, "epoch": epoch})
+                scope = _scope({"kind": kind, "epoch": epoch, **options})
                 result = getattr(self.db, reader_name)(
-                    conn, epoch, limit=limit, after=_decode(cursor, scope)
+                    conn, epoch, limit=limit, after=_decode(cursor, scope), **options
                 )
                 result["next_cursor"] = _encode(
                     result.pop("next_key"), scope, name=name
@@ -84,8 +91,14 @@ class VocabularyRoutesMixin:
 
     @pair_response
     async def list_vocabulary_tags(self, request: web.Request) -> web.Response:
+        options = {
+            "search": request.query.get("search", "").strip(),
+            "sort": request.query.get("sort", "created"),
+        }
+        # Name keys are text: kept server side so the cursor stays small.
+        name = options["sort"].startswith("name")
         return await self._vocabulary_page(
-            request, "vocabulary-tags", "_vocabulary_tags", False
+            request, "vocabulary-tags", "_vocabulary_tags", name, options
         )
 
     @pair_response

py/config.py

@@ -194,6 +194,8 @@ class GalleryConfig:
 
     Attributes:
         MONITORING_ENABLED (bool): Enable/disable automatic image monitoring
+        SAVE_FOREIGN_RUNS (bool): The monitor stores the certain prompt of an
+            output whose workflow has no PromptManager node (#247)
         MONITORING_DIRECTORIES (List[str]): Directories to monitor for new images
         SUPPORTED_EXTENSIONS (List[str]): Image file extensions to process
         PROCESSING_DELAY (float): Delay in seconds before processing new files
@@ -210,6 +212,8 @@ class GalleryConfig:
 
     # Image monitoring settings
     MONITORING_ENABLED = True
+    # Read by the monitor for every file, so a save applies without a restart
+    SAVE_FOREIGN_RUNS = True
     MONITORING_DIRECTORIES = []  # Auto-detect if empty
     SUPPORTED_EXTENSIONS = [".png", ".jpg", ".jpeg", ".webp", ".gif"]
     PROCESSING_DELAY = 2.0  # Seconds to wait before processing new files
@@ -251,6 +255,7 @@ class GalleryConfig:
         return {
             "monitoring": {
                 "enabled": cls.MONITORING_ENABLED,
+                "save_foreign_runs": cls.SAVE_FOREIGN_RUNS,
                 "directories": cls.MONITORING_DIRECTORIES,
                 "extensions": cls.SUPPORTED_EXTENSIONS,
                 "processing_delay": cls.PROCESSING_DELAY,
@@ -303,6 +308,14 @@ class GalleryConfig:
                 config_logger.warning(
                     "Ignoring gallery monitoring enabled: expected true or false"
                 )
+        if "save_foreign_runs" in monitoring:
+            if isinstance(monitoring["save_foreign_runs"], bool):
+                cls.SAVE_FOREIGN_RUNS = monitoring["save_foreign_runs"]
+            else:
+                config_logger.warning(
+                    "Ignoring gallery monitoring save_foreign_runs: expected true "
+                    "or false"
+                )
         if "directories" in monitoring:
             directories = cls._validated_directories(monitoring["directories"])
             if directories is not None:

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.0"
 license = {file = "LICENSE"}
 requires-python = ">=3.12"
 dependencies = ["watchdog>=2.1.0", "Pillow>=8.0.0"]

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). ----

tests/browser/fixtures.mjs

@@ -4,13 +4,32 @@
 // points at that worker's server. A server that dies fails the running test
 // with its stderr tail, and the next test in the worker gets a fresh server.
 // A test that fails gets the stderr its server wrote during it attached as
-// `server-stderr`.
+// `server-stderr`. The worker also records its browser's own log (#513): a
+// browser that exits abnormally prints one `pm-browser-exit|` line, and a test
+// that fails on a closed or crashed target gets that log attached as
+// `browser-stderr`.
+// The one-time older-images notice (#527) shows above every workspace page and as
+// a modal on the admin page until dismissed, so every context starts with it
+// dismissed. A spec about the notice sets `test.use({ olderImagesNotice: true })`.
 import { test as base, expect } from '../../tools/vendor/node_modules/@playwright/test/index.mjs';
 import { attachServerStderr, createServerSupervisor, startBrowserServer } from '../../tools/vendor/start-browser-server.mjs';
+import { attachBrowserLogs, createBrowserLogRecorder } from '../../tools/vendor/browser-logs.mjs';
 
 const SERVER_FIXTURE_TIMEOUT_MS = 90_000;
 
 export const test = base.extend({
+    browserLogs: [async ({}, use) => {
+        const recorder = createBrowserLogRecorder();
+        try {
+            await use(recorder);
+        } finally {
+            recorder.uninstall();
+        }
+    }, { scope: 'worker', auto: true }],
+    browserLogsOnFailure: [async ({ browserLogs }, use, testInfo) => {
+        await use();
+        await attachBrowserLogs(testInfo, browserLogs);
+    }, { auto: true }],
     testServer: [async ({}, use) => {
         const supervisor = createServerSupervisor({ start: () => startBrowserServer() });
         try {
@@ -28,6 +47,11 @@ export const test = base.extend({
         if (crash) throw crash;
     }, { auto: true, timeout: SERVER_FIXTURE_TIMEOUT_MS }],
     baseURL: async ({ workerServer }, use) => use(workerServer.url),
+    olderImagesNotice: [false, { option: true }],
+    storageState: async ({ baseURL, olderImagesNotice }, use) => use(olderImagesNotice ? undefined : {
+        cookies: [],
+        origins: [{ origin: new URL(baseURL).origin, localStorage: [{ name: 'pm_update_notice_dismissed', value: '3.2.0' }] }],
+    }),
 });
 
 export { expect };

tests/browser/focused-node.spec.mjs

@@ -485,7 +485,7 @@ test(`${variant} respects the existing Web UI display preference`,async({page})=
     });
     await page.evaluate(()=>node._focusedSettingsReady);
     await page.getByRole('button',{name:'Web UI ↗',exact:true}).click();
-    expect(await page.evaluate(()=>window.openArgs)).toEqual(['/prompt_manager/web','_blank']);
+    expect(await page.evaluate(()=>window.openArgs)).toEqual(['/prompt_manager/workspace#/library','_blank']);
 });
 
 test(`${variant} reports blocked Web UI popups without leaving the workflow`,async({page})=>{

tests/browser/full-inspector.spec.mjs

@@ -149,8 +149,8 @@ test('panel matches the concept order and copy',async({page,request})=>{
     await expect(panel.locator('.meta-grid dt')).toHaveText(['Seed','Steps','CFG','Sampler','Scheduler','Model']);
     await expect(panel.locator('.meta-grid dd')).toHaveText(['9007199254740993','28','6.5','dpmpp_2m','karras','fixture-model.safetensors']);
     const footer=panel.locator('.detail-footer');
-    // #331 D139: every listed file's footer ends with Collections… (folder icon).
-    await expect(footer.locator('button')).toHaveText(['Send to ComfyUI','Prompt history','Collections…']);
+    // #331 D139: every listed file's footer has Collections… (folder icon); #480 D159: Delete… ends it.
+    await expect(footer.locator('button')).toHaveText(['Send to ComfyUI','Prompt history','Collections…','Delete…']);
     await expect(footer.locator('button.send')).toHaveAccessibleName(`Send prompt pair to ComfyUI: ${POSITIVE_TITLE}`);
     await expect(footer.locator('button.send svg.icon')).toHaveCount(1);await expect(footer.locator('button:not(.send) svg.icon')).toHaveCount(2);
     await expect(panel).not.toContainText(/prompt evidence|Positive prompt tags/);
@@ -162,10 +162,10 @@ test('panel matches the concept order and copy',async({page,request})=>{
     await selectFile(page,'ambiguous.png');await expect(panel.locator('.prompt-section.positive .prompt-text')).toHaveText('embedded positive');
     await expect(panel.locator('span.chip.blue')).toHaveText('Embedded parameters');
     await expect(panel.locator('.panel-heading .chip')).toHaveText('File extension PNG');await expect(panel.locator('.panel-heading .chip .ca-visually-hidden')).toHaveText('File extension ');
-    await expect(panel.locator('.detail-footer button')).toHaveText(['Send to ComfyUI','Collections…']);
-    // Unlinked: no badge, category, tags, pair or history; the footer holds only Collections… (D139).
+    await expect(panel.locator('.detail-footer button')).toHaveText(['Send to ComfyUI','Collections…','Delete…']);
+    // Unlinked: no badge, category, tags, pair or history; the footer holds Collections… (D139) and Delete… (D159).
     reads.patch=null;await selectFile(page,'empty.png');await expect(panel.locator('.panel-heading h2')).toHaveText('empty.png');
-    await expect(panel.locator('.detail-footer button')).toHaveText(['Collections…']);
+    await expect(panel.locator('.detail-footer button')).toHaveText(['Collections…','Delete…']);
     expect(await panelStructure(panel)).toEqual({outer:['img.detail-cover','div.detail-inner','div.detail-footer'],
         inner:['div.panel-heading','p.small','section.prompt-section.positive','section.prompt-section.negative','dl.meta-grid']});
     await expect(panel.locator('.meta-grid dd')).toHaveText(Array(6).fill('Not recorded'));
@@ -266,7 +266,7 @@ test('pair send and history appear only with real roles and link',async({page,re
     await expect(panel.getByRole('button',{name:'Send prompt pair to ComfyUI: explicit-empty.png',exact:true})).toBeVisible();
     await expect(panel.getByRole('button',{name:'Prompt history',exact:true})).toHaveCount(0);
     await selectFile(page,'empty.png');await expect(panel.locator('.panel-heading h2')).toHaveText('empty.png');
-    await expect(panel.locator('.detail-footer button')).toHaveText(['Collections…']);await expect(panel.getByRole('button',{name:/^Send prompt pair/})).toHaveCount(0);
+    await expect(panel.locator('.detail-footer button')).toHaveText(['Collections…','Delete…']);await expect(panel.getByRole('button',{name:/^Send prompt pair/})).toHaveCount(0);
 });
 
 test('no provenance, raw or download controls in the side panel',async({page,request})=>{
@@ -362,10 +362,14 @@ test('Show in folder asks the server to open the listed file and announces each
     const body=sent.postDataJSON();
     expect(Object.keys(body).sort()).toEqual(['path','root_index','root_scope']);
     expect(body).toMatchObject({root_index:0,path:'inspector/complete.png'});expect(body.root_scope).toMatch(/^[0-9a-f]{64}$/);
-    await expect(toast).toHaveText('Opened the folder on this computer.');
+    await expect(toast).toHaveText('Opened the folder on the computer running ComfyUI.');
+    // #533: the answer is really on screen and reachable above the modal viewer, not covered by it.
+    await expect(toast).toBeVisible();
+    expect(await toast.evaluate(node=>{const box=node.getBoundingClientRect();const top=document.elementFromPoint(box.x+box.width/2,box.y+box.height/2);return top===node;})).toBe(true);
+    await expect(page.locator('#viewer-dialog').getByRole('status').filter({hasText:'Opened the folder on the computer running ComfyUI.'})).toHaveCount(1);
     const opened=async()=>(await (await request.get('/__test__/reveal')).json()).opened;
     expect(await opened()).toEqual(['output/inspector/complete.png']);
-    for (const [outcome,message] of [['unavailable','No file manager is available on this computer.'],['fail','The folder could not be opened. Try again.']]) {
+    for (const [outcome,message] of [['unavailable','No file manager is available on the computer running ComfyUI.'],['fail','The folder could not be opened. Try again.']]) {
         await expect(await request.post('/__test__/reveal',{data:{outcome}})).toBeOK();
         await show.click();await expect(toast).toHaveText(message);
     }
@@ -373,11 +377,56 @@ test('Show in folder asks the server to open the listed file and announces each
     await page.route('**/prompt_manager/images/reveal',route=>route.fulfill({status:429,contentType:'application/json',body:JSON.stringify({success:false,code:'busy',error:'The folder is already opening.'})}));
     await show.click();await expect(toast).toHaveText('The folder is already opening.');
     await page.unroute('**/prompt_manager/images/reveal');
-    // A proxied request counts as remote, so the real route answers 403 not-local.
+    // A proxied click opens the folder: the proxy's headers, not the browser's address, name the page (#533).
     await expect(await request.post('/__test__/reveal',{data:{outcome:'opened'}})).toBeOK();
     await page.route('**/prompt_manager/images/reveal',route=>route.continue({headers:{...route.request().headers(),'x-forwarded-for':'203.0.113.7'}}));
-    await show.click();await expect(toast).toHaveText('Show in folder works only in a browser on the computer running ComfyUI.');
-    expect(await opened()).toEqual(['output/inspector/complete.png']);
+    await show.click();await expect(toast).toHaveText('Opened the folder on the computer running ComfyUI.');
+    expect(await opened()).toHaveLength(2);
+    await page.unroute('**/prompt_manager/images/reveal');
+    // A plain cross-site POST is refused, and the toast says so.
+    // Chromium keeps its own Origin on a routed fetch, so the real route is asked directly and the toast gets its answer.
+    const refused=await request.post('/prompt_manager/images/reveal',{data:body,headers:{origin:'https://evil.example'}});
+    expect(refused.status()).toBe(403);expect((await refused.json()).code).toBe('not-same-origin');
+    await page.route('**/prompt_manager/images/reveal',route=>route.fulfill({status:403,contentType:'application/json',body:JSON.stringify({success:false,code:'not-same-origin',error:"This page's address does not match the server's."})}));
+    await show.click();await expect(toast).toHaveText("The folder was not opened: this page's address does not match the server's.");
+    await expect(toast).toBeVisible();
+    expect(await opened()).toHaveLength(2);
+    // #533: closing the viewer takes the toast back to the page, still showing the message.
+    await page.keyboard.press('Escape');await expect(page.locator('#viewer-dialog')).toHaveCount(0);
+    await expect(toast).toBeVisible();await expect(page.locator('body > #workspace-toast')).toHaveCount(1);
+});
+// #533 D176: real modals in the real top layer, with the page's own toast module.
+test('a toast stays on screen across nested, switched and closing modals',async({page,request})=>{
+    await fullFixture(page,request);
+    await page.evaluate(async()=>{
+        const url=performance.getEntriesByType('resource').map(entry=>entry.name).find(name=>name.includes('/toast.mjs'));
+        const {showWorkspaceToast}=await import(url);
+        window.say=showWorkspaceToast;
+        window.modal=(id,first=false)=>{const dialog=document.createElement('dialog');dialog.id=id;dialog.textContent=id;document.body[first?'prepend':'append'](dialog);dialog.showModal();return dialog;};
+        window.onTop=()=>{const node=document.getElementById('workspace-toast');const box=node.getBoundingClientRect();return box.width>0&&document.elementFromPoint(box.x+box.width/2,box.y+box.height/2)===node;};
+    });
+    const parent=()=>page.evaluate(()=>{const host=document.getElementById('workspace-toast')?.parentElement;return host?host.id||host.localName:null;});
+    const onTop=()=>page.evaluate(()=>window.onTop());
+    await page.evaluate(()=>{window.modal('a');say('one');});
+    expect(await parent()).toBe('a');expect(await onTop()).toBe(true);
+    // Nested: the top one holds it, even earlier in the page than the one under it.
+    await page.evaluate(()=>{window.modal('b',true);say('two');});
+    expect(await parent()).toBe('b');expect(await onTop()).toBe(true);
+    // The top one closes while the toast shows: the message stays, in the modal now on top.
+    await page.evaluate(()=>document.getElementById('b').close());
+    await expect.poll(parent).toBe('a');expect(await onTop()).toBe(true);
+    await expect(page.locator('#workspace-toast')).toHaveText('two');
+    // Switched: the next modal takes the next toast.
+    await page.evaluate(()=>{document.getElementById('a').close();window.modal('c');say('three');});
+    expect(await parent()).toBe('c');expect(await onTop()).toBe(true);
+    // Closed and removed at once, as the workspace's dialogs do: home, still showing.
+    await page.evaluate(()=>{const c=document.getElementById('c');c.close();c.remove();});
+    await expect.poll(parent).toBe('body');expect(await onTop()).toBe(true);
+    await expect(page.locator('#workspace-toast')).toHaveText('three');
+    // Removed while open, with no close event: the next message still shows.
+    await page.evaluate(()=>{window.modal('d');say('four');document.getElementById('d').remove();say('five');});
+    expect(await parent()).toBe('body');expect(await onTop()).toBe(true);
+    await expect(page.locator('#workspace-toast')).toHaveText('five');
 });
 test('full inspector shows six facts and scoped provenance',async({page,request},testInfo)=>{
     await page.setViewportSize({width:1600,height:1100});await fullFixture(page,request);const panel=inspector(page);
@@ -415,7 +464,7 @@ test('full inspector preserves long role text and exact sends',async({page,reque
         await fullFixture(page,request);const panel=inspector(page);const positive=panel.locator('.prompt-section.positive .prompt-text');await expect(positive).toContainText('猫 😀');const text=await positive.textContent();expect(text.length).toBeGreaterThan(1000);
         await panel.locator('.detail-footer button.send').click();const sender=page.getByRole('dialog',{name:'Send to ComfyUI',exact:true});
         await expect(sender.getByRole('combobox',{name:'ComfyUI window',exact:true}).locator('option')).toContainText(['Exact host']);
-        await sender.getByRole('combobox',{name:'ComfyUI window',exact:true}).selectOption({label:'Exact host'});await sender.getByRole('combobox',{name:'Destination node',exact:true}).selectOption({label:'Exact pair'});
+        await sender.getByRole('combobox',{name:'ComfyUI window',exact:true}).selectOption({label:'Exact host'});await sender.getByRole('combobox',{name:'Destination node',exact:true}).selectOption({label:'#pair Exact pair'});
         await sender.getByRole('button',{name:'Preview changes',exact:true}).click();await sender.getByRole('button',{name:'Apply to ComfyUI',exact:true}).click();await expect(sender).toContainText('Sent to ComfyUI.');
         expect(await host.evaluate(()=>window.sent)).toEqual([{positive:text,negative:'blurred edges, watermark'}]);
     } finally {await host.evaluate(()=>window.bridge.dispose());await host.close();}
@@ -558,7 +607,8 @@ test.describe('full inspector coarse pointer',()=>{
         }
         await openSelected(page);const viewer=page.locator('dialog#viewer-dialog');
         await expect(viewer.locator('.viewer-canvas img')).toBeVisible();
-        for(const target of await viewer.locator('#viewer-toolbar > :is(button,a),.dialog-header > button').all()) {
+        // #450 D156: the view switch is hidden at 700px and below; a shown one is measured like ×.
+        for(const target of await viewer.locator('#viewer-toolbar > :is(button,a),.dialog-header > button:visible').all()) {
             const box=await target.evaluate(node=>{const {width,height}=node.getBoundingClientRect();return {width,height,name:node.getAttribute('aria-label')};});
             expect(box.width,box.name).toBeGreaterThanOrEqual(80);expect(box.height,box.name).toBeGreaterThanOrEqual(80);
         }
@@ -621,6 +671,19 @@ test('viewer matches the concept shell and toolbar order',async({page,request})=
         const production=await page.evaluate(VIEWER_SHELL);
         // C8 rebuilds the details content; until then the shared file inspector marks the aside it renders into.
         production.paths=production.paths.map(path=>path.replace('aside#viewer-details.file-inspector','aside#viewer-details'));
+        // #450 D156: production's view switch sits right before × (hidden at 700px and below); the concept has none.
+        const modePaths=production.paths.filter(path=>path.includes('button#viewer-mode'));
+        expect(modePaths.map(path=>path.split(' > ').slice(2).join(' > '))).toEqual(['button#viewer-mode.icon-btn','button#viewer-mode.icon-btn > svg.icon','button#viewer-mode.icon-btn > svg.icon > path']);
+        expect(production.paths[production.paths.indexOf(modePaths[2])+1]).toBe('dialog#viewer-dialog > div.dialog-header > button#viewer-close');
+        production.paths=production.paths.filter(path=>!modePaths.includes(path));
+        // #532 D170: production's shortcuts button sits right before the view switch; the concept has none.
+        const keysPaths=production.paths.filter(path=>path.includes('button#viewer-keys'));
+        expect(keysPaths.map(path=>path.split(' > ').slice(2).join(' > '))).toEqual(['button#viewer-keys.icon-btn','button#viewer-keys.icon-btn > svg.icon','button#viewer-keys.icon-btn > svg.icon > path']);
+        production.paths=production.paths.filter(path=>!keysPaths.includes(path));
+        // #480 D159: the Images viewer's Delete… (`danger`) follows Download; the concept has none.
+        expect(production.toolbar[8].text).toBe('Delete…');
+        production.toolbar.splice(8,1);
+        production.paths=production.paths.filter(path=>!path.includes('button.danger'));
         // #249 D34: after Next → production appends its own status lines (position, then the transform and slideshow
         // reasons), and Download stays the original-file link; every other control keeps the concept's text and order.
         const appended=production.toolbar.slice(9);

tests/browser/history-canonical-comparison.mjs

@@ -107,7 +107,7 @@ export async function exerciseCurrentPairSend(page, roles) {
         const send = viewer.getByRole('button', { name: /^Send prompt pair to ComfyUI: / });await send.click();
         const sender = page.getByRole('dialog', { name: 'Send to ComfyUI', exact: true });await expect(sender).toBeVisible();
         await sender.getByRole('combobox', { name: 'ComfyUI window', exact: true }).selectOption({ label: 'Canonical ComfyUI' });
-        await sender.getByRole('combobox', { name: 'Destination node', exact: true }).selectOption({ label: 'Canonical destination' });
+        await sender.getByRole('combobox', { name: 'Destination node', exact: true }).selectOption({ label: '#canonical-pair Canonical destination' });
         await sender.getByRole('button', { name: 'Preview changes', exact: true }).click();
         await expect(sender).toContainText(roles.positive);await expect(sender).toContainText(roles.negative);
         await sender.getByRole('button', { name: 'Apply to ComfyUI', exact: true }).click();await expect(sender).toContainText('Sent to ComfyUI.');

tests/browser/history-concept.spec.mjs

@@ -26,7 +26,8 @@ const SECTIONS=['heading','history','noPageScroll'];
  */
 export function mapHistoryConcept(reference) {
     const {history}=reference.oracle;
-    return {...reference,history:{...history,
+    // D163 (#512): the heading as laid out with the More menu button (the oracle's heading).
+    return {...reference,heading:reference.oracle.heading,history:{...history,
         // D24: "N tracked runs" is the recorded window's `recorded_run_count`; the fixture records one run per image, so
         // the concept's "4 tracked runs" holds by construction (asserted on both documents below).
         summary:history.summary,
@@ -120,11 +121,14 @@ async function conceptAsRecorded(page) {
         const add=[...document.querySelectorAll('.timeline-panel .row-actions > button')].find(node=>node.textContent==='+ Add negative');
         const archived=add.cloneNode(false);archived.removeAttribute('data-action');archived.removeAttribute('data-prompt');
         archived.setAttribute('aria-pressed','false');archived.textContent=label;add.after(archived);
-        window.__historyOracle={timeline,order:[...lanes].reverse(),texts,archived};
+        // D163 (#512): the heading ends with the "⋯" More menu button, drawn as the concept's Edit metadata button.
+        const edit=[...document.querySelectorAll('#main > .heading .heading-actions > button')].find(node=>node.textContent.trim()==='Edit metadata');
+        const moreButton=edit.cloneNode(false);moreButton.removeAttribute('data-action');moreButton.textContent='⋯';edit.parentElement.append(moreButton);
+        window.__historyOracle={timeline,order:[...lanes].reverse(),texts,archived,moreButton};
     },SHOW_ARCHIVED);
     await settle(page);
     return ()=>page.evaluate(()=>{
-        const {timeline,order,texts,archived}=window.__historyOracle;
+        const {timeline,order,texts,archived,moreButton}=window.__historyOracle;moreButton.remove();
         order.forEach((lane,index)=>{lane.style.setProperty('--lane-offset',`${38+index*47}px`);timeline.append(lane);});
         for(const [span,text] of texts)span.textContent=text;
         archived.remove();delete window.__historyOracle;
@@ -190,7 +194,7 @@ function productionSent(host) {
     return async page=>{
         const sender=page.getByRole('dialog',{name:'Send to ComfyUI',exact:true});
         await sender.getByRole('combobox',{name:'ComfyUI window',exact:true}).selectOption({label:'Concept ComfyUI'});
-        await sender.getByRole('combobox',{name:'Destination node',exact:true}).selectOption({label:'Concept destination'});
+        await sender.getByRole('combobox',{name:'Destination node',exact:true}).selectOption({label:'#pair-1 Concept destination'});
         await sender.getByRole('button',{name:'Preview changes',exact:true}).click();
         await expect.poll(()=>host.evaluate(()=>window.previewed.length)).toBe(1);
         const roles=await host.evaluate(()=>window.previewed.pop());

tests/browser/images-concept.spec.mjs

@@ -31,13 +31,15 @@ function captionDay(seconds,now=new Date()) {
 }
 
 /**
- * D8 (amended by #249 PR D and #331): the selection bar's real actions only; Remove links… is left out. Add to
- * collection compares in place (D138).
+ * D8 (amended by #249 PR D and #331): the selection bar's real actions only. Add to collection compares in place
+ * (D138). #480 D159: Remove links… returns as Delete…; OMITTED_BULK names the concept button it replaces.
  */
 const OMITTED_BULK=Object.freeze(['Remove links…']);
 const OMITTED_BULK_SELECTORS=Object.freeze(['bulk-remove'].map(action=>`#bulkbar [data-action="${action}"]`));
 /** D85: Auto-tag selected follows the concept's Add tags, a concept button of its own text. */
 const AUTO_TAG_CLONE=Object.freeze([{selector:'#bulkbar [data-action="bulk-tags"]',text:'Auto-tag selected'}]);
+/** #480 D159: Delete… takes the place of Remove links… (the concept's `danger` button, under another text). */
+const DELETE_CLONE=Object.freeze([{selector:'#bulkbar [data-action="bulk-remove"]',text:'Delete…'}]);
 
 /**
  * `withoutItems` for a row with padding and a border (the `.bulkbar`): its items flow again inside the content box,
@@ -452,10 +454,12 @@ for(const [width,height] of SIZES)for(const theme of THEMES)test(`images selecti
     const restore=await mutate(page,{remove:[...OMITTED_BULK_SELECTORS]});
     const removed=(await inventoryOf(page,'#main')).bulkbar;
     await restore();
-    // The oracle for production's bar: the concept page itself without the omitted actions and with D85's button.
-    const restoreProduction=await mutate(page,{remove:[...OMITTED_BULK_SELECTORS],clone:AUTO_TAG_CLONE});
+    // The oracle for production's bar: the concept page itself with D85's button and D159's Delete… in place of
+    // Remove links….
+    const restoreClones=await mutate(page,{clone:[...AUTO_TAG_CLONE,...DELETE_CLONE]});
+    const restoreRemoved=await mutate(page,{remove:[...OMITTED_BULK_SELECTORS]});
     const oracle=await inventoryOf(page,'#main');
-    await restoreProduction();
+    await restoreRemoved();await restoreClones();
     await captureCanonicalState(page,testInfo,`concept-images-selected-${width}-${theme}`);
     await openProductionLoaded(page,theme,'no-preference');
     const emptyActual=await inventoryOf(page,'#workspace-main');
@@ -468,9 +472,12 @@ for(const [width,height] of SIZES)for(const theme of THEMES)test(`images selecti
         'Add to collection','Export metadata','Remove links…','Clear']);
     expect.soft(diffInventory(stripExact(mapBulkbar(reference.bulkbar,0).bulkbar.bar),stripExact(removed.bar)),'bar without the omitted actions').toEqual([]);
     expect.soft(actual.bulkbar.bar.items.map(item=>item.text),'production bar').toEqual(['1 selected','Add tags','Auto-tag selected',
-        'Set category','Add to collection','Export metadata','Clear']);
+        'Set category','Add to collection','Export metadata','Delete…','Clear']);
     expect.soft(compare(empty,emptyActual,manifest,['bulkbar','cards','panel']),'nothing checked').toEqual([]);
-    expect.soft(compare(oracle,actual,manifest,['bulkbar','cards','panel']),'one card checked').toEqual([]);
+    // D186: the panel's sticky offset clears the shown bar; at one column the bar is one sideways row (its items are checked above).
+    const stuckPanel={...oracle,panel:{...oracle.panel,...(width<=700?{top:actual.panel.top}:{}),box:{...oracle.panel.box,top:actual.panel.box.top}}};
+    expect(parseFloat(actual.panel.box.top)).toBeGreaterThanOrEqual(20);
+    expect.soft(compare(stuckPanel,actual,manifest,width<=700?['panel']:['bulkbar','cards','panel']),'one card checked').toEqual([]);
     // The bar comparison notices a change to the production bar.
     await mutate(page,{css:'.bulkbar { padding: 4px }'});
     expect(compare(oracle,await inventoryOf(page,'#workspace-main'),manifest,['bulkbar']).map(item=>item.path)).toContain('bulkbar.bar.paddingTop');

tests/browser/images-delete-concept.spec.mjs

@@ -0,0 +1,106 @@
+import {test,expect} from './fixtures.mjs';
+import {SIZES,THEMES,seedConceptAt,openConcept,openProduction,diffInventory,mutate} from './concept-inventory.mjs';
+
+// #480 (D159): the Images selection bar's Delete… against the concept's "Remove links…" (app.js a432, `danger`) and
+// its review against the concept's `maintenancePlan` (a953–957): the bar button's place and style, the review's
+// `dl.meta-grid` rows and its actions, on the concept-shaped fixture, at 1600x1100 and 390x844 in both themes.
+// Paired screenshots are artifacts only; there are no PNG baselines.
+test.use({locale:'en-US',timezoneId:'UTC'});
+
+const CARDS='.gallery-body > .image-grid > .media-card';
+
+/** Runs in the page: the selection bar's danger button and where it sits among the bar's buttons. */
+function measureBar() {
+    const bar=document.querySelector('#bulkbar > .bulkbar');
+    const buttons=[...bar.querySelectorAll(':scope > button')];
+    const danger=bar.querySelector(':scope > button.danger');
+    const css=(node,...keys)=>Object.fromEntries(keys.map(key=>[key,getComputedStyle(node)[key]]));
+    return {
+        // The danger action sits right before Clear, the last button.
+        dangerBeforeClear:buttons.indexOf(danger)===buttons.length-2,
+        lastText:buttons.at(-1).textContent.trim(),
+        danger:css(danger,'color','backgroundColor','borderTopColor','fontSize','fontWeight','paddingTop','paddingLeft',
+            'borderTopLeftRadius','height'),
+    };
+}
+
+/** Runs in the page: the open review's grid and actions (the parts production shares with the concept). */
+function measureReview() {
+    const dialog=document.querySelector('dialog#modal[open]');
+    const css=(node,...keys)=>node?Object.fromEntries(keys.map(key=>[key,getComputedStyle(node)[key]])):null;
+    const grid=dialog.querySelector('#modal-body dl.meta-grid');
+    const entry=grid.querySelector(':scope > div');
+    const actions=dialog.querySelector('#modal-body > .dialog-actions');
+    const [cancel]=actions.children;
+    return {
+        grid:css(grid,'display','gridTemplateColumns','columnGap','rowGap'),
+        dt:css(entry.querySelector('dt'),'fontSize','color','textTransform','letterSpacing'),
+        dd:css(entry.querySelector('dd'),'fontSize','color','marginLeft'),
+        terms:[...grid.querySelectorAll('dt')].map(dt=>dt.textContent.trim()),
+        actions:css(actions,'display','justifyContent','columnGap','marginTop'),
+        cancel:{text:cancel.textContent.trim(),...css(cancel,'backgroundColor','color','fontSize','paddingTop','paddingLeft')},
+    };
+}
+
+async function checkFirst(page) {
+    await page.locator(CARDS).first().locator('.image-overlays > label > input[type="checkbox"]').check();
+}
+
+async function concept(page,theme) {
+    await openConcept(page,'images',theme);
+    await checkFirst(page);
+    const bar=await page.evaluate(measureBar);
+    await page.locator('#main #bulkbar').getByRole('button',{name:'Remove links…',exact:true}).click();
+    await expect(page.locator('dialog#modal[open] #modal-title')).toHaveText('Review maintenance plan');
+    await page.mouse.move(0,0);
+    return {bar,review:await page.evaluate(measureReview)};
+}
+
+async function production(page,theme) {
+    await openProduction(page,'images',theme);
+    await expect(page.locator(CARDS).first()).toBeVisible();
+    await checkFirst(page);
+    const bar=await page.evaluate(measureBar);
+    await page.locator('#workspace-main #bulkbar').getByRole('button',{name:'Delete…',exact:true}).click();
+    const dialog=page.locator('dialog#modal[open]');
+    await expect(dialog.locator('#modal-title')).toHaveText('Delete 1 image');
+    await dialog.getByRole('radio',{name:/^Remove from the library only/}).check();
+    await page.mouse.move(0,0);
+    return {bar,review:await page.evaluate(measureReview)};
+}
+
+/**
+ * D159: production's rows follow the chosen mode (Scope, then what the files lose, Backup and Media on disk; the
+ * concept's sample has Scope, Protected, Backup and Media on disk). The geometry, tokens and the Cancel button match.
+ */
+function mapConcept(reference,actual) {
+    return {...reference,review:{...reference.review,terms:actual.review.terms}};
+}
+
+test.beforeEach(async({page,request},testInfo)=>{testInfo.manifest=await seedConceptAt(page,request);});
+
+for(const theme of THEMES)for(const [width,height] of SIZES) {
+    test(`delete matches the concept's remove links ${theme} ${width}x${height}`,async({page},testInfo)=>{
+        await page.setViewportSize({width,height});
+        const reference=await concept(page,theme);
+        await page.screenshot({path:testInfo.outputPath(`concept-remove-links-${theme}-${width}.png`)});
+        const actual=await production(page,theme);
+        await page.screenshot({path:testInfo.outputPath(`production-delete-${theme}-${width}.png`)});
+        expect(reference.review.terms).toEqual(['Scope','Protected','Backup','Media on disk']);
+        expect(actual.review.terms[0]).toBe('Scope');
+        expect(actual.review.terms.slice(-2)).toEqual(['Backup','Media on disk']);
+        expect(diffInventory(mapConcept(reference,actual),actual)).toEqual([]);
+    });
+}
+
+test('delete comparison detects mutations',async({page})=>{
+    await page.setViewportSize({width:1600,height:1100});
+    const reference=await concept(page,'dark');
+    const actual=await production(page,'dark');
+    const expected=mapConcept(reference,actual);
+    expect(diffInventory(expected,actual)).toEqual([]);
+    const undo=await mutate(page,{css:'dialog#modal dl.meta-grid { column-gap: 31px; } dialog#modal .dialog-actions { gap: 29px; }'});
+    const paths=diffInventory(expected,{...actual,review:await page.evaluate(measureReview)}).map(item=>item.path);
+    expect(paths).toEqual(expect.arrayContaining(['review.grid.columnGap','review.actions.columnGap']));
+    await undo();
+});

tests/browser/images-removed-concept.spec.mjs

@@ -0,0 +1,65 @@
+import {test,expect} from './fixtures.mjs';
+import {seedConceptAt,openConcept,openProduction} from './concept-inventory.mjs';
+
+// #498 (D162): the concept has no Show removed images screen, so production builds it from the concept's own parts:
+// the heading action button (as the concept's Images heading button), the card's `.chip` (as a concept card chip) and
+// the selection bar's buttons (as the concept bar's plain buttons). Each is compared with its concept counterpart.
+test.use({locale:'en-US',timezoneId:'UTC'});
+
+const CARDS='.gallery-body > .image-grid > .media-card';
+const KEYS=['color','backgroundColor','borderTopColor','borderTopWidth','fontSize','fontWeight','paddingTop','paddingLeft',
+    'borderTopLeftRadius','height'];
+const CHIP_KEYS=['color','backgroundColor','borderTopColor','fontSize','fontWeight','paddingTop','paddingLeft',
+    'borderTopLeftRadius'];
+
+/** Runs in the page: the computed style of the first node `selector` matches. */
+function styleOf([selector,keys]) {
+    const node=document.querySelector(selector);
+    return node?Object.fromEntries(keys.map(key=>[key,getComputedStyle(node)[key]])):null;
+}
+
+async function measure(page,selectors) {
+    const found={};
+    for (const [name,selector,keys] of selectors) found[name]=await page.evaluate(styleOf,[selector,keys]);
+    return found;
+}
+
+test.beforeEach(async({page,request})=>{await seedConceptAt(page,request);});
+
+for (const theme of ['dark','light']) {
+    test(`Show removed images uses the concept's heading button, chip and bar buttons (${theme})`,async({page})=>{
+        await page.setViewportSize({width:1600,height:1100});
+        await openConcept(page,'images',theme);
+        await page.locator(CARDS).first().locator('.image-overlays > label > input[type="checkbox"]').check();
+        await page.mouse.move(0,0);
+        const reference=await measure(page,[
+            ['heading','#main .heading-actions > button:not(.ca-primary)',KEYS],
+            ['chip','#main .chip',CHIP_KEYS],
+            ['bar','#main #bulkbar > .bulkbar > button:not(.danger)',KEYS],
+        ]);
+
+        await openProduction(page,'images',theme);
+        await expect(page.locator(CARDS).first()).toBeVisible();
+        await page.locator(CARDS).first().locator('.image-overlays > label > input[type="checkbox"]').check();
+        await page.locator('#workspace-main #bulkbar').getByRole('button',{name:'Delete…',exact:true}).click();
+        const dialog=page.locator('dialog#modal[open]');
+        await dialog.getByRole('radio',{name:/^Remove from the library only/}).check();
+        await dialog.locator('.dialog-actions button.danger').click();
+        const toggle=page.locator('.heading-actions').getByRole('button',{name:/^Show removed images \(1\)$/});
+        await expect(toggle).toBeVisible();
+        await page.mouse.move(0,0);
+        const heading=await measure(page,[['heading','#workspace-main .heading-actions > button[aria-pressed]',KEYS]]);
+        await toggle.click();
+        await expect(page.locator(`${CARDS} .chip`,{hasText:'Removed'})).toBeVisible();
+        await page.locator(CARDS).first().locator('.image-overlays > label > input[type="checkbox"]').check();
+        await expect(page.locator('#workspace-main #bulkbar').getByRole('button',{name:'Return to library'})).toBeVisible();
+        await page.mouse.move(0,0);
+        const actual={...heading,...await measure(page,[
+            ['chip',`${CARDS} .media-info .chip`,CHIP_KEYS],
+            ['bar','#workspace-main #bulkbar > .bulkbar > button',KEYS],
+        ])};
+        // Every part was found on both sides: two missing nodes must never compare equal.
+        expect(Object.values(reference).every(Boolean)).toBe(true);
+        expect(actual).toEqual(reference);
+    });
+}

tests/browser/inspector-concept.spec.mjs

@@ -56,7 +56,8 @@ function withCategoryScope(inventory) {
  * equal production. `reference.viewerPlain` is the concept's own full details
  * with its tag chips made plain (D20), D117's summary button added and, under a coarse pointer, D34's targets, all at
  * runtime: the browser lays out what follows them, as the Images spec's removed toolbar items do. D139's
- * "Collections…" is drawn after Prompt history in both the concept's panel footer and its full details.
+ * "Collections…" is drawn after Prompt history in both the concept's panel footer and its full details; #480's
+ * D159 "Delete…" follows it in the panel footer.
  */
 export function mapInspectorConcept(reference,{viewport}) {
     const {panel,viewerPlain,fileControls}=reference;
@@ -125,6 +126,13 @@ async function withCollectionsButton(page,selector) {
         for(const key of Object.keys(button.dataset))delete button.dataset[key];
         button.append(document.querySelector('.filter-row .chip > svg.icon').cloneNode(true),' Collections…');
         history.after(button);(window.__collectionsButtons??=[]).push(button);
+        // #480 D159: Delete… ends the panel's actions, drawn as the concept's own `danger` button (its Remove links…).
+        if(selector.startsWith('#image-inspector')) {
+            const remove=history.cloneNode(false);
+            for(const key of Object.keys(remove.dataset))delete remove.dataset[key];
+            remove.className='danger';remove.textContent='Delete…';
+            button.after(remove);window.__collectionsButtons.push(remove);
+        }
     },selector);
     await settle(page);
     return async()=>{await page.evaluate(()=>{for(const button of window.__collectionsButtons??[])button.remove();delete window.__collectionsButtons;});};
@@ -251,7 +259,7 @@ async function productionSendOf(page,context) {
         const sender=page.getByRole('dialog',{name:'Send to ComfyUI',exact:true});await expect(sender).toBeVisible();
         const windows=sender.getByRole('combobox',{name:'ComfyUI window',exact:true});
         await expect(windows.locator('option')).toContainText(['Test ComfyUI']);await windows.selectOption({label:'Test ComfyUI'});
-        await sender.getByRole('combobox',{name:'Destination node',exact:true}).selectOption({label:'My prompt pair'});
+        await sender.getByRole('combobox',{name:'Destination node',exact:true}).selectOption({label:'#pair-1 My prompt pair'});
         await sender.getByRole('button',{name:'Preview changes',exact:true}).click();
         await sender.getByRole('button',{name:'Apply to ComfyUI',exact:true}).click();
         await expect(sender).toContainText('Sent to ComfyUI.');
@@ -373,7 +381,7 @@ test('a recipe-linked image without recipe metadata shows its positive prompt me
     await expect(tags.locator('.role-label')).toHaveText('Prompt tags · 3');
     await expect(tags.locator('.tags > button.chip')).toHaveText(['minimal','warm_light','editorial']);
     await expect(tags.locator('span.chip')).toHaveCount(0);
-    await expect(panelOf(page).locator('.detail-footer button')).toHaveText(['Send to ComfyUI','Prompt history','Collections…']);
+    await expect(panelOf(page).locator('.detail-footer button')).toHaveText(['Send to ComfyUI','Prompt history','Collections…','Delete…']);
 });
 
 // #391: a tag chip replaces the collection; until its listing answers the caption says it is loading, never that the

tests/browser/job-end.mjs

@@ -0,0 +1,13 @@
+// A job's end toast follows the job, which has no time limit: under load it outlasts the 5 s expect default (#533, as
+// #520). Do `action`, whose POST to `url` answers with the job it started, then wait for that job's event stream to end.
+export const APPLY_URL = '/prompt_manager/maintenance/apply';
+export const RETURN_URL = '/prompt_manager/images/removed/return';
+
+export async function startAndAwaitJob(page, action, url = APPLY_URL) {
+    const applied = page.waitForResponse(response => response.url().endsWith(url));
+    await action();
+    const { job } = await (await applied).json();
+    const events = await page.request.get(`/prompt_manager/jobs/${job.id}/events`);
+    if (!events.ok()) throw new Error(`Job ${job.id} events answered ${events.status()}`);
+    return job;
+}

tests/browser/library-concept.spec.mjs

@@ -44,7 +44,12 @@ export function mapLibraryConcept(reference) {
         // "4 runs" and "Used today". The two-line clamp is
         // applied below through `clampedCards`, measured on the concept with `CLAMP_CSS` (its samples wrap at 390px).
         footer:card.footer}));
-    return {...reference,library:{...library,toolbar,
+    // D152 (#443): Show archived prompts (N) comes first in the heading actions, a plain button like Export (the concept
+    // fixture has nothing archived, so N is 0).
+    const actions=reference.heading.actions;
+    const plain=actions.buttons.find(button=>!button.primary);
+    const heading={...reference.heading,actions:{...actions,buttons:[{...plain,text:'Show archived prompts (0)'},...actions.buttons]}};
+    return {...reference,heading,library:{...library,toolbar,
         // D35: the caption uses the singular for one.
         caption:{...library.caption,top:library.caption.top-rise,spans:library.caption.spans.map(span=>({...span,text:countCopy(span.text)}))},
         ...(({grid,cards:clamped})=>({grid:{...grid,top:grid.top-rise},cards:clamped}))(clampedCards({grid:library.grid,cards})),
@@ -228,7 +233,7 @@ function productionSent(host) {
     return async page=>{
         const sender=page.getByRole('dialog',{name:'Send to ComfyUI',exact:true});
         await sender.getByRole('combobox',{name:'ComfyUI window',exact:true}).selectOption({label:'Concept ComfyUI'});
-        await sender.getByRole('combobox',{name:'Destination node',exact:true}).selectOption({label:'Concept destination'});
+        await sender.getByRole('combobox',{name:'Destination node',exact:true}).selectOption({label:'#pair-1 Concept destination'});
         await sender.getByRole('button',{name:'Preview changes',exact:true}).click();
         await expect.poll(()=>host.evaluate(()=>window.previewed.length)).toBe(1);
         const roles=await host.evaluate(()=>window.previewed.pop());
@@ -369,7 +374,9 @@ for(const [width,height] of SIZES)for(const theme of THEMES)test(`the Library se
     await captureCanonicalState(page,testInfo,`production-library-selection-${width}-${theme}`);
     await testInfo.attach('library-selection',{body:JSON.stringify({reference,actual},null,2),contentType:'application/json'});
     expect(reference.bar.buttons.map(button=>button.text),'the concept bar has every Library action').toEqual(['Add tags','Set category','Clear']);
-    expect(actual).toEqual(reference);
+    // D186: at one column the bar is one row that scrolls sideways, with the same buttons.
+    const mapped=width<=700?{...reference,bar:{...reference.bar,flexWrap:'nowrap',buttons:reference.bar.buttons.map((button,index)=>({...button,left:actual.bar.buttons[index].left,top:actual.bar.buttons[index].top}))}}:reference;
+    expect(actual).toEqual(mapped);
     // The comparison notices a moved checkbox (spec §7 mutation check).
     await mutate(page,{css:'.prompt-card .image-overlays { inset: 20px 9px auto }'});
     expect((await measureSelection(page,{card:CARDS,main:'#workspace-main'})).overlay.top).not.toBe(reference.overlay.top);

tests/browser/library-delete-concept.spec.mjs

@@ -0,0 +1,148 @@
+import { startCoverage, saveCoverage } from '../../tools/vendor/browser-coverage.mjs';
+import { test, expect } from './fixtures.mjs';
+import { SIZES, THEMES, seedConcept, openConcept, openProduction, diffInventory, mutate } from './concept-inventory.mjs';
+
+// #443 D152: the Library's Delete… review against the concept's own reviewed archive: the recipe editor's "Archive…"
+// (app.js 976), which opens `maintenancePlan("archive <title>")` (953–958, 1309–1310). The concept's shell, grid tokens,
+// `p.small` and actions are kept; its title, demo note, rows, note text and `ca-primary` start are replaced by D152's,
+// and the start takes the concept's `danger` class, measured on the editor's own "Archive…". The bar's Delete… is
+// compared with the concept Images bar's danger action, "Remove links…" (app.js 432).
+test.use({ locale: 'en-US', timezoneId: 'UTC' });
+
+test.beforeEach(async ({ page, request, browserName }) => {
+    await startCoverage(page, browserName);
+    await seedConcept(request);
+    page.conceptErrors = [];
+    page.on('pageerror', error => page.conceptErrors.push(error.message));
+});
+test.afterEach(async ({ page, browserName }) => {
+    await saveCoverage(page, browserName);
+    expect(page.conceptErrors).toEqual([]);
+});
+
+/** Runs in the page: the open review dialog's title, body tree, grid, notes and actions with the concept's tokens. */
+function measureReview() {
+    const dialog = document.querySelector('dialog#modal[open]');
+    const css = (node, ...keys) => node ? Object.fromEntries(keys.map(key => [key, getComputedStyle(node)[key]])) : null;
+    const name = node => [node.localName, ...[...node.classList].sort()].join('.');
+    const text = node => node?.textContent.trim() ?? null;
+    const body = dialog.querySelector('#modal-body');
+    const grid = body.querySelector(':scope > dl.meta-grid');
+    const notes = [...body.querySelectorAll(':scope > p.small')];
+    const actions = body.querySelector(':scope > .dialog-actions');
+    return {
+        title: text(dialog.querySelector('#modal-title')),
+        tree: [...body.children].map(name).filter(item => item !== 'p.demo-note' && item !== 'h3'),
+        grid: { rows: [...grid.children].map(row => [text(row.querySelector('dt')), text(row.querySelector('dd'))]),
+            ...css(grid, 'display', 'gridTemplateColumns', 'columnGap', 'rowGap'),
+            dt: css(grid.querySelector('dt'), 'fontSize', 'color', 'textTransform'), dd: css(grid.querySelector('dd'), 'fontSize', 'color') },
+        notes: notes.map(note => ({ text: text(note), ...css(note, 'fontSize', 'color', 'marginTop') })),
+        actions: { tree: [...actions.children].map(name), ...css(actions, 'display', 'justifyContent', 'columnGap', 'marginTop'),
+            buttons: [...actions.children].map(item => ({ text: text(item), className: item.className,
+                ...css(item, 'backgroundColor', 'color', 'fontSize', 'paddingTop', 'paddingLeft', 'borderTopColor') })) },
+    };
+}
+
+/** Runs in the page: a button's class and tokens (the concept's danger button, or production's). */
+function measureButton(selector) {
+    const node = document.querySelector(selector);
+    const style = getComputedStyle(node);
+    return { className: node.className, ...Object.fromEntries(['backgroundColor', 'color', 'fontSize', 'fontWeight', 'paddingTop',
+        'paddingLeft', 'borderTopColor', 'borderTopLeftRadius', 'height'].map(key => [key, key === 'height'
+        ? `${Math.round(node.getBoundingClientRect().height * 100) / 100}px` : style[key]])) };
+}
+
+async function conceptReview(page, theme) {
+    await openConcept(page, 'prompt/alpine', theme);
+    await page.locator('#main .heading-actions').getByRole('button', { name: 'Edit metadata', exact: true }).click();
+    const archive = page.locator('dialog#modal[open] .dialog-actions button', { hasText: 'Archive…' });
+    await expect(archive).toBeVisible();await page.mouse.move(0, 0);
+    const danger = await page.evaluate(measureButton, 'dialog#modal[open] .dialog-actions button.danger');
+    await archive.click();
+    await expect(page.locator('dialog#modal[open] #modal-title')).toHaveText('Review maintenance plan');
+    await page.mouse.move(0, 0);
+    return { danger, review: await page.evaluate(measureReview) };
+}
+
+async function seedMixed(request) {
+    // One unpaired prompt beside the concept's paired ones, so the review deletes one and archives one.
+    await expect(await request.post('/prompt_manager/save', { data: { text: 'Unpaired study', category: 'Studies' } })).toBeOK();
+    const items = (await (await request.get('/prompt_manager/library?limit=100')).json()).items;
+    return { unpaired: items.find(item => item.text === 'Unpaired study'), paired: items.find(item => item.variant_count > 0) };
+}
+
+async function productionReview(page, request, theme) {
+    const { unpaired, paired } = await seedMixed(request);
+    await openProduction(page, 'library', theme);
+    for (const item of [unpaired, paired]) await page.locator(`article[data-prompt-id="${item.id}"] .image-overlays input`).check();
+    await page.locator('#library-bulkbar .bulkbar').getByRole('button', { name: 'Delete…', exact: true }).click();
+    await expect(page.locator('dialog#modal[open] #modal-title')).toHaveText('Delete or archive selected prompts');
+    await page.mouse.move(0, 0);
+    return page.evaluate(measureReview);
+}
+
+/** D152's changes to the concept's review; everything else must equal the concept. */
+function mapDeleteConcept({ danger, review }) {
+    const [cancel] = review.actions.buttons;
+    const note = review.notes.at(-1);
+    return {
+        ...review,
+        title: 'Delete or archive selected prompts',
+        // #482: the paired prompt is deletable, so the review asks Archive or Delete permanently (a D90 radio row).
+        tree: ['dl.meta-grid', 'div.field', 'p.small', 'p.small', 'div.dialog-actions'],
+        grid: { ...review.grid, rows: [['Delete permanently', '1 prompt not in a saved pair'],
+            ['Archive', '1 prompt in a saved pair: hidden, restorable'],
+            ['Image links removed', '0 images will no longer show these prompts (files stay on disk)'],
+            ['Backup', 'Create and verify before deleting']] },
+        notes: [
+            { ...note, text: 'Prompts in a saved pair, as its positive or its negative, are archived instead of deleted, so the '
+                + 'pair and its runs stay intact. To delete them later or bring them back, press Show archived prompts, select them '
+                + 'and press Delete… or Restore.' },
+            // A second paragraph takes the concept's own `p + p` margin (style.css 93-95).
+            { ...note, marginTop: '10px', text: "Deleting removes each prompt's text, title, category, tags, rating, notes, run count, last use and "
+                + 'image links. Deleted prompts cannot be brought back one by one: the only way back is to restore the safety '
+                + 'backup taken first (System › Backup & export), which returns the whole library to that moment. It is kept '
+                + 'among the last 10 safety backups. Image files, favorites, collections and recorded runs stay. A later scan, a '
+                + 'new image or a node run with the same text adds it again as a new prompt.' },
+        ],
+        actions: { ...review.actions, tree: [review.actions.tree[0], 'button.danger'], buttons: [cancel, { text: 'Delete 1 and archive 1 prompt', className: danger.className,
+            backgroundColor: danger.backgroundColor, color: danger.color, fontSize: danger.fontSize, paddingTop: danger.paddingTop,
+            paddingLeft: danger.paddingLeft, borderTopColor: danger.borderTopColor }] },
+    };
+}
+
+for (const theme of THEMES) for (const [width, height] of SIZES) {
+    test(`the Delete… review matches the concept's reviewed archive ${theme} ${width}x${height}`, async ({ page, request }, testInfo) => {
+        await page.setViewportSize({ width, height });
+        const concept = await conceptReview(page, theme);
+        expect(concept.review.tree).toEqual(['dl.meta-grid', 'p.small', 'div.dialog-actions']);
+        await page.screenshot({ path: testInfo.outputPath(`concept-archive-review-${theme}-${width}.png`) });
+        const production = await productionReview(page, request, theme);
+        await page.screenshot({ path: testInfo.outputPath(`production-delete-review-${theme}-${width}.png`) });
+        expect(diffInventory(mapDeleteConcept(concept), production)).toEqual([]);
+    });
+}
+
+test('the Delete… review comparison detects mutations', async ({ page, request }) => {
+    await page.setViewportSize({ width: 1600, height: 1100 });
+    const concept = await conceptReview(page, 'dark');
+    await productionReview(page, request, 'dark');
+    const expected = mapDeleteConcept(concept);
+    const paths = async () => diffInventory(expected, await page.evaluate(measureReview)).map(item => item.path);
+    expect(await paths()).toEqual([]);
+    await mutate(page, { css: '#modal-body .dialog-actions .danger { padding-left: 2px }' });
+    expect(await paths()).toContain('actions.buttons.1.paddingLeft');
+});
+
+test('the bar\'s Delete… matches the concept Images bar\'s danger action', async ({ page, request }) => {
+    await page.setViewportSize({ width: 1600, height: 1100 });
+    await openConcept(page, 'images', 'dark');
+    await page.locator('#image-grid .media-card').first().locator('input[data-select]').check();
+    await expect(page.locator('#bulkbar .bulkbar')).toBeVisible();await page.mouse.move(0, 0);
+    const reference = await page.evaluate(measureButton, '#bulkbar .bulkbar button.danger');
+    const { paired } = await seedMixed(request);
+    await openProduction(page, 'library', 'dark');
+    await page.locator(`article[data-prompt-id="${paired.id}"] .image-overlays input`).check();
+    await expect(page.locator('#library-bulkbar .bulkbar')).toBeVisible();await page.mouse.move(0, 0);
+    expect(await page.evaluate(measureButton, '#library-bulkbar .bulkbar button.danger')).toEqual(reference);
+});

tests/browser/library-editor-canonical-comparison.mjs

@@ -26,9 +26,15 @@ export function inventoryOf(dialog) {
     };
 }
 // Documented differences only: prototype Save label, prototype demo note, data-dependent values, and D153 (#447): the
-// concept's "Rating" select is not drawn.
+// concept's "Rating" select is not drawn. D161 (#481): the concept's one danger "Archive…" becomes "Archive variant…" and
+// is followed by the prompt's own "Delete prompt…", drawn as the same danger button.
+function mapButton(item) {
+    if(item.text==='Save in prototype') return [{...item,text:'Save'}];
+    if(item.text==='Archive…') return [{...item,text:'Archive variant…'},{...item,text:'Delete prompt…'}];
+    return [item];
+}
 export function mapConcept(reference) {
-    return {...reference,demo:false,fields:reference.fields.filter(item=>item.label!=='Rating'),actions:{...reference.actions,buttons:reference.actions.buttons.map(item=>item.text==='Save in prototype'?{...item,text:'Save'}:item)}};
+    return {...reference,demo:false,fields:reference.fields.filter(item=>item.label!=='Rating'),actions:{...reference.actions,buttons:reference.actions.buttons.flatMap(mapButton)}};
 }
 
 // Flatten an inventory to path->value so a comparison reports exactly which leaves differ.

tests/browser/library-editor-visual.spec.mjs

@@ -24,7 +24,7 @@ for(const [width,height,theme] of [[1600,1100,'dark'],[390,844,'light']])test(`e
     await variants(page).getByRole('button',{name:'Select negative variant 2',exact:true}).click();
     await page.getByRole('button',{name:'Edit metadata',exact:true}).click();await expect(editor(page)).toBeVisible();
     await expect(editor(page).getByRole('button',{name:'Save',exact:true})).toBeEnabled();
-    await editor(page).getByRole('button',{name:'Archive…',exact:true}).click();
+    await editor(page).getByRole('button',{name:'Archive variant…',exact:true}).click();
     await expect(editor(page).getByRole('button',{name:'Archive',exact:true})).toBeEnabled();await shot('editor-archive');
 
     let release;const held=new Promise(resolve=>{release=resolve;});