Merge remote-tracking branch 'origin/main' into feat/480-delete-images

An aria-disabled primary ignores every activation, a chosen one starts once,
and the apply helper refuses a missing mode. Tests cover a forced click, a
double click, Enter and Space on the primary, Enter in the dialog and a
submit event, and the server refusing an apply without an exact mode.

.github/workflows/portability.yml

@@ -51,5 +51,27 @@ 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

TESTING.md

@@ -251,6 +251,21 @@ the process exits non-zero.
   stacks.
 - When the watchdog fires under coverage, that run's coverage data is lost; the
   job fails anyway and the dump is what matters.
+- `PM_TEST_SLOWEST=N` (off by default) lists the N slowest tests after the
+  summary, each timed from its `start:` line to the next one like the hang
+  limit. The Windows and macOS jobs (`portability.yml`) set it to 40, so every
+  run shows where its time went (#487). The list prints only when the run
+  ends: a job GitHub cancels or a watchdog exit prints none, so read the
+  timestamps between `start:` lines instead.
+- On the Windows runners `TEMP` sits on the OS disk C:, where a flush took about
+  6 ms and a test database's create-to-delete about 111 ms (D: 0.5 ms and 23 ms,
+  macOS 19 ms). `portability.yml` points `TEMP`/`TMP` at a folder under
+  `RUNNER_TEMP` on D: before the suite. It switches 8.3 short names on for D:
+  and spells the folder by its short name, as the runner's own TEMP was
+  (`C:\Users\RUNNER~1\...`), logs it, and warns when there is none.
+  A test that waits on a fixed busy timeout, a network name (`\\server\share`:
+  use `\\localhost\<missing share>`) or a default TLS context (it reads the
+  Windows certificate store) costs seconds there; the slowest list shows it.
 
 Every job in `test.yml`, `code-quality.yml` and `viewerjs-check.yml` has
 `timeout-minutes` (`tests/vendor/ci-workflow.test.mjs` checks it). The rule is
@@ -2399,6 +2414,59 @@ npm run --prefix tools/vendor test:workflows -- workspace-collections.spec.mjs c
   images-concept.spec.mjs inspector-concept.spec.mjs shell-concept.spec.mjs --project chromium --project touch
 ```
 
+### Deleting images (#480)
+
+Images › Delete… removes the checked images (or "Select all N matching", at most 500, or
+the one file of the side panel or the viewer) from the library, and on request moves the
+files to the system trash. **No code path unlinks or overwrites a media file.**
+
+- **D159** (#480, maintainer request): the concept's "Remove links…" (`danger`, app.js
+  432) returns as **Delete…** in the same place (after Export metadata, before Clear), and
+  as the last action of the side panel and after Download in the Images viewer. Its
+  review is the concept's `maintenancePlan` (`dl.meta-grid`, Cancel and a primary) with a
+  radio group first, "What should happen to the files?", nothing checked (the ruling: ask
+  every time): "Remove from the library only" or "Also move the files to the trash". The
+  rows follow the chosen mode (Scope, Not changed, Prompt links, Image details, Backup,
+  Media on disk) and the primary is `danger`: "Remove N images from the library" / "Move
+  N images to the trash". `tests/browser/images-delete-concept.spec.mjs` compares the bar
+  button and the review's grid and actions with the concept.
+- **Routes.** `POST /prompt_manager/images/delete/preview {items | filter}` reviews both
+  modes at once under the maintenance preview lease and answers a reviewed plan
+  (`_plan_response`, operation `delete_images`): `library.count`, `trash.count`, `refused`
+  and `trash_refused` (`{count, counts, rows ≤ 50}` with coded reasons), `records`,
+  `orphaned_prompts` and the backup. Each target is matched to its entry in the current
+  listing (`not-listed` otherwise). `POST /prompt_manager/maintenance/apply {token, mode}`
+  (`mode` required for this plan, refused for any other) starts job kind `image_delete`
+  (admission `records`, not cancellable; groups `trashed`, `removed`, `not_changed`,
+  `errors`): sources unchanged, the promised safety backup, then per file a recheck
+  (containment and fingerprint) and
+  - library mode: one transaction deletes the file's records and writes its removal mark;
+  - trash mode: the file moves first (`utils/trash.py`); a refused move changes nothing;
+    then its records go. A records failure stops the job (`records-failed`); later files
+    are "Not attempted".
+- **What goes** (`database/image_removal.py`): image links under every spelling (pair
+  links by cascade, workflow documents released), image tags, auto-tag marks, pending
+  undecided Needs review items, the own category, collection memberships and the
+  favorite. A decided or revised review item keeps its decision and the annotation rows it
+  cites. Prompts, recorded runs and the vocabulary stay. Previews are removed.
+- **Removal marks** (`image_removals`, `image_removals_meta`): a library-only removal
+  hides exactly that file (locator, size, `mtime_ns`) from the workspace listing, the
+  output scan, the duplicate find, Tag output images and linking. A modified or replaced
+  file shows again. An explicit tag, category, favorite or collection write clears the
+  mark. Older builds ignore the tables (removed files show there); restoring an older
+  backup shows every file again; restoring a pre-delete backup after trash mode brings
+  back records of files that are in the trash (Missing-file links cleans them up). The
+  legacy gallery lists every file on disk, removed or not. Hard links have one mark per
+  path, and on FAT/exFAT/SMB a same-size replacement can stay hidden.
+- **Trash backends** (`utils/trash.py`, no dependency): Linux and BSD implement the
+  freedesktop Trash specification (home trash on the same mount, else the mount's
+  `.Trash/$uid` or `.Trash-$uid`), reserve the name with `O_EXCL` and move with
+  `renameat2(RENAME_NOREPLACE)`; without it, across devices, on a read-only mount or with
+  a trash inside an output root the move is refused (`no-trash`, `trash-in-root`). Windows
+  and macOS report `no-trash` until their own backends land; library-only removal works
+  everywhere. Tests use `tests/fake_trash.py`; the browser test server installs it, so no
+  test touches a real trash.
+
 ### Prompt library cards and recorded deviations (#249 PR E)
 
 The shelf uses the shared concept toolbar and caption (E1) and the concept's
@@ -2869,6 +2937,12 @@ mapped differences"):
   change within the workspace waits the same way: the shell renders the next route once a
   video still starting on Inspect settles (`leaveGuard`; the latest route wins). Route
   changes at `loadstart` + 0–7 ms hung 4–5 of 120 pages before and 0 of 120 after.
+  The shared viewer's own video gets the same guard (#483): a new selection while the
+  viewer D155 opened is still starting its video leaves the viewer open, untouched, until
+  that video reports metadata or an error (or 15 s), then closes it; the URL is revoked
+  after that, and a viewer asked for meanwhile (the new file's D155 open) opens then. The
+  merge queue hung once without it (run 40903238387: the PNG selection 12 ms after the
+  inline video showed, with the viewer's new video still starting).
   A real page leave cannot wait. With no page code releasing anything, navigating
   away hung 8 of 120 pages and reload 22 of 120 (3 and 8 of 120 with the page's own
   release), so that WebKit teardown stays a known limitation; dispose still releases at

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/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,305 @@
+"""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,
+    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 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:
+            return clear_marks(conn, locators)

database/image_removal_schema.py

@@ -0,0 +1,208 @@
+"""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 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/models.py

@@ -44,6 +44,7 @@ try:
 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 .workflow_documents import (
     DOCUMENT_ID_COLUMNS,
     migrate_workflow_documents,
@@ -366,6 +367,8 @@ class PromptModel:
         migrate_autotag_state(conn)
         # Image categories (#332): independent of every table above.
         migrate_image_categories(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)
 
@@ -1242,6 +1245,11 @@ class PromptModel:
             )
             if categories_error:
                 return False, categories_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,6 +18,7 @@ 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
@@ -465,6 +466,7 @@ class PromptDatabase(
     WorkflowStorageOperations,
     CategoryImportOperations,
     ImageCategoryOperations,
+    ImageRemovalOperations,
 ):
     """Database operations class for managing prompts."""
 
@@ -1806,6 +1808,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)

py/api/__init__.py

@@ -40,7 +40,6 @@ 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 .vocabulary import VocabularyRoutesMixin
@@ -75,6 +74,7 @@ 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 .source_rescope import SourceRescopeMixin
 
 try:
@@ -344,6 +344,7 @@ class PromptManagerAPI(
     LoraIntegrationMixin,
     LoraImportRoutesMixin,
     ImageCategoryRoutesMixin,
+    ImageDeleteMixin,
     SourceRescopeMixin,
 ):
     """REST API handler for PromptManager operations and web interface.
@@ -424,6 +425,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:
@@ -654,39 +661,35 @@ class PromptManagerAPI(
 
         @routes.get("/prompt_manager/workspace")
         async def serve_workspace_ui(request: web.Request) -> web.Response:
-            return self._serve_html_page(request, "workspace.html", "Workspace")
+            return self._serve_html_page("workspace.html", "Workspace")
 
         @routes.get("/prompt_manager/web")
         async def serve_web_ui(request):
-            return self._serve_html_page(request, "index.html", "Web UI", cache=False)
+            return self._serve_html_page("index.html", "Web UI", cache=False)
 
         @routes.get("/prompt_manager/gallery.html")
         async def serve_gallery_ui(request):
-            return self._serve_html_page(request, "metadata.html", "Gallery")
+            return self._serve_html_page("metadata.html", "Gallery")
 
         @routes.get("/prompt_manager/admin")
         async def serve_admin_ui(request):
-            return self._serve_html_page(request, "admin.html", "Admin UI")
+            return self._serve_html_page("admin.html", "Admin UI")
 
         @routes.get("/prompt_manager/gallery")
         async def serve_gallery_admin_ui(request):
-            return self._serve_html_page(request, "gallery.html", "Gallery")
+            return self._serve_html_page("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(
-                request, "lib", request.match_info.get("filepath")
-            )
+            return self._serve_static_file("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(
-                request, "js", request.match_info.get("filepath")
-            )
+            return self._serve_static_file("js", request.match_info.get("filepath"))
 
         # ── Register domain-specific routes from mixins ───────────────
 
@@ -715,6 +718,7 @@ 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_rescope_routes(routes)
 
         # Register gzip compression middleware (once)
@@ -731,11 +735,8 @@ class PromptManagerAPI(
 
         self.logger.info("All routes registered with decorator pattern")
 
-    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``).
-        """
+    def _serve_static_file(self, subdir, filepath):
+        """Serve ``web/<subdir>/<filepath>``; 403 outside it, 404 when missing."""
         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.
@@ -750,13 +751,10 @@ class PromptManagerAPI(
             return web.Response(text="Not Found", status=404)
         with open(file_path, "rb") as f:
             content = f.read()
-        return revalidated_response(request, content, content_type)
+        return web.Response(body=content, content_type=content_type)
 
-    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``).
-        """
+    def _serve_html_page(self, filename, title, cache=True):
+        """Serve ``web/<filename>``; 404 when missing, 500 when unreadable."""
         html_path = os.path.join(_get_project_root(), "web", filename)
         try:
             if cache and html_path in self._html_cache:
@@ -773,9 +771,7 @@ class PromptManagerAPI(
                     content_type="text/html",
                     status=404,
                 )
-            return revalidated_response(
-                request, content.encode("utf-8"), "text/html", charset="utf-8"
-            )
+            return web.Response(text=content, content_type="text/html", charset="utf-8")
         except Exception:
             self.logger.exception(f"Failed to load {title}")
             return web.Response(

py/api/admin.py

@@ -883,6 +883,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 +893,12 @@ class AdminRoutesMixin:
                     stat = os.stat(path)
                 except OSError:
                     continue
+                mark = marks.get(os.path.normcase(os.path.realpath(str(path))))
+                if mark is not None and tuple(mark) == (
+                    stat.st_size,
+                    stat.st_mtime_ns,
+                ):
+                    continue
                 key = _physical_file_key(path, stat)
                 if key not in seen:
                     seen.add(key)
@@ -1551,8 +1559,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/image_delete.py

@@ -0,0 +1,480 @@
+"""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.annotation_schema import MAX_LOCATOR_LENGTH
+    from ...utils import trash
+except ImportError:
+    from database import image_removal_schema
+    from database.annotation_schema import MAX_LOCATOR_LENGTH
+    from utils import trash
+
+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
+TRASH_FOLDERS = frozenset({".trash", ".trashes", "$recycle.bin"})
+
+# 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.",
+    "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 _is_trash_folder(name: str) -> bool:
+    folded = name.casefold()
+    return folded in TRASH_FOLDERS or (
+        folded.startswith(".trash-") and folded[7:].isdigit()
+    )
+
+
+def lexical_problem(relative: str) -> str | None:
+    """The refusal code a root-relative POSIX path earns before any file access."""
+    parts = relative.split("/")
+    if "thumbnails" in parts[:-1]:
+        return "thumbnail"
+    if any(_is_trash_folder(part) for part in parts[:-1]):
+        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, False
+        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 = True
+                job.record(
+                    "errors",
+                    {
+                        "path": target["rel"],
+                        "root": target["root_label"],
+                        "root_index": target["root_index"],
+                        "root_scope": target["root_scope"],
+                        "detail": RECORDS_FAILED,
+                    },
+                    f"! {target['root_label']}/{target['rel']} · {RECORDS_FAILED}",
+                )
+            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:
+            job.scope["failure"] = "records-failed"
+            job.publish({"type": "error", "code": "records-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
+                return "not_changed", code, 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)
+        hidden = set()
+        for locator, (size, mtime_ns) in marks.items():
+            entry = located.get(locator)
+            if entry is None:
+                continue
+            try:
+                info = os.lstat(entry.path)
+            except OSError:
+                continue
+            if (info.st_size, info.st_mtime_ns) == (size, mtime_ns):
+                hidden.add(entry.normalized)
+        return hidden
+
+    def _removed_paths(self, paths) -> set:
+        """The absolute paths among *paths* a removal mark hides (blocking)."""
+        marks = self._removal_marks()
+        if not marks:
+            return set()
+        hidden = set()
+        for path in paths:
+            mark = marks.get(os.path.normcase(os.path.realpath(str(path))))
+            if mark is None:
+                continue
+            try:
+                info = os.lstat(path)
+            except OSError:
+                continue
+            if (info.st_size, info.st_mtime_ns) == tuple(mark):
+                hidden.add(str(path))
+        return hidden
+
+
+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_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

@@ -2450,6 +2450,10 @@ class ImageRoutesMixin:
             def key(entry):
                 return _workspace_key(entry, order)
 
+        # #480: files removed from the library stay on disk, hidden here.
+        hidden = self._hidden_entries(listings)
+        if hidden:
+            entries = [entry for entry in entries if entry.normalized not in hidden]
         return (
             entries,
             key,

py/api/jobs.py

@@ -123,6 +123,8 @@ MODEL_DOWNLOAD_JOB_KIND = "model_download"
 DUPLICATES_JOB_KIND = "duplicates"
 DUPLICATE_GROUP_NAMES = ("groups_found",)
 DUPLICATE_CLEANUP_GROUP_NAMES = ("removed", "kept", "errors")
+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/maintenance.py

@@ -488,7 +488,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 +499,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 +553,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:

py/api/revalidation.py

@@ -1,47 +0,0 @@
-"""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
-    )

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})=>{
@@ -627,6 +627,10 @@ test('viewer matches the concept shell and toolbar order',async({page,request})=
         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));
+        // #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/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,7 +472,7 @@ 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([]);
     // The bar comparison notices a change to the production bar.

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/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;});};
@@ -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/server.py

@@ -220,6 +220,12 @@ def create_app(root: Path) -> web.Application:
     PromptManagerConfig.DEFAULT_DB_PATH = str(root / "browser.db")
     GalleryConfig.MONITORING_DIRECTORIES = [str(root / "output"), str(root / "input")]
 
+    # #480: Delete… never reaches a real system trash from a page test; the fake
+    # renames into this server's root, which goes when the server exits.
+    from tests import fake_trash
+
+    fake_trash.install(root / "fake-trash")
+
     from py.api import PromptManagerAPI
     from py import autotag
     from utils.hashing import generate_prompt_hash

tests/browser/workspace-collections.spec.mjs

@@ -527,7 +527,7 @@ test('Add to collection creates a collection and adds to it ignoring case (CUJ 1
     await expect(imageCards(page)).toHaveCount(21);
     for (const file of ['loose.png', 'landscape_01.png', 'experiment_01.png']) await check(page, file);
     await expect(bar(page).locator('button')).toHaveText(['Add tags', 'Auto-tag selected', 'Set category', 'Add to collection',
-        'Export metadata', 'Clear']);
+        'Export metadata', 'Delete…', 'Clear']);
     await barButton(page, 'Add to collection').click();
     // The concept's simpleInput with Add (D138) and the existing names as suggestions.
     await expect(addDialog(page).locator('label')).toHaveText('Collection name');
@@ -578,7 +578,7 @@ test('Remove from collection takes the checked images off the open collection (C
     await check(page, 'landscape_01.png');
     await check(page, 'landscape_02.png');
     await expect(bar(page).locator('button')).toHaveText(['Add tags', 'Auto-tag selected', 'Set category', 'Add to collection',
-        'Remove from collection', 'Export metadata', 'Clear']);
+        'Remove from collection', 'Export metadata', 'Delete…', 'Clear']);
     await barButton(page, 'Remove from collection').click();
     await expect(toast(page)).toHaveText('Removed 2 images from “Landscapes”.');
     await expect(imageCards(page)).toHaveCount(10);
@@ -715,7 +715,8 @@ test('Collections… in the side panel adds, removes and creates for one image (
     // The newest image, landscape_01, is the panel's.
     await expect(panel(page).getByRole('heading', { level: 2 })).toHaveText('landscape_01.png');
     const open = panel(page).getByRole('button', { name: 'Collections…', exact: true });
-    await expect(panel(page).locator('.detail-footer > button').last()).toHaveText('Collections…');
+    // #480 D159: Delete… follows Collections… at the end of the panel's actions.
+    await expect(panel(page).locator('.detail-footer > button').nth(-2)).toHaveText('Collections…');
     await open.click();
     await expect(collectionsDialog(page)).toBeVisible();
     await expect(choice(page, 'Landscapes')).toBeChecked();

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

@@ -0,0 +1,171 @@
+import { startCoverage, saveCoverage } from '../../tools/vendor/browser-coverage.mjs';
+import { test, expect } from './fixtures.mjs';
+import { seedConceptAt, openProduction } from './concept-inventory.mjs';
+
+// #480 (D159): Images › Delete…: the review asks every time (nothing preselected), "Remove from the library only"
+// hides the file and keeps it on disk, "Also move the files to the trash" moves it (the server's fake trash here).
+test.use({ locale: 'en-US', timezoneId: 'UTC' });
+
+const CARDS = '#workspace-main article[data-image-key]';
+const bar = page => page.locator('#workspace-main #bulkbar');
+const review = page => page.getByRole('dialog', { name: /^Delete \d+ images?$/ });
+const toast = page => page.locator('#workspace-toast');
+
+test.beforeEach(async ({ page, request, browserName }) => {
+    await startCoverage(page, browserName);
+    await seedConceptAt(page, request);
+    page.deleteErrors = [];
+    page.on('pageerror', error => page.deleteErrors.push(error.message));
+    page.on('dialog', dialog => { page.deleteErrors.push(`browser ${dialog.type()}`); dialog.dismiss().catch(() => {}); });
+});
+
+test.afterEach(async ({ page, browserName }) => {
+    await saveCoverage(page, browserName);
+    expect(page.deleteErrors).toEqual([]);
+});
+
+async function openImages(page) {
+    await page.setViewportSize({ width: 1600, height: 1100 });
+    await openProduction(page, 'images', 'dark');
+    await expect(page.locator(CARDS).first()).toBeVisible();
+    return page.locator(CARDS).count();
+}
+
+async function check(page, index) {
+    const card = page.locator(CARDS).nth(index);
+    const key = await card.getAttribute('data-image-key');
+    await card.locator('.image-overlays input[type="checkbox"]').check();
+    return key;
+}
+
+async function deleteChecked(page, mode) {
+    await bar(page).getByRole('button', { name: 'Delete…', exact: true }).click();
+    const dialog = review(page);
+    await expect(dialog).toBeVisible();
+    const primary = dialog.locator('.dialog-actions button.danger');
+    // Ask every time: no mode is preselected and the primary waits for one.
+    await expect(dialog.getByRole('radio')).toHaveCount(2);
+    for (const radio of await dialog.getByRole('radio').all()) await expect(radio).not.toBeChecked();
+    await expect(primary).toHaveAttribute('aria-disabled', 'true');
+    await expect(dialog.locator('dl.meta-grid')).toBeHidden();
+    const label = mode === 'trash' ? 'Also move the files to the trash' : 'Remove from the library only';
+    await dialog.getByRole('radio', { name: new RegExp(`^${label}`) }).check();
+    await expect(dialog.locator('dl.meta-grid')).toBeVisible();
+    await expect(primary).not.toHaveAttribute('aria-disabled', 'true');
+    return { dialog, primary };
+}
+
+for (const mode of ['library', 'trash']) {
+    test(`${mode}: the deleted image leaves Images and stays gone after a reload`, async ({ page }) => {
+        const before = await openImages(page);
+        const key = await check(page, 0);
+        const { dialog, primary } = await deleteChecked(page, mode);
+        await expect(dialog.locator('dt', { hasText: 'Media on disk' })).toBeVisible();
+        await expect(primary).toHaveText(mode === 'trash' ? 'Move 1 image to the trash' : 'Remove 1 image from the library');
+        await primary.click();
+        await expect(toast(page)).toContainText(mode === 'trash' ? 'Moved 1 image to the trash.' : 'Removed 1 image from the library.');
+        await expect(page.locator(`${CARDS}[data-image-key="${key}"]`)).toHaveCount(0);
+        await expect(page.locator(CARDS)).toHaveCount(before - 1);
+        await expect(bar(page)).toBeEmpty();
+        await page.reload();
+        await expect(page.locator(CARDS).first()).toBeVisible();
+        await expect(page.locator(`${CARDS}[data-image-key="${key}"]`)).toHaveCount(0);
+    });
+}
+
+test('Cancel changes nothing', async ({ page }) => {
+    const before = await openImages(page);
+    await check(page, 0);
+    const { dialog } = await deleteChecked(page, 'library');
+    await dialog.getByRole('button', { name: 'Cancel', exact: true }).click();
+    await expect(dialog).toBeHidden();
+    await expect(page.locator(CARDS)).toHaveCount(before);
+});
+
+test('switching modes redraws the rows from one review', async ({ page }) => {
+    await openImages(page);
+    await check(page, 0);
+    let previews = 0;
+    page.on('request', request => { if (request.url().endsWith('/images/delete/preview')) previews += 1; });
+    const { dialog, primary } = await deleteChecked(page, 'library');
+    await expect(dialog.locator('dd', { hasText: 'Files stay on disk, hidden from Images.' })).toBeVisible();
+    await dialog.getByRole('radio', { name: /^Also move the files to the trash/ }).check();
+    await expect(dialog.locator('dd', { hasText: 'Nothing is deleted permanently' })).toBeVisible();
+    await expect(primary).toHaveText('Move 1 image to the trash');
+    expect(previews).toBe(1);
+});
+
+test('Delete… in the inspector reviews that one image', async ({ page }) => {
+    const before = await openImages(page);
+    const card = page.locator(CARDS).first();
+    await card.getByRole('button', { name: /^Inspect / }).click();
+    const button = page.locator('#workspace-main .file-inspector button.danger', { hasText: 'Delete…' });
+    await expect(button).toBeVisible();
+    await button.click();
+    const dialog = review(page);
+    await expect(dialog).toBeVisible();
+    await dialog.getByRole('radio', { name: /^Remove from the library only/ }).check();
+    await dialog.locator('.dialog-actions button.danger').click();
+    await expect(page.locator(CARDS)).toHaveCount(before - 1);
+});
+
+test('Delete… in the viewer closes it after the delete', async ({ page }) => {
+    const before = await openImages(page);
+    await page.locator(CARDS).first().getByRole('button', { name: /^Open / }).click();
+    const viewer = page.locator('dialog#viewer-dialog[open]');
+    await expect(viewer).toBeVisible();
+    await viewer.getByRole('button', { name: 'Delete…', exact: true }).click();
+    const dialog = review(page);
+    await expect(dialog).toBeVisible();
+    await dialog.getByRole('radio', { name: /^Remove from the library only/ }).check();
+    await dialog.locator('.dialog-actions button.danger').click();
+    await expect(viewer).toBeHidden();
+    await expect(page.locator(CARDS)).toHaveCount(before - 1);
+});
+
+// The review never starts a delete without an explicit choice (live test 2026-10-29): every activation of the
+// aria-disabled primary is ignored, the dialog has no form to submit, and a chosen primary starts once.
+test('without a chosen mode nothing can start a delete', async ({ page }) => {
+    const before = await openImages(page);
+    await check(page, 0);
+    const applies = [];
+    page.on('request', request => { if (request.url().includes('/maintenance/apply')) applies.push(request.postDataJSON()); });
+    await bar(page).getByRole('button', { name: 'Delete…', exact: true }).click();
+    const dialog = review(page);
+    await expect(dialog).toBeVisible();
+    const primary = dialog.locator('.dialog-actions button.danger');
+    await expect(primary).toHaveAttribute('aria-disabled', 'true');
+    await expect(primary).toHaveText('Choose what happens to the files');
+    // Visibly dimmed while no mode is chosen.
+    await expect(primary).toHaveCSS('opacity', '0.55');
+    await expect(primary).toHaveCSS('cursor', 'not-allowed');
+    await primary.click({ force: true });
+    await primary.dblclick({ force: true });
+    await primary.focus();
+    await page.keyboard.press('Enter');
+    await page.keyboard.press('Space');
+    await dialog.locator('#modal-title').click();
+    await page.keyboard.press('Enter');
+    expect(await dialog.locator('form').count()).toBe(0);
+    await page.evaluate(() => document.querySelector('dialog#modal[open]')?.dispatchEvent(new Event('submit', { bubbles: true })));
+    await expect(dialog).toBeVisible();
+    expect(applies).toEqual([]);
+    for (const radio of await dialog.getByRole('radio').all()) await expect(radio).not.toBeChecked();
+    await dialog.getByRole('button', { name: 'Cancel', exact: true }).click();
+    await expect(page.locator(CARDS)).toHaveCount(before);
+    expect(applies).toEqual([]);
+});
+
+test('a chosen mode starts exactly one delete, with that mode, however often it is pressed', async ({ page }) => {
+    const before = await openImages(page);
+    await check(page, 0);
+    const applies = [];
+    page.on('request', request => { if (request.url().includes('/maintenance/apply')) applies.push(request.postDataJSON()); });
+    const { primary } = await deleteChecked(page, 'library');
+    await primary.dblclick();
+    await expect(toast(page)).toContainText('Removed 1 image from the library.');
+    await expect(page.locator(CARDS)).toHaveCount(before - 1);
+    expect(applies.length).toBe(1);
+    expect(applies[0].mode).toBe('library');
+    expect(Object.keys(applies[0]).sort()).toEqual(['mode', 'token']);
+});

tests/browser/workspace-images.spec.mjs

@@ -1794,7 +1794,7 @@ test('card checkboxes select images into the concept bar and Clear empties it',
     await checkbox(first, 'same.png').check();
     await expect(bulkbar(page).locator('.bulkbar > strong')).toHaveText('1 selected');
     // D85: Add tags (the concept's) and Auto-tag selected lead the real actions.
-    await expect(barButtons(page)).toHaveText(['Add tags', 'Auto-tag selected', 'Set category', 'Add to collection', 'Export metadata', 'Clear']);
+    await expect(barButtons(page)).toHaveText(['Add tags', 'Auto-tag selected', 'Set category', 'Add to collection', 'Export metadata', 'Delete…', 'Clear']);
     // Toggling neither opens the viewer, nor inspects, nor moves focus.
     await expect(page.locator('dialog[open]')).toHaveCount(0);
     await expect(checkbox(first, 'same.png')).toBeFocused();

tests/browser/workspace-inspect.spec.mjs

@@ -704,6 +704,147 @@ test('a video replaced while still starting is reset once it settles', async ({
     }
 });
 
+test('a new selection while the viewer\'s own video is still starting leaves it until it settles (#483)', async ({ context, browserName }) => {
+    test.setTimeout(180000);
+    // D155 opens the viewer once the inline preview has settled; the viewer then starts a video of its own. A new
+    // selection from that video's own `loadstart` (readyState HAVE_NOTHING), or a few milliseconds after it, must not
+    // reset it or take it off the page while it starts (WebKit, GStreamer: TESTING.md D62; merge-queue run
+    // 40903238387). The viewer stays open until it reports metadata or an error, then closes; the URL goes last.
+    for (let attempt = 0; attempt < 10; attempt++) {
+        const page = await context.newPage();
+        try {
+            if (attempt === 0) await startCoverage(page, browserName);
+            await page.addInitScript(() => {
+                const load = HTMLMediaElement.prototype.load;
+                HTMLMediaElement.prototype.load = function () { if (!this.hasAttribute('src')) { this.__resets = (this.__resets ?? 0) + 1; this.__resetReadyState ??= this.readyState; } return load.call(this); };
+                const revoke = URL.revokeObjectURL;
+                URL.revokeObjectURL = url => {
+                    const starting = window.__starting;
+                    if (url === starting?.url) starting.revoked = { src: starting.node.hasAttribute('src'), connected: starting.node.isConnected, resets: starting.node.__resets ?? 0 };
+                    return revoke(url);
+                };
+            });
+            await openInspect(page);
+            await page.evaluate(() => {
+                const input = document.querySelector('#local-file');
+                new MutationObserver((records, observer) => {
+                    const video = document.querySelector('#viewer-dialog video');
+                    if (!video) return;
+                    observer.disconnect();
+                    const replace = () => {
+                        window.__starting = { node: video, url: video.getAttribute('src'), readyState: video.readyState };
+                        input.files = window.__pngFiles; input.dispatchEvent(new Event('change', { bubbles: true }));
+                        window.__starting.replaced = { connected: video.isConnected, src: video.hasAttribute('src'), resets: video.__resets ?? 0 };
+                    };
+                    video.addEventListener('loadstart', () => { if (window.__delay) setTimeout(replace, window.__delay); else replace(); }, { once: true });
+                }).observe(document.body, { childList: true, subtree: true });
+            });
+            const png = parityPng('f02.png');
+            await page.evaluate(bytes => {
+                const transfer = new DataTransfer();
+                transfer.items.add(new File([new Uint8Array(bytes)], 'f02.png', { type: 'image/png' }));
+                window.__pngFiles = transfer.files;
+            }, [...png.buffer]);
+            await page.evaluate(delay => { window.__delay = delay; }, attempt % 5);
+            await choose(page, localFixture('core-tags.webm', 'video/webm'));
+            await expect(preview(page).locator('img.local-file-preview')).toBeVisible({ timeout: 8000 });
+            await expect.poll(() => page.evaluate(() => window.__starting?.revoked ?? null)).toEqual({ src: false, connected: false, resets: 1 });
+            const { readyState, replaced } = await page.evaluate(() => ({ readyState: window.__starting.readyState, replaced: window.__starting.replaced }));
+            if (attempt % 5 === 0) expect(readyState).toBe(0);
+            // Starting when replaced: still on the page with its URL, never reset; settled: released at once.
+            expect(replaced).toEqual(readyState === 0 ? { connected: true, src: true, resets: 0 } : { connected: false, src: false, resets: 1 });
+            expect(await page.evaluate(() => window.__starting.node.__resets)).toBe(1);
+            // Its one reset came after it had metadata, never while it was starting.
+            expect(await page.evaluate(() => window.__starting.node.__resetReadyState)).toBeGreaterThanOrEqual(1);
+            // The new file's viewer opens after the old one closed, and the page still answers.
+            await expect(viewer(page).getByRole('img', { name: 'f02.png', exact: true })).toBeVisible({ timeout: 8000 });
+            await expect(viewer(page).locator('video')).toHaveCount(0);
+            if (attempt === 0) await saveCoverage(page, browserName);
+        } finally {
+            await page.close();
+        }
+    }
+});
+
+/** #483: holds the viewer's video in HAVE_NOTHING (its `loadedmetadata` never fires) so a hold can be driven by hand. */
+async function stallViewerVideo(page) {
+    await page.addInitScript(() => {
+        const describe = Object.getOwnPropertyDescriptor(HTMLMediaElement.prototype, 'readyState');
+        Object.defineProperty(HTMLMediaElement.prototype, 'readyState', { configurable: true, get() {
+            return this.closest?.('#viewer-dialog') && window.__stallViewer ? 0 : describe.get.call(this);
+        } });
+        window.addEventListener('loadedmetadata', event => {
+            if (window.__stallViewer && event.target.closest?.('#viewer-dialog')) event.stopImmediatePropagation();
+        }, true);
+    });
+}
+
+test('a held viewer closed by hand, or replaced twice, opens only the latest file\'s viewer at once (#483)', async ({ page }) => {
+    await stallViewerVideo(page);
+    await openInspect(page);
+    await page.evaluate(() => { window.__stallViewer = true; });
+    await choose(page, localFixture('core-tags.webm', 'video/webm'));
+    await expect(viewer(page).locator('video')).toBeAttached();
+    // Two selections while the viewer's video is (made to look) still starting: the viewer stays, untouched.
+    await chooser(page).setInputFiles(parityPng('f02.png', 'first.png'));
+    await chooser(page).setInputFiles(parityPng('f02.png', 'second.png'));
+    await expect(preview(page).locator('img.local-file-preview')).toBeVisible();
+    await expect(viewer(page).locator('video')).toBeAttached();
+    // Open file metadata pressed meanwhile (the page behind the viewer is inert, so by script): the latest request.
+    await action(page).evaluate(node => node.click());
+    // Closing it by hand ends the hold at once: the latest file's viewer opens, once, well before D62's 15 s.
+    await page.keyboard.press('Escape');
+    await expect(viewer(page).getByRole('img', { name: 'second.png', exact: true })).toBeVisible({ timeout: 5000 });
+    await expect(viewer(page).getByRole('img', { name: 'first.png', exact: true })).toHaveCount(0);
+    await expect(viewer(page).locator('video')).toHaveCount(0);
+    await expect(page.locator('dialog[open]')).toHaveCount(1);
+});
+
+test('a held viewer revokes its URL only once it is gone, drops a stale open, and ends on page leave (#483)', async ({ page }) => {
+    await stallViewerVideo(page);
+    await openInspect(page);
+    await trackReleases(page);
+    await page.evaluate(() => { window.__stallViewer = true; });
+    const videoUrl = () => page.evaluate(() => document.querySelector('#viewer-dialog video')?.getAttribute('src') ?? null);
+    const released = url => page.evaluate(url => window.__released.filter(item => item.url === url), url);
+    await choose(page, localFixture('core-tags.webm', 'video/webm'));
+    await expect(viewer(page).locator('video')).toBeAttached();
+    const first = await videoUrl();
+    await chooser(page).setInputFiles(parityPng('f02.png', 'held.png'));
+    await expect(preview(page).locator('img.local-file-preview')).toBeVisible();
+    // The session asked to revoke the video's URL; it waits while the held viewer still plays it.
+    expect(await released(first)).toEqual([]);
+    // An Open asked for during the hold, then an AVI (which never opens the viewer by itself): the open for the earlier
+    // file is stale and is dropped when the hold ends.
+    await action(page).evaluate(node => node.click());
+    const avi = { name: 'clip.avi', mimeType: 'video/x-msvideo', buffer: Buffer.concat([Buffer.from('RIFF'), Buffer.alloc(4), Buffer.from('AVI LIST'), Buffer.alloc(64)]) };
+    await chooser(page).setInputFiles(avi);
+    await expect(page.locator('#workspace-toast')).toHaveText(TOAST);
+    await page.keyboard.press('Escape');
+    await expect(page.locator('dialog[open]')).toHaveCount(0);
+    await expect.poll(() => released(first)).toEqual([{ url: first, live: 0 }]);
+    // Page leave while held: the viewer goes and the deferred URL is revoked with nothing holding it.
+    await choose(page, localFixture('core-tags.webm', 'video/webm'));
+    await expect(viewer(page).locator('video')).toBeAttached();
+    const second = await videoUrl();
+    await chooser(page).setInputFiles(parityPng('f02.png', 'leave.png'));
+    await expect(viewer(page).locator('video')).toBeAttached();
+    expect(await released(second)).toEqual([]);
+    await page.evaluate(() => window.dispatchEvent(new PageTransitionEvent('pagehide', { persisted: true })));
+    await expect(page.locator('dialog[open]')).toHaveCount(0);
+    await expect.poll(() => released(second)).toEqual([{ url: second, live: 0 }]);
+    // Another route while held: the page's dispose ends the hold and revokes the URL with nothing holding it.
+    await page.evaluate(() => window.dispatchEvent(new PageTransitionEvent('pageshow', { persisted: true })));
+    await choose(page, localFixture('core-tags.webm', 'video/webm'));
+    await expect(viewer(page).locator('video')).toBeAttached();
+    const third = await videoUrl();
+    await chooser(page).setInputFiles(parityPng('f02.png', 'route.png'));
+    await expect(viewer(page).locator('video')).toBeAttached();
+    await page.evaluate(() => { window.location.hash = '#/library'; });
+    await expect(main(page).getByRole('heading', { level: 1, name: 'Prompt library', exact: true })).toBeVisible();
+    await expect.poll(() => released(third)).toEqual([{ url: third, live: 0 }]);
+});
+
 test('leaving for another route while a video starts waits for it to settle', async ({ context, browserName }) => {
     test.setTimeout(180000);
     // A WebKit (GStreamer) video detached before its pipeline had started deadlocked the page (TESTING.md D62); a

tests/browser/workspace-stale-source.spec.mjs

@@ -37,6 +37,15 @@ test.afterEach(async ({ page, request, browserName }) => {
     await expect(await request.post('/__test__/tagging/reset', { data: {} })).toBeOK();
 });
 
+async function openSettled(page, url) {
+    // The page reads its selected file's details as soon as it lists. A source change before that read is answered
+    // makes the page find the file again on its own, so the test's old-numbering locators are gone before they are
+    // used (#484): every test changes sources only after the read was answered.
+    const inspected = page.waitForResponse(item => new URL(item.url()).pathname.startsWith('/prompt_manager/images/inspect/') &&
+        item.request().method() === 'GET');
+    await page.goto(url);
+    expect((await inspected).status()).toBe(200);
+}
 async function openTag(page, path = FIRST, root = 0) {
     const observed = answered(page, '/image-annotations/observe');
     await tagButton(page, path, root).click();
@@ -53,7 +62,7 @@ async function savedTags(request, path, root) {
 }
 
 test('renumbered sources under an open Tag dialog: it finds the file again, keeps the draft and Generate works', async ({ page, request }) => {
-    await page.goto(images);
+    await openSettled(page, images);
     await expect(card(page, FIRST)).toHaveClass(/\bselected\b/);
     expect((await openTag(page)).status()).toBe(200);
     await manual(page).fill('kept_draft');
@@ -91,7 +100,7 @@ test('renumbered sources under an open Tag dialog: it finds the file again, keep
 });
 
 test('the inspector finds a renumbered file again; checked files from the old numbering are cleared', async ({ page, request }) => {
-    await page.goto(images);
+    await openSettled(page, images);
     await expect(card(page, FIRST)).toHaveClass(/\bselected\b/);
     await card(page, FIRST).getByRole('checkbox', { name: /^Select / }).check();
     await expect(page.locator('.bulkbar')).toBeVisible();
@@ -113,7 +122,7 @@ test('the inspector finds a renumbered file again; checked files from the old nu
 });
 
 test('a scope that cannot be mapped: Generate and Save are visibly disabled, the reason and Refresh sit beside them', async ({ page }) => {
-    await page.goto(images);
+    await openSettled(page, images);
     await expect(card(page, FIRST)).toBeVisible();
     await expect(await page.request.post('/__test__/scope-key/rotate')).toBeOK();
     const rescoped = answered(page, '/images/rescope');
@@ -157,7 +166,7 @@ test('a scope that cannot be mapped: Generate and Save are visibly disabled, the
 });
 
 test('an unmappable inspector read keeps its visible Refresh, and Refresh recovers the panel', async ({ page }) => {
-    await page.goto(images);
+    await openSettled(page, images);
     await expect(card(page, FIRST)).toHaveClass(/\bselected\b/);
     await expect(await page.request.post('/__test__/scope-key/rotate')).toBeOK();
     const rescoped = answered(page, '/images/rescope');
@@ -173,7 +182,7 @@ test('an unmappable inspector read keeps its visible Refresh, and Refresh recove
 
 test('a Load more cursor the server cannot verify offers Refresh instead of a dead Retry', async ({ page, request }) => {
     await expect(await request.post('/__test__/tagging/reset', { data: { extra_images: 60 } })).toBeOK();
-    await page.goto(images);
+    await openSettled(page, images);
     await expect(cards(page).first()).toBeVisible();
     await expect(page.locator('#gallery-continuation')).toContainText('more load as you scroll');
     await expect(await request.post('/__test__/scope-key/rotate')).toBeOK();
@@ -191,7 +200,7 @@ test('a Load more cursor the server cannot verify offers Refresh instead of a de
 });
 
 test('a viewer whose media source was renumbered closes onto the file, kept selected', async ({ page, request }) => {
-    await page.goto(images);
+    await openSettled(page, images);
     await expect(card(page, FIRST)).toHaveClass(/\bselected\b/);
     await expect(await request.post('/__test__/sources/reverse')).toBeOK();
     const rescoped = answered(page, '/images/rescope');
@@ -217,7 +226,7 @@ test('a collection page finds a renumbered member again and stays on its members
     await expect(added).toBeOK();
     const collections = (await (await request.get('/prompt_manager/collections')).json()).collections;
     const { id } = collections.find(item => item.name === name);
-    await page.goto(`/prompt_manager/workspace#/collection/${id}`);
+    await openSettled(page, `/prompt_manager/workspace#/collection/${id}`);
     await expect(cards(page)).toHaveCount(1);
     await expect(card(page, FIRST)).toHaveClass(/\bselected\b/);
     await expect(await request.post('/__test__/sources/reverse')).toBeOK();

tests/browser/workspace-tagging-bulk.spec.mjs

@@ -73,7 +73,7 @@ const jobStart = page => page.waitForRequest(item => item.method() === 'POST' &&
 
 test('selection bar shows Add tags then Auto-tag selected before Set category', async ({ page }) => {
     await select(page, FIRST);
-    await expect(bar(page).locator(':scope > *')).toHaveText(['1 selected', 'Add tags', 'Auto-tag selected', 'Set category', 'Add to collection', 'Export metadata', 'Clear']);
+    await expect(bar(page).locator(':scope > *')).toHaveText(['1 selected', 'Add tags', 'Auto-tag selected', 'Set category', 'Add to collection', 'Export metadata', 'Delete…', 'Clear']);
 });
 
 test('Add tags runs a job and the cards show the tags after it finishes', async ({ page, request }) => {

tests/fake_trash.py

@@ -0,0 +1,52 @@
+"""A fake system trash for tests (#480): renames into a folder outside every root.
+
+``install(folder)`` replaces ``utils.trash.BACKEND`` and returns the fake;
+``uninstall`` puts the real one back. ``refuse[path] = code`` refuses one
+file with that code. The fake never copies or unlinks; it records each call.
+The disposable browser server installs it when ``PM_TEST_FAKE_TRASH=1``, so
+no test touches a real trash.
+"""
+
+import os
+
+from utils import trash
+
+
+class FakeTrash:
+    def __init__(self, folder):
+        self.folder = str(folder)
+        os.makedirs(self.folder, exist_ok=True)
+        self.refuse = {}
+        self.moved = []  # (source, destination)
+        self.calls = []
+
+    def available(self, path, roots):
+        self.calls.append(("available", str(path)))
+        return self.refuse.get(str(path))
+
+    def move(self, path, roots):
+        self.calls.append(("move", str(path)))
+        code = self.refuse.get(str(path))
+        if code:
+            raise trash.TrashRefused(code)
+        destination = os.path.join(
+            self.folder, f"{len(self.moved)}-{os.path.basename(path)}"
+        )
+        if os.path.lexists(destination):
+            raise trash.TrashRefused(trash.FAILED)
+        os.rename(path, destination)
+        self.moved.append((str(path), destination))
+        return "fake trash"
+
+
+_ORIGINAL = trash.BACKEND
+
+
+def install(folder):
+    fake = FakeTrash(folder)
+    trash.BACKEND = fake
+    return fake
+
+
+def uninstall():
+    trash.BACKEND = _ORIGINAL

tests/fixtures/legacy_contract_snapshot.json

@@ -118,6 +118,7 @@
     "POST /prompt_manager/image-annotations/observe",
     "POST /prompt_manager/image-annotations/suggest",
     "POST /prompt_manager/images/clear-thumbnails",
+    "POST /prompt_manager/images/delete/preview",
     "POST /prompt_manager/images/generate-thumbnails",
     "POST /prompt_manager/images/generate-thumbnails/progress",
     "POST /prompt_manager/images/link",

tests/js/workspace-images-delete.test.js

@@ -0,0 +1,149 @@
+import test from 'node:test';
+import assert from 'node:assert/strict';
+
+// #480 (D159): Images › Delete…: the review's rows, sentences, labels and outcome toasts.
+const loaded = await import('../../web/js/workspace/images-delete.mjs').catch(() => null);
+const module = () => { assert.ok(loaded, 'images-delete.mjs is not implemented'); return loaded; };
+const pageScope = () => {
+    const controller = new AbortController();
+    return { signal: controller.signal, cleanup: () => {}, abort: () => controller.abort() };
+};
+const plan = (over = {}) => ({
+    token: 't', library: { count: 3 }, trash: { count: 2 },
+    refused: { count: 1, counts: { 'not-listed': 1 }, rows: [{ path: 'sub/gone.png', code: 'not-listed', detail: 'No longer listed in Images.' }] },
+    trash_refused: { count: 1, counts: { 'no-trash': 1 }, rows: [{ path: 'b.png', code: 'no-trash', detail: 'This drive has no trash.' }] },
+    records: { prompt_links: 2, pair_links: 0, image_tags: 1, auto_tagged: 0, categories: 1, collections: 2, favorites: 0, review_items: 0 },
+    orphaned_prompts: { count: 1, titles: ['Red fox'] },
+    backup: { policy: 'on' }, disk: null, ...over,
+});
+
+test('the words of the review and its routes', () => {
+    const { PREVIEW_URL, MAX_DELETE, TOO_MANY, QUESTION, MODES } = module();
+    assert.equal(PREVIEW_URL, '/prompt_manager/images/delete/preview');
+    assert.equal(MAX_DELETE, 500);
+    assert.equal(TOO_MANY, 'Select at most 500 images.');
+    assert.equal(QUESTION, 'What should happen to the files?');
+    assert.deepEqual(MODES.map(mode => mode.label), ['Remove from the library only', 'Also move the files to the trash']);
+});
+
+test('library rows: scope, refusals, links, details, backup and media', () => {
+    const { deleteRows } = module();
+    const rows = Object.fromEntries(deleteRows(plan(), 'library'));
+    assert.equal(rows.Scope, '3 images');
+    assert.equal(rows['Not changed'], '1: gone.png. No longer listed in Images.');
+    assert.equal(rows['Prompt links'], '2 links. 1 prompt loses its last image: Red fox. The prompt stays on the Library page.');
+    assert.equal(rows['Image details'], '1 with image tags, 1 category, 2 collection memberships');
+    assert.equal(rows.Backup, 'Create and verify before changes (the database, not the files)');
+    assert.equal(rows['Media on disk'], 'Files stay on disk, hidden from Images.');
+});
+
+test('trash rows add the trash refusals and the trash media line', () => {
+    const { deleteRows } = module();
+    const rows = Object.fromEntries(deleteRows(plan({ backup: { policy: 'off' } }), 'trash'));
+    assert.equal(rows.Scope, '2 images');
+    assert.match(rows['Not changed'], /^2: gone\.png, b\.png\. No longer listed in Images\. This drive has no trash\.$/);
+    assert.equal(rows.Backup, 'Off: no safety backup (System › Backup & export › Recovery policy)');
+    assert.match(rows['Media on disk'], /Nothing is deleted permanently/);
+});
+
+test('rows leave out what is zero', () => {
+    const { deleteRows } = module();
+    const bare = plan({ refused: { count: 0, rows: [] }, trash_refused: { count: 0, rows: [] },
+        records: { prompt_links: 0 }, orphaned_prompts: { count: 0, titles: [] } });
+    assert.deepEqual(deleteRows(bare, 'library').map(([term]) => term), ['Scope', 'Backup', 'Media on disk']);
+    const many = plan({ orphaned_prompts: { count: 5, titles: ['A', 'B', 'C'] } });
+    assert.match(Object.fromEntries(deleteRows(many, 'library'))['Prompt links'],
+        /5 prompts lose their last image: A, B, C and 2 more\. The prompts stay on the Library page\./);
+    assert.equal(module().refusedValue({ count: 5, rows: [{ path: 'a.png' }] }), '5: a.png and 4 more');
+    assert.equal(module().refusedValue({ count: 0 }), null);
+});
+
+test('sentences follow the mode and the backup policy', () => {
+    const { deleteSentences, SENTENCES, USE_LIBRARY_ONLY, NO_TRASH_HERE } = module();
+    assert.deepEqual(deleteSentences(plan(), 'library'), [SENTENCES.libraryBackup]);
+    assert.deepEqual(deleteSentences(plan({ backup: { policy: 'off' } }), 'library'), [SENTENCES.libraryNoBackup]);
+    assert.deepEqual(deleteSentences(plan(), 'trash'), [SENTENCES.trash, USE_LIBRARY_ONLY]);
+    const nowhere = plan({ trash: { count: 0 }, trash_refused: { count: 3, rows: [] } });
+    assert.deepEqual(deleteSentences(nowhere, 'trash'), [SENTENCES.trash, NO_TRASH_HERE, USE_LIBRARY_ONLY]);
+    assert.deepEqual(deleteSentences(plan({ trash_refused: { count: 0 } }), 'trash'), [SENTENCES.trash]);
+});
+
+test('primary labels and the close-only reasons', () => {
+    const { primaryLabel, closeOnlyReason, NONE_DELETABLE } = module();
+    assert.equal(primaryLabel('library', 1), 'Remove 1 image from the library');
+    assert.equal(primaryLabel('trash', 4), 'Move 4 images to the trash');
+    assert.equal(closeOnlyReason(plan()), null);
+    assert.equal(closeOnlyReason(plan({ library: { count: 0 }, trash: { count: 0 } })), NONE_DELETABLE);
+    assert.equal(closeOnlyReason(plan({ disk: { sufficient: false, required_bytes: 2048, free_bytes: 1024 } })),
+        'There is not enough free space for the safety backup (needs 2,048 bytes, has 1,024).');
+});
+
+test('outcome toasts per mode and failure', () => {
+    const { outcomeToast } = module();
+    assert.equal(outcomeToast({ scope: { mode: 'library' }, counters: { removed: 3, not_changed: 0 }, total: 3 }),
+        'Removed 3 images from the library.');
+    assert.equal(outcomeToast({ scope: { mode: 'trash' }, counters: { trashed: 2, not_changed: 1 }, total: 3 }),
+        'Moved 2 images to the trash. 1 was not changed (see Activity).');
+    assert.equal(outcomeToast({ scope: { mode: 'trash' }, counters: { trashed: 0, not_changed: 2 }, total: 2 }),
+        'Moved 0 images to the trash. 2 were not changed (see Activity).');
+    assert.equal(outcomeToast({ scope: { mode: 'library', failure: 'records-failed' }, counters: { removed: 1 }, total: 4 }),
+        'Stopped after 1 of 4 images; the rest were not changed. See Activity.');
+    assert.equal(outcomeToast({ scope: { failure: 'backup-failed' } }), 'The safety backup failed; nothing was changed.');
+    assert.equal(outcomeToast({ scope: { failure: 'sources-changed' } }), 'The sources changed; nothing was changed.');
+});
+
+test('review failures name the cause', () => {
+    const { reviewFailure, SOURCE_CHANGED, STORAGE_CHANGED, TOO_MANY, REVIEW_FAILED } = module();
+    assert.equal(reviewFailure({ code: 'source-changed' }), SOURCE_CHANGED);
+    assert.equal(reviewFailure({ code: 'storage-changed' }), STORAGE_CHANGED);
+    assert.equal(reviewFailure({ code: 'too-many-images' }), TOO_MANY);
+    assert.equal(reviewFailure({ code: 'busy', message: 'Scan running' }), 'Scan running');
+    assert.equal(reviewFailure(new Error('boom')), REVIEW_FAILED);
+});
+
+test('over 500 refuses with the toast before any request', async () => {
+    const { openDeleteImages, TOO_MANY } = module();
+    const toasts = [];
+    const api = { post: () => assert.fail('no request') };
+    await openDeleteImages({ api, scope: pageScope(), images: [], filter: {}, count: 501, notify: text => toasts.push(text) });
+    assert.deepEqual(toasts, [TOO_MANY]);
+    await openDeleteImages({ api, scope: pageScope(), images: [], notify: text => toasts.push(text) });
+    assert.deepEqual(toasts, [TOO_MANY]);
+});
+
+test('a failed review toasts and releases its trigger', async () => {
+    const { openDeleteImages, SOURCE_CHANGED } = module();
+    const toasts = [], sent = [];
+    const attributes = new Map();
+    const trigger = { textContent: 'Delete…', getAttribute: name => attributes.get(name) ?? null,
+        setAttribute: (name, value) => attributes.set(name, value), removeAttribute: name => attributes.delete(name) };
+    const api = { post: async (url, body) => { sent.push([url, body]); throw Object.assign(new Error('x'), { code: 'source-changed' }); } };
+    await openDeleteImages({ api, scope: pageScope(), trigger, notify: text => toasts.push(text),
+        images: [{ root_index: 0, root_scope: 'a'.repeat(64), path: 'a.png', filename: 'a.png' }] });
+    assert.deepEqual(sent, [['/prompt_manager/images/delete/preview', { items: [{ root_index: 0, root_scope: 'a'.repeat(64), path: 'a.png' }] }]]);
+    assert.deepEqual(toasts, [SOURCE_CHANGED]);
+    assert.equal(trigger.textContent, 'Delete…');
+    assert.equal(trigger.getAttribute('aria-disabled'), null);
+});
+
+test('apply sends the token and the mode, and toasts that the job started', async () => {
+    const { applyDelete } = module();
+    const sent = [], toasts = [];
+    const api = { post: async (url, body) => { sent.push([url, body]); return { job: { id: 'j', title: 'Delete images (move to the trash)' } }; } };
+    const store = { track: job => job, load: () => {} };
+    const job = await applyDelete({ api, store, plan: plan(), mode: 'trash', notify: text => toasts.push(text) });
+    assert.equal(job.id, 'j');
+    assert.deepEqual(sent, [['/prompt_manager/maintenance/apply', { token: 't', mode: 'trash' }]]);
+    assert.deepEqual(toasts, ['Delete images (move to the trash) started.']);
+    const refused = await applyDelete({ api: { post: async () => { throw Object.assign(new Error('Busy'), { code: 'busy', running: null }); } },
+        store, plan: plan(), mode: 'library', notify: text => toasts.push(text) });
+    assert.equal(refused, null);
+});
+
+test('apply refuses to send a delete without an explicit mode', async () => {
+    const { applyDelete } = module();
+    const api = { post: () => assert.fail('no request without a mode') };
+    for (const mode of [undefined, null, '', 'TRASH', 'delete']) {
+        await assert.rejects(applyDelete({ api, store: null, plan: plan(), mode, notify: () => {} }), TypeError);
+    }
+});

tests/run_suite.py

@@ -28,6 +28,10 @@ stack shows which). A dump after the summary is a hang at interpreter exit
 
 A limit is a measurement, not a dial: change it only with new measurements
 written next to it (TESTING.md "Time limits and hangs").
+
+PM_TEST_SLOWEST=N (opt-in, #487) lists the N slowest tests after the summary.
+A test's time runs from its ``start:`` line to the next one, so it includes
+the class and module fixtures in between, like the hang limit.
 """
 
 import argparse
@@ -36,6 +40,7 @@ import os
 import shutil
 import sys
 import tempfile
+import time
 import unittest
 from pathlib import Path
 
@@ -63,23 +68,57 @@ def read_limit(name, default):
     return int(text)
 
 
-def make_result_class(stream, per_test_limit, shutdown_limit):
+class TestClock:
+    """Seconds from each test's start to the next start (or the run's end)."""
+
+    def __init__(self, clock=time.perf_counter):
+        self._clock = clock
+        self._current = None
+        self._started = 0.0
+        self.durations = []
+
+    def start(self, test_id):
+        self.stop()
+        self._current = test_id
+        self._started = self._clock()
+
+    def stop(self):
+        if self._current is not None:
+            self.durations.append((self._clock() - self._started, self._current))
+            self._current = None
+
+    def report(self, count):
+        """Return the lines that list the ``count`` slowest tests."""
+        slowest = sorted(self.durations, key=lambda item: item[0], reverse=True)
+        lines = [f"slowest {min(count, len(slowest))} of {len(slowest)} tests:"]
+        lines += [f"{seconds:8.2f}s {test_id}" for seconds, test_id in slowest[:count]]
+        return lines
+
+
+def make_result_class(stream, per_test_limit, shutdown_limit, slowest=0):
     """Build a TextTestResult that names each test and re-arms the watchdog."""
 
     def arm(seconds):
         faulthandler.dump_traceback_later(seconds, exit=True, file=stream)
 
+    clock = TestClock()
+
     class WatchdogResult(unittest.TextTestResult):
         def startTest(self, test):
             # At default verbosity the progress dots share a line; start ours.
             prefix = "\n" if self.dots else ""
             stream.write(f"{prefix}start: {test.id()}\n")
             stream.flush()
+            clock.start(test.id())
             super().startTest(test)
             arm(per_test_limit)
 
         def stopTestRun(self):
+            clock.stop()
             super().stopTestRun()
+            if slowest:
+                stream.write("\n".join(["", *clock.report(slowest), ""]))
+                stream.flush()
             arm(shutdown_limit)
 
     return WatchdogResult
@@ -138,6 +177,7 @@ def main(argv=None):
     try:
         per_test_limit = read_limit("PM_TEST_HANG_LIMIT", DEFAULT_PER_TEST_LIMIT)
         shutdown_limit = read_limit("PM_TEST_SHUTDOWN_LIMIT", DEFAULT_SHUTDOWN_LIMIT)
+        slowest = read_limit("PM_TEST_SLOWEST", 0)
     except LimitError as error:
         stream.write(f"{error}\n")
         return USAGE_EXIT
@@ -152,7 +192,9 @@ def main(argv=None):
             verbosity=args.verbosity,
             # As unittest.main does: show warnings unless -W/PYTHONWARNINGS set them.
             warnings=None if sys.warnoptions else "default",
-            resultclass=make_result_class(stream, per_test_limit, shutdown_limit),
+            resultclass=make_result_class(
+                stream, per_test_limit, shutdown_limit, slowest
+            ),
         )
         result = runner.run(suite)
     finally:

tests/test_api_core.py

@@ -684,132 +684,6 @@ class TestStaticRoutes(CoreRoutesTestCase):
         self.assertEqual(resp.status, 403)
 
 
-class TestRevalidatedCaching(CoreRoutesTestCase):
-    """Every served page and static file is revalidated on reload (#478).
-
-    ``Cache-Control: no-cache`` makes the browser ask before it reuses a copy,
-    and the ETag lets an unchanged file answer 304 without a body, so an update
-    reaches the browser on a normal reload and an unchanged reload stays cheap.
-    """
-
-    STATIC_PATHS = (
-        "/prompt_manager/js/workspace/index.mjs",
-        "/prompt_manager/js/admin.js",
-        "/prompt_manager/js/inspection-semantics.json",
-        "/prompt_manager/lib/viewerjs/viewer.min.css",
-    )
-
-    async def _assert_revalidated(self, path):
-        resp = await self.client.request("GET", path)
-        self.assertEqual(resp.status, 200, path)
-        self.assertEqual(resp.headers.get("Cache-Control"), "no-cache", path)
-        etag = resp.headers.get("ETag")
-        self.assertTrue(etag, path)
-        body = await resp.read()
-        self.assertTrue(body, path)
-
-        again = await self.client.request("GET", path, headers={"If-None-Match": etag})
-        self.assertEqual(again.status, 304, path)
-        self.assertEqual(again.headers.get("ETag"), etag, path)
-        self.assertEqual(again.headers.get("Cache-Control"), "no-cache", path)
-        self.assertEqual(await again.read(), b"", path)
-
-        stale = await self.client.request(
-            "GET", path, headers={"If-None-Match": 'W/"stale"'}
-        )
-        self.assertEqual(stale.status, 200, path)
-        self.assertEqual(await stale.read(), body, path)
-        return etag
-
-    async def test_static_files_are_revalidated(self):
-        for path in self.STATIC_PATHS:
-            await self._assert_revalidated(path)
-
-    async def test_pages_are_revalidated(self):
-        for path in UI_PATHS:
-            await self._assert_revalidated(path)
-
-    async def test_etag_matches_in_a_list_and_as_wildcard(self):
-        path = "/prompt_manager/js/workspace/index.mjs"
-        etag = (await self.client.request("GET", path)).headers["ETag"]
-        for header in (f'W/"other", {etag}', etag.removeprefix("W/"), "*"):
-            resp = await self.client.request(
-                "GET", path, headers={"If-None-Match": header}
-            )
-            self.assertEqual(resp.status, 304, header)
-
-    async def test_changed_file_gets_a_new_etag_and_its_new_body(self):
-        web_dir = self._fake_root()
-        module = web_dir / "js" / "m.mjs"
-        module.write_text("export const v = 1;")
-        path = "/prompt_manager/js/m.mjs"
-        first = await self.client.request("GET", path)
-        etag = first.headers["ETag"]
-        # Same size and, on coarse clocks, the same mtime: only the bytes differ.
-        module.write_text("export const v = 2;")
-        resp = await self.client.request("GET", path, headers={"If-None-Match": etag})
-        self.assertEqual(resp.status, 200)
-        self.assertNotEqual(resp.headers["ETag"], etag)
-        self.assertEqual(await resp.text(), "export const v = 2;")
-
-    async def test_gzipped_and_plain_bodies_share_a_weak_etag(self):
-        """The gzip middleware changes the bytes, so the tag must be weak."""
-        resp = await self.client.request("GET", "/prompt_manager/js/admin.js")
-        self.assertTrue(resp.headers["ETag"].startswith('W/"'))
-
-    async def test_errors_carry_no_validator(self):
-        resp = await self.client.request("GET", "/prompt_manager/js/nope.js")
-        self.assertEqual(resp.status, 404)
-        self.assertNotIn("ETag", resp.headers)
-
-
-@web.middleware
-async def _comfyui_cache_control(request, handler):
-    """ComfyUI's own middleware (middleware/cache_middleware.py), JS/CSS branch.
-
-    It only fills a missing Cache-Control, and ``.mjs`` is not ``.js``: that is
-    why the workspace modules went out with no header at all before #478.
-    """
-    response = await handler(request)
-    if request.path.endswith(".js") or request.path.endswith(".css"):
-        response.headers.setdefault("Cache-Control", "no-store")
-    return response
-
-
-class TestRevalidatedCachingBehindMiddlewares(CoreRoutesTestCase):
-    """The host's middlewares keep the header and the 304 intact."""
-
-    async def get_application(self):
-        app = await super().get_application()
-        app.middlewares.append(_comfyui_cache_control)
-        app.middlewares.append(api_module._gzip_middleware)
-        return app
-
-    async def test_gzipped_files_keep_no_cache_and_answer_304(self):
-        for path in (
-            "/prompt_manager/js/workspace/index.mjs",
-            "/prompt_manager/js/admin.js",
-            "/prompt_manager/lib/viewerjs/viewer.min.css",
-            "/prompt_manager/workspace",
-        ):
-            resp = await self.client.request(
-                "GET", path, headers={"Accept-Encoding": "gzip"}
-            )
-            self.assertEqual(resp.status, 200, path)
-            self.assertEqual(resp.headers.get("Content-Encoding"), "gzip", path)
-            self.assertEqual(resp.headers["Cache-Control"], "no-cache", path)
-            again = await self.client.request(
-                "GET",
-                path,
-                headers={
-                    "Accept-Encoding": "gzip",
-                    "If-None-Match": resp.headers["ETag"],
-                },
-            )
-            self.assertEqual(again.status, 304, path)
-            self.assertEqual(again.headers["Cache-Control"], "no-cache", path)
-
-
 # ── instance helpers ──────────────────────────────────────────────────
 
 

tests/test_api_diagnostics_report.py

@@ -45,7 +45,9 @@ SEEDED_PATHS = (
     "/srv/someone/renders",
     "C:\\Users\\someone\\Documents\\renders",
     "D:/Users/someone/prompts.db",
-    "\\\\nas\\share\\someone\\renders",
+    # A share on this machine: on Windows the report resolves every source,
+    # and an unknown server name made that wait about 2.9 s (#487).
+    "\\\\localhost\\pm-no-such-share\\someone\\renders",
 )
 _WINDOWS_ABSOLUTE = re.compile(r"^[A-Za-z]:[\\/]")
 TOP_KEYS = {

tests/test_api_history.py

@@ -25,6 +25,7 @@ try:
 except ImportError:
     from tests.test_api_images import ImageAPITestCase
 
+from database import restore_stage
 from database.operations import PromptDatabase
 from py.api.library import _encode, _scope
 from tests.test_execution_context_database import bundle, snapshot
@@ -43,6 +44,21 @@ OTHER_SESSION = "aaaaaaaa-3333-4333-8333-333333333333"
 MISSING_RUN = "bbbbbbbb-4444-4444-8444-444444444444"
 
 
+@contextmanager
+def no_wal_truncate_wait():
+    """Restore with the restore notification held back, as these tests need.
+
+    The read under test keeps its snapshot open across the restore, so the
+    publish's WAL truncate can never succeed; without the short wait it would
+    sit out the full CHECKPOINT_BUSY_MS (30 s) each time (#487).
+    """
+    with (
+        patch("database.execution_history.notify_history_restore"),
+        patch.object(restore_stage, "CHECKPOINT_BUSY_MS", 50),
+    ):
+        yield
+
+
 class HistoryAPITests(ImageAPITestCase):
     async def tearDownAsync(self):
         if hasattr(self.api, "close_history_reads"):
@@ -346,7 +362,7 @@ class HistoryAPITests(ImageAPITestCase):
 
         def during_snapshot(*args, **kwargs):
             result = original(*args, **kwargs)
-            with patch("database.execution_history.notify_history_restore"):
+            with no_wal_truncate_wait():
                 writer.model.restore_from_file(source)
             return result
 
@@ -890,7 +906,7 @@ class HistoryAPITests(ImageAPITestCase):
 
                 def during_snapshot(*args, **kwargs):
                     result = original(*args, **kwargs)
-                    with patch("database.execution_history.notify_history_restore"):
+                    with no_wal_truncate_wait():
                         writer.model.restore_from_file(source)
                     return result
 

tests/test_history_read_fence.py

@@ -7,6 +7,7 @@ from contextlib import closing
 from pathlib import Path
 from unittest.mock import patch
 
+from database import restore_stage
 from database.execution_history import HistoryEpochChanged, notify_history_restore
 from tests import test_execution_history_database as history_tests
 
@@ -41,7 +42,12 @@ class HistoryReadFenceTests(unittest.TestCase):
         generation = fence.begin()
         with reader.recorded_history_snapshot() as epoch:
             original = reader.get_history_run(start.run_id)
-            with patch("database.execution_history.notify_history_restore"):
+            # The open snapshot keeps the WAL from being truncated; do not
+            # sit out the full 30 s checkpoint wait for it (#487).
+            with (
+                patch("database.execution_history.notify_history_restore"),
+                patch.object(restore_stage, "CHECKPOINT_BUSY_MS", 50),
+            ):
                 writer.model.restore_from_file(source)
             self.assertEqual(reader.get_execution_epoch(), epoch)
             self.assertEqual(original["run"]["storage_epoch"], epoch)

tests/test_image_delete_routes.py

@@ -0,0 +1,334 @@
+"""Delete selected images: the review and the apply job (#480).
+
+``POST /prompt_manager/images/delete/preview`` reviews both modes;
+``POST /prompt_manager/maintenance/apply {token, mode}`` removes the files'
+records (library mode, with a removal mark) or moves the files to the trash
+first (trash mode, the fake trash here). No path unlinks a media file.
+"""
+
+import asyncio
+import os
+import sys
+import sqlite3
+from pathlib import Path
+from unittest.mock import patch
+
+try:
+    from tests import fake_trash
+    from tests.disk_waits import DISK_BOUND_SECONDS
+    from tests.test_api_duplicate_files import DuplicateFilesTestCase
+except ImportError:  # discovered with tests/ as the top-level directory
+    import fake_trash
+    from disk_waits import DISK_BOUND_SECONDS
+    from test_api_duplicate_files import DuplicateFilesTestCase
+
+from py.api import image_delete
+from py.api.image_delete import file_check, lexical_problem
+
+PREVIEW = "/prompt_manager/images/delete/preview"
+APPLY = "/prompt_manager/maintenance/apply"
+LISTING = "/prompt_manager/images/output"
+
+
+class ImageDeleteTestCase(DuplicateFilesTestCase):
+    async def get_application(self):
+        app = await super().get_application()
+        self.fake = fake_trash.install(Path(self.tmpdir) / "fake-trash")
+        self.addCleanup(fake_trash.uninstall)
+        return app
+
+    def target(self, path, index=0):
+        roots = self.api._get_all_output_dirs()
+        return {
+            "root_index": index,
+            "root_scope": self.api._gallery_root_scope(roots[index], index),
+            "path": path,
+        }
+
+    async def review(self, paths, status=200):
+        items = [p if isinstance(p, dict) else self.target(p) for p in paths]
+        return await self.post(PREVIEW, {"items": items}, status)
+
+    async def delete(self, paths, mode):
+        plan = await self.review(paths)
+        body = await self.post(APPLY, {"token": plan["token"], "mode": mode}, 202)
+        job = self.api._jobs.get(body["job"]["id"])
+        await asyncio.wait_for(asyncio.shield(job.task), DISK_BOUND_SECONDS)
+        return plan, job
+
+    async def listed(self):
+        response = await self.client.get(LISTING, params={"mode": "workspace"})
+        page = await response.json()
+        return [image["path"] for image in page["images"]]
+
+    def link(self, name, text="a prompt"):
+        prompt_id = self.save(text)
+        return self.api.db.link_image_to_prompt(
+            str(prompt_id), str(self.output_dir / name)
+        )
+
+    def count(self, sql):
+        conn = sqlite3.connect(self.api.db.model.db_path)
+        try:
+            return conn.execute(sql).fetchone()[0]
+        finally:
+            conn.close()
+
+    @staticmethod
+    def details(job, group):
+        return [row["detail"] for row in job.detail()["outcomes"][group]]
+
+
+class TestReview(ImageDeleteTestCase):
+    async def test_review_counts_records_and_refuses_what_it_cannot_delete(self):
+        self.write("a.png", b"aaa", mtime=1_000_000)
+        self.write("b.png", b"bbb", mtime=2_000_000)
+        self.link("a.png", "only a")
+        self.fake.refuse[str(self.output_dir / "b.png")] = "no-trash"
+        plan = await self.review(["a.png", "b.png", "missing.png"])
+        self.assertEqual(plan["library"], {"count": 2})
+        self.assertEqual(plan["trash"], {"count": 1})
+        self.assertEqual(plan["refused"]["counts"], {"not-listed": 1})
+        self.assertEqual(plan["trash_refused"]["counts"], {"no-trash": 1})
+        self.assertEqual(plan["records"]["prompt_links"], 1)
+        self.assertEqual(plan["orphaned_prompts"]["count"], 1)
+        self.assertIn(plan["backup"]["policy"], ("on", "off"))
+        self.assertNotIn(self.tmpdir, str(plan))
+        # The review wrote nothing.
+        self.assertEqual(self.count("SELECT COUNT(*) FROM generated_images"), 1)
+        self.assertEqual(self.fake.moved, [])
+
+    async def test_body_shapes_and_the_cap(self):
+        await self.post(PREVIEW, {}, 400)
+        await self.post(PREVIEW, {"items": [], "mode": "trash"}, 400)
+        too_many = [self.target(f"{n}.png") for n in range(501)]
+        await self.post(PREVIEW, {"items": too_many}, 400)
+
+    async def test_mode_is_required_and_only_for_this_plan(self):
+        self.write("a.png", b"aaa")
+        self.link("a.png")
+        plan = await self.review(["a.png"])
+        # No explicit, exact mode: refused, never a default (least of all trash).
+        for body in (
+            {"token": plan["token"]},
+            {"token": plan["token"], "mode": None},
+            {"token": plan["token"], "mode": ""},
+            {"token": plan["token"], "mode": "TRASH"},
+            {"token": plan["token"], "mode": "delete"},
+            {"token": plan["token"], "mode": ["trash"]},
+            {"token": plan["token"], "mode": True},
+        ):
+            with self.subTest(body=body):
+                await self.post(APPLY, body, 400)
+        self.assertTrue((self.output_dir / "a.png").exists())
+        self.assertEqual(self.fake.moved, [])
+        self.assertEqual(self.count("SELECT COUNT(*) FROM generated_images"), 1)
+        # The refusals left the token usable: an explicit mode still applies.
+        body = await self.post(APPLY, {"token": plan["token"], "mode": "library"}, 202)
+        job = self.api._jobs.get(body["job"]["id"])
+        await asyncio.wait_for(asyncio.shield(job.task), DISK_BOUND_SECONDS)
+        self.assertEqual(job.scope["mode"], "library")
+        self.assertTrue((self.output_dir / "a.png").exists())
+        self.assertEqual(self.fake.moved, [])
+
+    async def test_mode_on_another_plan_is_refused(self):
+        plan = await self.post(
+            "/prompt_manager/maintenance/preview",
+            {"operation": "missing_links", "options": {}},
+            200,
+        )
+        await self.post(APPLY, {"token": plan["token"], "mode": "trash"}, 400)
+
+
+class TestLibraryMode(ImageDeleteTestCase):
+    async def test_records_go_and_the_file_stays_hidden(self):
+        self.write("a.png", b"aaa", mtime=1_000_000)
+        self.write("b.png", b"bbb", mtime=2_000_000)
+        self.link("a.png")
+        self.assertEqual(sorted(await self.listed()), ["a.png", "b.png"])
+        before = os.stat(self.output_dir / "a.png")
+        _plan, job = await self.delete(["a.png"], "library")
+        self.assertEqual(job.state, "finished", job.summary())
+        self.assertEqual(self.count("SELECT COUNT(*) FROM generated_images"), 0)
+        after = os.stat(self.output_dir / "a.png")
+        self.assertEqual(
+            (before.st_size, before.st_mtime_ns), (after.st_size, after.st_mtime_ns)
+        )
+        self.assertEqual(await self.listed(), ["b.png"])
+        # A later scan or link never brings it back.
+        self.assertEqual(self.link("a.png", "rescanned"), 0)
+        self.assertEqual(
+            self.api._removed_paths([self.output_dir / "a.png"]),
+            {str(self.output_dir / "a.png")},
+        )
+        self.assertEqual(self.fake.moved, [])
+
+    async def test_a_modified_file_shows_again(self):
+        self.write("a.png", b"aaa", mtime=1_000_000)
+        await self.delete(["a.png"], "library")
+        self.assertEqual(await self.listed(), [])
+        self.write("a.png", b"new and longer", mtime=3_000_000)
+        self.api.invalidate_gallery_cache()
+        self.assertEqual(await self.listed(), ["a.png"])
+
+    async def test_a_safety_backup_is_taken_first(self):
+        self.write("a.png", b"aaa")
+        with patch.object(self.api, "_safety_backup_policy", return_value="on"):
+            await self.delete(["a.png"], "library")
+        self.assertEqual(len(self.backups()), 1)
+
+
+class TestTrashMode(ImageDeleteTestCase):
+    async def test_the_file_moves_first_then_its_records_go(self):
+        self.write("a.png", b"aaa")
+        self.link("a.png")
+        seen = {}
+        original = self.fake.move
+
+        def move(path, roots):
+            # The records still exist while the file moves.
+            seen["links"] = self.count("SELECT COUNT(*) FROM generated_images")
+            return original(path, roots)
+
+        self.fake.move = move
+        _plan, job = await self.delete(["a.png"], "trash")
+        self.assertEqual(job.state, "finished", job.summary())
+        self.assertEqual(seen["links"], 1)
+        self.assertFalse((self.output_dir / "a.png").exists())
+        [(source, destination)] = self.fake.moved
+        self.assertTrue(os.path.exists(destination))
+        self.assertEqual(self.count("SELECT COUNT(*) FROM generated_images"), 0)
+        self.assertEqual(self.api.db.removal_marks(), {})
+
+    async def test_a_refused_move_changes_nothing(self):
+        self.write("a.png", b"aaa")
+        self.link("a.png")
+        plan = await self.review(["a.png"])
+        self.fake.refuse[str(self.output_dir / "a.png")] = "locked"
+        body = await self.post(APPLY, {"token": plan["token"], "mode": "trash"}, 202)
+        job = self.api._jobs.get(body["job"]["id"])
+        await asyncio.wait_for(asyncio.shield(job.task), DISK_BOUND_SECONDS)
+        self.assertTrue((self.output_dir / "a.png").exists())
+        self.assertEqual(self.count("SELECT COUNT(*) FROM generated_images"), 1)
+        self.assertEqual(self.api.db.removal_marks(), {})
+        self.assertEqual(job.counters["not_changed"], 1)
+        self.assertEqual(
+            self.details(job, "not_changed"), [image_delete.REASONS["locked"]]
+        )
+
+    async def test_a_file_changed_after_the_review_is_not_changed(self):
+        self.write("a.png", b"aaa", mtime=1_000_000)
+        plan = await self.review(["a.png"])
+        self.write("a.png", b"changed bytes", mtime=2_000_000)
+        body = await self.post(APPLY, {"token": plan["token"], "mode": "trash"}, 202)
+        job = self.api._jobs.get(body["job"]["id"])
+        await asyncio.wait_for(asyncio.shield(job.task), DISK_BOUND_SECONDS)
+        self.assertTrue((self.output_dir / "a.png").exists())
+        self.assertEqual(self.fake.moved, [])
+        self.assertEqual(
+            self.details(job, "not_changed"), [image_delete.REASONS["changed"]]
+        )
+
+    async def test_a_records_failure_stops_the_job(self):
+        self.write("a.png", b"aaa", mtime=1_000_000)
+        self.write("b.png", b"bbb", mtime=2_000_000)
+        with patch.object(
+            type(self.api.db),
+            "remove_image_records",
+            side_effect=sqlite3.OperationalError("locked"),
+        ):
+            _plan, job = await self.delete(["a.png", "b.png"], "library")
+        self.assertEqual(job.state, "failed")
+        self.assertEqual(job.scope.get("failure"), "records-failed")
+        self.assertEqual(job.counters["errors"], 1)
+        self.assertEqual(
+            self.details(job, "not_changed"), [image_delete.REASONS["not-attempted"]]
+        )
+
+    async def test_nothing_is_ever_unlinked(self):
+        self.write("a.png", b"aaa")
+        self.write("b.png", b"bbb")
+        media = str(self.output_dir)
+        real_remove, real_unlink = os.remove, os.unlink
+
+        def guard(real):
+            def call(path, *args, **kwargs):
+                text = os.path.realpath(str(path))
+                if (
+                    text.startswith(os.path.realpath(media))
+                    and "thumbnails" not in text
+                ):
+                    raise AssertionError(f"unlinked a media file: {path}")
+                return real(path, *args, **kwargs)
+
+            return call
+
+        with (
+            patch("os.remove", guard(real_remove)),
+            patch("os.unlink", guard(real_unlink)),
+        ):
+            await self.delete(["a.png"], "trash")
+            await self.delete(["b.png"], "library")
+        self.assertTrue((self.output_dir / "b.png").exists())
+
+
+class TestFileCheck(ImageDeleteTestCase):
+    def test_lexical_codes(self):
+        self.assertEqual(lexical_problem("thumbnails/a.png"), "thumbnail")
+        self.assertEqual(lexical_problem(".Trash-1000/files/a.png"), "trash-folder")
+        self.assertEqual(lexical_problem("x/$RECYCLE.BIN/a.png"), "trash-folder")
+        self.assertEqual(lexical_problem("a.txt"), "not-media")
+        self.assertIsNone(lexical_problem(".Trashy/a.png"))
+        self.assertIsNone(lexical_problem("sub/a.png"))
+
+    def test_links_and_folders_leading_outside_are_refused(self):
+        root = str(self.output_dir)
+        outside = Path(self.tmpdir) / "elsewhere"
+        outside.mkdir()
+        (outside / "x.png").write_bytes(b"x")
+        try:
+            os.symlink(outside / "x.png", self.output_dir / "link.png")
+            os.symlink(outside, self.output_dir / "linked", target_is_directory=True)
+        except OSError:
+            self.skipTest("symlinks are not permitted here")
+        self.assertEqual(
+            file_check(str(self.output_dir / "link.png"), root, "link.png")[0], "link"
+        )
+        self.assertEqual(
+            file_check(str(self.output_dir / "linked/x.png"), root, "linked/x.png")[0],
+            "outside",
+        )
+        self.assertEqual(
+            file_check(str(self.output_dir / "gone.png"), root, "gone.png")[0], "gone"
+        )
+        self.write("ok.png", b"ok")
+        self.assertIsNone(
+            file_check(str(self.output_dir / "ok.png"), root, "ok.png")[0]
+        )
+
+    def test_reasons_cover_every_code(self):
+        from utils import trash
+
+        for code in trash.CODES:
+            self.assertIn(code, image_delete.REASONS)
+
+
+class TestNativeBackend(ImageDeleteTestCase):
+    """The real backend of this OS: Windows and macOS have no trash yet (#480)."""
+
+    async def test_library_only_works_and_the_trash_is_unavailable_here(self):
+        fake_trash.uninstall()
+        self.write("a.png", b"aaa")
+        self.link("a.png")
+        plan = await self.review(["a.png"])
+        native_trash = sys.platform not in ("win32", "darwin")
+        if not native_trash:
+            self.assertEqual(plan["trash"], {"count": 0})
+            self.assertEqual(plan["trash_refused"]["counts"], {"no-trash": 1})
+        body = await self.post(APPLY, {"token": plan["token"], "mode": "library"}, 202)
+        job = self.api._jobs.get(body["job"]["id"])
+        await asyncio.wait_for(asyncio.shield(job.task), DISK_BOUND_SECONDS)
+        self.assertEqual(job.counters["removed"], 1, job.summary())
+        self.assertTrue((self.output_dir / "a.png").exists())
+        self.assertEqual(self.count("SELECT COUNT(*) FROM generated_images"), 0)
+        self.assertEqual(await self.listed(), [])

tests/test_image_removal.py

@@ -0,0 +1,319 @@
+"""Removing one file's records from the library, and removal marks (#480)."""
+
+import hashlib
+import os
+import sqlite3
+import tempfile
+import unittest
+from pathlib import Path
+
+try:
+    from tests.open_handles import assert_closed
+except ImportError:  # discovered with tests/ as the top-level directory
+    from open_handles import assert_closed
+
+from database.image_removal_schema import (
+    image_removals_schema_error,
+    migrate_image_removals,
+)
+from database.models import PromptModel
+from database.operations import PromptDatabase
+
+
+def _locator(path):
+    return os.path.normcase(os.path.realpath(str(path)))
+
+
+class _RemovalCase(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temp.cleanup)
+        self.addCleanup(assert_closed, self, self.temp.name)
+        self.addCleanup(PromptModel.reset_schema_cache)
+        base = Path(self.temp.name)
+        self.root = base / "output"
+        self.root.mkdir()
+        # A second spelling of the folder (a symlink); Windows without the
+        # privilege has none, and the records then carry one spelling.
+        self.alias = base / "alias"
+        try:
+            self.alias.symlink_to(self.root, target_is_directory=True)
+        except OSError:
+            self.alias = None
+        self.db = PromptDatabase(str(base / "prompts.db"))
+        self.addCleanup(self.db.close_all)
+        self.epoch = self.db.get_execution_epoch()
+
+    def file(self, name, data=b"image-bytes"):
+        path = self.root / name
+        path.write_bytes(data)
+        return path
+
+    def removal(self, path):
+        root_key = os.path.normcase(os.path.realpath(str(self.root)))
+        return {
+            "spellings": [str(path)]
+            + ([str(self.alias / path.name)] if self.alias else []),
+            "locator": _locator(path),
+            "favorites": [(root_key, path.name.lower())],
+            "category": (root_key, path.name.lower()),
+        }
+
+    def link(self, path, text="a prompt", spelled=None):
+        prompt_id = self.db.save_prompt(text)
+        image_id = self.db.link_image_to_prompt(str(prompt_id), str(spelled or path))
+        self.assertTrue(image_id)
+        return prompt_id, image_id
+
+    def tag(self, path, tags):
+        data = path.read_bytes()
+        info = os.stat(path)
+        return self.db.replace_image_tags(
+            locator=_locator(path),
+            sha256=hashlib.sha256(data).hexdigest(),
+            size=info.st_size,
+            mtime_ns=info.st_mtime_ns,
+            expected_epoch=self.epoch,
+            expected_tag_revision=0,
+            tags=tags,
+        )
+
+    def queue(self, path):
+        data = path.read_bytes()
+        info = os.stat(path)
+        return self.db.record_review_item(
+            locator=_locator(path),
+            spelled_path=os.path.realpath(str(path)),
+            sha256=hashlib.sha256(data).hexdigest(),
+            size=info.st_size,
+            mtime_ns=info.st_mtime_ns,
+            epoch=self.epoch,
+            resolver_version=1,
+            evidence={"candidates": []},
+            candidate_count=0,
+            limited=True,
+        )
+
+    def count(self, sql, *params):
+        conn = sqlite3.connect(self.db.model.db_path)
+        try:
+            return conn.execute(sql, params).fetchone()[0]
+        finally:
+            conn.close()
+
+    def decorate(self, path):
+        """Give *path* one record of every kind."""
+        removal = self.removal(path)
+        root_key, relative = removal["favorites"][0]
+        self.tag(path, ["portrait"])
+        self.db.set_image_favorite(root_key, relative)
+        self.db.set_image_categories([removal["category"]], "keepers")
+        self.db.add_to_collection(items=[(root_key, relative)], name="Set")
+        return removal
+
+
+class ImageRemovalRecordsTests(_RemovalCase):
+    def test_counts_and_removes_every_record_under_every_spelling(self):
+        path = self.file("a.png")
+        other = self.file("b.png", b"other")
+        orphan, _ = self.link(path, "only this image")
+        shared, _ = self.link(path, "two images")
+        self.db.link_image_to_prompt(str(shared), str(other))
+        # A record under the symlinked folder spelling (or a third prompt).
+        self.link(path, "aliased", spelled=(self.alias or self.root) / "a.png")
+        removal = self.decorate(path)
+        self.assertTrue(self.queue(path))
+
+        counted = self.db.count_removal_records([removal])
+        self.assertEqual(
+            counted["records"],
+            {
+                "prompt_links": 3,
+                "pair_links": 0,
+                "image_tags": 1,
+                "auto_tagged": 0,
+                "categories": 1,
+                "collections": 1,
+                "favorites": 1,
+                "review_items": 1,
+            },
+        )
+        self.assertEqual(counted["orphaned_prompts"]["count"], 2)
+
+        removed = self.db.remove_image_records(removal)
+        self.assertEqual(removed["prompt_links"], 3)
+        self.assertEqual(removed["review_items"], 1)
+        self.assertEqual(self.count("SELECT COUNT(*) FROM generated_images"), 1)
+        for table in (
+            "media_image_tags",
+            "media_annotation_files",
+            "media_review_items",
+            "image_favorites",
+            "image_categories",
+            "image_collection_items",
+        ):
+            with self.subTest(table=table):
+                self.assertEqual(self.count(f"SELECT COUNT(*) FROM {table}"), 0)
+        # Prompts and the collection itself stay.
+        self.assertEqual(self.count("SELECT COUNT(*) FROM prompts"), 3)
+        self.assertEqual(self.count("SELECT COUNT(*) FROM image_collections"), 1)
+        self.assertTrue(path.exists())
+        self.assertIsNotNone(self.db.get_prompt_by_id(orphan))
+
+    def test_a_kept_review_item_keeps_the_annotation_rows(self):
+        path = self.file("a.png")
+        removal = self.decorate(path)
+        self.assertTrue(self.queue(path))
+        conn = sqlite3.connect(self.db.model.db_path)
+        try:
+            # A revised item is not settled; it still cites the bytes, whose
+            # foreign key is ON DELETE RESTRICT.
+            conn.execute("UPDATE media_review_items SET revision=1")
+            conn.commit()
+        finally:
+            conn.close()
+        removed = self.db.remove_image_records(removal)
+        self.assertEqual(removed["review_items"], 0)
+        self.assertEqual(removed["image_tags"], 1)
+        self.assertEqual(self.count("SELECT COUNT(*) FROM media_image_tags"), 0)
+        self.assertEqual(self.count("SELECT COUNT(*) FROM media_review_items"), 1)
+        self.assertEqual(self.count("SELECT COUNT(*) FROM media_annotation_files"), 1)
+
+    def test_mark_is_written_with_the_records_and_bumps_the_revision(self):
+        path = self.file("a.png")
+        self.link(path)
+        before = self.db.removal_revision()
+        info = os.stat(path)
+        self.db.remove_image_records(
+            self.removal(path), mark=(info.st_size, info.st_mtime_ns)
+        )
+        self.assertEqual(
+            self.db.removal_marks(),
+            {_locator(path): (info.st_size, info.st_mtime_ns)},
+        )
+        self.assertEqual(self.db.removal_revision(), before + 1)
+        self.assertEqual(self.db.clear_removal_marks([_locator(path)]), 1)
+        self.assertEqual(self.db.removal_marks(), {})
+        self.assertEqual(self.db.removal_revision(), before + 2)
+        self.assertEqual(self.db.clear_removal_marks([_locator(path)]), 0)
+
+    def test_a_failure_rolls_back_everything(self):
+        path = self.file("a.png")
+        self.link(path)
+        removal = self.decorate(path)
+        broken = {**removal, "category": ("only-one-value",)}
+        with self.assertRaises(sqlite3.Error):
+            self.db.remove_image_records(broken, mark=(1, 1))
+        self.assertEqual(self.count("SELECT COUNT(*) FROM generated_images"), 1)
+        self.assertEqual(self.count("SELECT COUNT(*) FROM media_image_tags"), 1)
+        self.assertEqual(self.db.removal_marks(), {})
+
+    def test_parity_with_the_legacy_single_delete(self):
+        path = self.file("a.png")
+        _prompt, image_id = self.link(path)
+        twin = PromptDatabase(str(Path(self.temp.name) / "twin.db"))
+        try:
+            twin_prompt = twin.save_prompt("a prompt")
+            twin_image = twin.link_image_to_prompt(str(twin_prompt), str(path))
+            self.assertTrue(twin.delete_image(twin_image))
+            self.db.remove_image_records(self.removal(path))
+            for table in (
+                "generated_images",
+                "image_prompt_pairs",
+                "workflow_documents",
+            ):
+                with self.subTest(table=table):
+                    conn = sqlite3.connect(twin.model.db_path)
+                    try:
+                        legacy = conn.execute(
+                            f"SELECT COUNT(*) FROM {table}"
+                        ).fetchone()[0]
+                    finally:
+                        conn.close()
+                    self.assertEqual(
+                        self.count(f"SELECT COUNT(*) FROM {table}"), legacy
+                    )
+        finally:
+            twin.close_all()
+        self.assertIsNotNone(image_id)
+
+
+class RemovalMarkGateTests(_RemovalCase):
+    """A marked file is never linked again; any record write shows it again."""
+
+    def hide(self, path):
+        info = os.stat(path)
+        self.db.remove_image_records(
+            self.removal(path), mark=(info.st_size, info.st_mtime_ns)
+        )
+
+    def test_a_marked_file_is_not_linked_again(self):
+        path = self.file("a.png")
+        self.hide(path)
+        prompt_id = self.db.save_prompt("scanned again")
+        self.assertEqual(self.db.link_image_to_prompt(str(prompt_id), str(path)), 0)
+        self.assertEqual(self.count("SELECT COUNT(*) FROM generated_images"), 0)
+        self.assertIn(_locator(path), self.db.removal_marks())
+
+    def test_a_changed_file_drops_its_stale_mark_and_links(self):
+        path = self.file("a.png")
+        self.hide(path)
+        path.write_bytes(b"new bytes, new size")
+        prompt_id = self.db.save_prompt("a new image at the same path")
+        self.assertTrue(self.db.link_image_to_prompt(str(prompt_id), str(path)))
+        self.assertEqual(self.db.removal_marks(), {})
+
+    def test_explicit_record_writes_clear_the_mark(self):
+        writers = {
+            "favorite": lambda path, key: self.db.set_image_favorite(*key),
+            "category": lambda path, key: self.db.set_image_categories([key], "c"),
+            "collection": lambda path, key: self.db.add_to_collection(
+                items=[key], name="Set"
+            ),
+            "tags": lambda path, key: self.tag(path, ["t"]),
+        }
+        for kind, write in writers.items():
+            with self.subTest(kind=kind):
+                path = self.file(f"{kind}.png", kind.encode())
+                self.hide(path)
+                self.assertIn(_locator(path), self.db.removal_marks())
+                write(path, self.removal(path)["favorites"][0])
+                self.assertNotIn(_locator(path), self.db.removal_marks())
+
+
+class ImageRemovalSchemaTests(unittest.TestCase):
+    def setUp(self):
+        self.conn = sqlite3.connect(":memory:")
+        self.addCleanup(self.conn.close)
+
+    def test_missing_is_allowed_only_for_old_backups(self):
+        self.assertIsNone(image_removals_schema_error(self.conn))
+        self.assertIsNotNone(
+            image_removals_schema_error(self.conn, allow_missing=False)
+        )
+
+    def test_install_is_valid_and_idempotent(self):
+        migrate_image_removals(self.conn)
+        migrate_image_removals(self.conn)
+        self.assertIsNone(image_removals_schema_error(self.conn, allow_missing=False))
+        self.assertEqual(
+            self.conn.execute("SELECT revision FROM image_removals_meta").fetchall(),
+            [(0,)],
+        )
+
+    def test_invalid_values_and_objects_are_refused(self):
+        migrate_image_removals(self.conn)
+        self.conn.execute("INSERT INTO image_removals VALUES ('', 1, 1, 'now')")
+        self.assertIn("invalid values", image_removals_schema_error(self.conn))
+        self.conn.execute("DELETE FROM image_removals")
+        self.conn.execute("CREATE INDEX extra ON image_removals(size)")
+        self.assertIn("unsupported", image_removals_schema_error(self.conn))
+
+    def test_a_foreign_table_definition_is_refused(self):
+        self.conn.execute("CREATE TABLE image_removals (locator TEXT)")
+        with self.assertRaises(sqlite3.DatabaseError):
+            migrate_image_removals(self.conn)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_lora_manager_layout.py

@@ -14,6 +14,7 @@ import sys
 import tempfile
 import unittest
 from pathlib import Path
+from types import SimpleNamespace
 from unittest.mock import patch
 
 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
@@ -753,6 +754,17 @@ class LocalMediaTests(LayoutTestCase):
         self.assertEqual(layout_module.VIDEO_SUFFIXES, frozenset(VIDEO_EXTENSIONS))
 
 
+# Windows path rules without the filesystem. On Windows ntpath.realpath asks
+# the network for \\server\share and waits seconds for a host that does not
+# exist (#487); elsewhere ntpath.realpath already is ntpath.abspath.
+LEXICAL_NT = SimpleNamespace(
+    normcase=ntpath.normcase,
+    realpath=ntpath.abspath,
+    dirname=ntpath.dirname,
+    sep=ntpath.sep,
+)
+
+
 class FolderGuardTests(unittest.TestCase):
     """Portable path-flavor controls (TESTING.md Windows portability preflight)."""
 
@@ -762,7 +774,7 @@ class FolderGuardTests(unittest.TestCase):
 
     def guard_nt(self, path, home="C:\\Users\\Vito"):
         return layout_module._guard_folder(
-            path, home, pathmod=ntpath, isdir=lambda _p: True
+            path, home, pathmod=LEXICAL_NT, isdir=lambda _p: True
         )
 
     def test_filesystem_roots_and_home_are_rejected(self):
@@ -832,8 +844,12 @@ class FolderGuardTests(unittest.TestCase):
 
     def test_missing_folders_add_nothing(self):
         self.assertIsNone(layout_module._guard_folder(str(self.tmp / "x"), "/home"))
+        # A missing share on this machine: an unknown server name made
+        # Windows wait about 2.8 s for the network (#487).
         self.assertIsNone(
-            layout_module._guard_folder("\\\\server\\share\\loras", str(self.tmp))
+            layout_module._guard_folder(
+                "\\\\localhost\\pm-no-such-share\\loras", str(self.tmp)
+            )
         )
         self.assertEqual(
             layout_module._guard_folder(str(self.tmp), "/nonexistent-home"),
@@ -843,17 +859,17 @@ class FolderGuardTests(unittest.TestCase):
     def test_windows_model_path_matches_a_lower_case_root(self):
         self.assertTrue(
             layout_module._inside(
-                "C:/Models/Loras/A.safetensors", ("c:\\models\\loras",), ntpath
+                "C:/Models/Loras/A.safetensors", ("c:\\models\\loras",), LEXICAL_NT
             )
         )
         self.assertFalse(
             layout_module._inside(
-                "C:/Models/LorasX/A.safetensors", ("c:\\models\\loras",), ntpath
+                "C:/Models/LorasX/A.safetensors", ("c:\\models\\loras",), LEXICAL_NT
             )
         )
         self.assertFalse(
             layout_module._inside(
-                "\\\\other\\share\\A.safetensors", ("c:\\models\\loras",), ntpath
+                "\\\\other\\share\\A.safetensors", ("c:\\models\\loras",), LEXICAL_NT
             )
         )
 

tests/test_lora_utils.py

@@ -9,6 +9,7 @@ import email
 import json
 import os
 import shutil
+import ssl
 import sys
 import tempfile
 import threading
@@ -669,6 +670,15 @@ class TestStaleDownloadSweep(unittest.TestCase):
         self.assertTrue(old.exists())
 
 
+def _unused_tls_context():
+    """A TLS context for the fake handlers, which never open a connection.
+
+    HTTPSHandler() builds a default context, and on Windows that reads the
+    system certificate store: about 0.14 s per handler (#487).
+    """
+    return ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+
+
 class _FakeHTTPSHandler(urllib.request.HTTPSHandler):
     """Serves canned responses through a real urllib opener, recording requests.
 
@@ -677,7 +687,7 @@ class _FakeHTTPSHandler(urllib.request.HTTPSHandler):
     """
 
     def __init__(self, redirect_to):
-        super().__init__()
+        super().__init__(context=_unused_tls_context())
         self.redirect_to = redirect_to
         self.requests = []
 
@@ -745,7 +755,7 @@ class _ChainHTTPSHandler(urllib.request.HTTPSHandler):
     urllib opener, recording every request; an unknown URL answers 404."""
 
     def __init__(self, routes):
-        super().__init__()
+        super().__init__(context=_unused_tls_context())
         self.routes = routes
         self.requests = []
 

tests/test_run_suite.py

@@ -18,11 +18,13 @@ from pathlib import Path
 REPO_ROOT = Path(__file__).resolve().parent.parent
 SUBPROCESS_TIMEOUT = 60
 WATCHDOG_ENV = ("PM_TEST_HANG_LIMIT", "PM_TEST_SHUTDOWN_LIMIT")
+# Integer settings the runner reads; a bad value is a usage error.
+RUNNER_ENV = (*WATCHDOG_ENV, "PM_TEST_SLOWEST")
 
 
 def run_suite(folder, extra_env=None, *args, verbosity=("-v",)):
     """Run the runner on ``folder`` and return the completed process."""
-    env = {k: v for k, v in os.environ.items() if k not in WATCHDOG_ENV}
+    env = {k: v for k, v in os.environ.items() if k not in RUNNER_ENV}
     # Windows writes a piped stderr in the ANSI codepage; ask for the UTF-8
     # this helper decodes, so non-ASCII values and paths survive.
     env["PYTHONIOENCODING"] = "utf-8"
@@ -269,8 +271,57 @@ class RunSuiteTests(unittest.TestCase):
         self.assertIn("ok", result.stderr, detail)
         self.assertIn("Timeout (", result.stderr, detail)
 
+    def test_slowest_tests_are_listed_when_asked(self):
+        self.write(
+            "test_probe_slowest.py",
+            """
+            import time
+            import unittest
+
+            class Probe(unittest.TestCase):
+                def test_fast(self):
+                    pass
+
+                def test_slow(self):
+                    time.sleep(0.3)
+
+                def test_slower(self):
+                    time.sleep(0.6)
+            """,
+        )
+        result = run_suite(self.folder, {"PM_TEST_SLOWEST": "2"})
+        self.assertEqual(result.returncode, 0, result.stderr)
+        report = result.stderr.split("slowest 2 of 3 tests:\n", 1)
+        self.assertEqual(len(report), 2, result.stderr)
+        lines = [line.split() for line in report[1].strip().splitlines()[:2]]
+        self.assertEqual(
+            [line[1] for line in lines],
+            [
+                "test_probe_slowest.Probe.test_slower",
+                "test_probe_slowest.Probe.test_slow",
+            ],
+            result.stderr,
+        )
+        self.assertGreaterEqual(float(lines[0][0].rstrip("s")), 0.6)
+        self.assertNotIn("test_fast", report[1])
+
+    def test_slowest_tests_are_not_listed_by_default(self):
+        self.write(
+            "test_probe_quiet.py",
+            """
+            import unittest
+
+            class Probe(unittest.TestCase):
+                def test_ok(self):
+                    pass
+            """,
+        )
+        result = run_suite(self.folder)
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertNotIn("slowest", result.stderr)
+
     def test_invalid_limit_is_a_usage_error(self):
-        for name in WATCHDOG_ENV:
+        for name in RUNNER_ENV:
             for value in ("abc", "0", "-3", "1.5", "\u00b2"):
                 with self.subTest(name=name, value=value):
                     result = run_suite(self.folder, {name: value})

tests/test_trash_freedesktop.py

@@ -0,0 +1,298 @@
+"""The freedesktop trash backend (#480): rename into a trash or refuse, never unlink.
+
+Every test builds a disposable tree: a fake ``XDG_DATA_HOME`` and, through the
+backend's ``mounts`` seam, a fake "drive" mount point, so the real trash of the
+user running the tests is never touched. Moves are real renames on one
+filesystem.
+"""
+
+import os
+import shutil
+import stat
+import sys
+import tempfile
+import unittest
+from unittest import mock
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from utils import trash  # noqa: E402
+
+POSIX = os.name == "posix" and sys.platform != "darwin"
+
+
+@unittest.skipUnless(POSIX, "freedesktop trash is the Linux/BSD backend")
+class FreedesktopTrashTests(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.mkdtemp(prefix="pm-trash-")
+        self.addCleanup(shutil.rmtree, self.tmp, True)
+        self.data_home = os.path.join(self.tmp, "xdg")
+        self.drive = os.path.join(self.tmp, "drive")
+        self.root = os.path.join(self.drive, "output")
+        self.home_root = os.path.join(self.tmp, "home-output")
+        for folder in (self.data_home, self.root, self.home_root):
+            os.makedirs(folder)
+        self.mounts = ["/", os.path.realpath(self.drive)]
+        self.backend = trash.FreedesktopBackend(
+            data_home=self.data_home, mounts=lambda: list(self.mounts)
+        )
+
+    def _file(self, folder, name="a.png", content=b"png-bytes"):
+        path = os.path.join(folder, name)
+        with open(path, "wb") as handle:
+            handle.write(content)
+        return path
+
+    def _info(self, trash_dir, name):
+        with open(os.path.join(trash_dir, "info", name + ".trashinfo")) as handle:
+            return handle.read()
+
+    # ── choosing the trash ────────────────────────────────────────────
+
+    def test_file_on_the_home_mount_goes_to_the_home_trash(self):
+        path = self._file(self.home_root)
+        label = self.backend.move(path, [self.home_root])
+        home_trash = os.path.join(self.data_home, "Trash")
+        self.assertEqual(label, trash.HOME_TRASH_LABEL)
+        self.assertFalse(os.path.exists(path))
+        with open(os.path.join(home_trash, "files", "a.png"), "rb") as handle:
+            self.assertEqual(handle.read(), b"png-bytes")
+        info = self._info(home_trash, "a.png")
+        self.assertIn("[Trash Info]\n", info)
+        self.assertIn("Path=" + os.path.realpath(path) + "\n", info)
+        self.assertRegex(info, r"DeletionDate=\d{4}-\d\d-\d\dT\d\d:\d\d:\d\d\n")
+
+    def test_file_on_another_mount_uses_its_trash_dash_uid(self):
+        path = self._file(self.root, "b c%.png")
+        self.backend.move(path, [self.root])
+        top = os.path.join(self.drive, f".Trash-{os.getuid()}")
+        info_mode = stat.S_IMODE(os.lstat(top).st_mode)
+        self.assertEqual(info_mode, 0o700)
+        self.assertTrue(os.path.exists(os.path.join(top, "files", "b c%.png")))
+        # A top-directory trash records the path relative to its top folder,
+        # percent-encoded.
+        self.assertIn("Path=output/b%20c%25.png\n", self._info(top, "b c%.png"))
+
+    def test_sticky_dot_trash_folder_is_used(self):
+        shared = os.path.join(self.drive, ".Trash")
+        os.mkdir(shared)
+        os.chmod(shared, 0o1777)
+        path = self._file(self.root)
+        self.backend.move(path, [self.root])
+        self.assertTrue(
+            os.path.exists(os.path.join(shared, str(os.getuid()), "files", "a.png"))
+        )
+
+    def test_dot_trash_without_sticky_bit_is_skipped(self):
+        shared = os.path.join(self.drive, ".Trash")
+        os.mkdir(shared, 0o777)
+        os.chmod(shared, 0o777)
+        path = self._file(self.root)
+        self.backend.move(path, [self.root])
+        self.assertFalse(os.path.exists(os.path.join(shared, str(os.getuid()))))
+        top = os.path.join(self.drive, f".Trash-{os.getuid()}")
+        self.assertTrue(os.path.exists(os.path.join(top, "files", "a.png")))
+
+    def test_dot_trash_that_is_a_link_is_skipped(self):
+        elsewhere = os.path.join(self.tmp, "elsewhere")
+        os.mkdir(elsewhere)
+        os.chmod(elsewhere, 0o1777)
+        os.symlink(elsewhere, os.path.join(self.drive, ".Trash"))
+        path = self._file(self.root)
+        self.backend.move(path, [self.root])
+        self.assertEqual(os.listdir(elsewhere), [])
+
+    def test_trash_dash_uid_that_is_a_link_is_refused(self):
+        elsewhere = os.path.join(self.tmp, "elsewhere")
+        os.mkdir(elsewhere)
+        os.symlink(elsewhere, os.path.join(self.drive, f".Trash-{os.getuid()}"))
+        path = self._file(self.root)
+        with self.assertRaises(trash.TrashRefused) as caught:
+            self.backend.move(path, [self.root])
+        self.assertEqual(caught.exception.code, trash.NO_TRASH)
+        self.assertTrue(os.path.exists(path))
+        self.assertEqual(os.listdir(elsewhere), [])
+
+    def test_trash_inside_an_output_root_is_refused(self):
+        # The drive's top folder is itself an output root: its .Trash-uid
+        # would be listed as images.
+        path = self._file(self.root)
+        with self.assertRaises(trash.TrashRefused) as caught:
+            self.backend.move(path, [self.drive])
+        self.assertEqual(caught.exception.code, trash.TRASH_IN_ROOT)
+        self.assertTrue(os.path.exists(path))
+        self.assertFalse(
+            os.path.exists(os.path.join(self.drive, f".Trash-{os.getuid()}"))
+        )
+
+    def test_home_trash_inside_an_output_root_is_refused(self):
+        path = self._file(self.home_root)
+        with self.assertRaises(trash.TrashRefused) as caught:
+            self.backend.move(path, [self.home_root, self.tmp])
+        self.assertEqual(caught.exception.code, trash.TRASH_IN_ROOT)
+        self.assertTrue(os.path.exists(path))
+
+    def test_read_only_top_folder_is_no_trash(self):
+        path = self._file(self.root)
+        os.chmod(self.drive, 0o555)
+        self.addCleanup(os.chmod, self.drive, 0o755)
+        if os.access(self.drive, os.W_OK):
+            self.skipTest("running as root: permissions are not enforced")
+        self.assertEqual(self.backend.available(path, [self.root]), trash.NO_TRASH)
+        with self.assertRaises(trash.TrashRefused) as caught:
+            self.backend.move(path, [self.root])
+        self.assertEqual(caught.exception.code, trash.NO_TRASH)
+        self.assertTrue(os.path.exists(path))
+
+    def test_available_reports_without_writing(self):
+        path = self._file(self.root)
+        self.assertIsNone(self.backend.available(path, [self.root]))
+        self.assertEqual(
+            self.backend.available(path, [self.drive]), trash.TRASH_IN_ROOT
+        )
+        self.assertEqual(os.listdir(self.drive), ["output"])
+
+    # ── names and the no-replace rename ───────────────────────────────
+
+    def test_existing_name_in_the_trash_is_never_replaced(self):
+        top = os.path.join(self.drive, f".Trash-{os.getuid()}")
+        os.makedirs(os.path.join(top, "files"), mode=0o700)
+        os.makedirs(os.path.join(top, "info"), mode=0o700)
+        os.chmod(top, 0o700)
+        # A stray trashed file with no info file (another tool crashed).
+        stray = self._file(os.path.join(top, "files"), "a.png", b"older-file")
+        path = self._file(self.root)
+        self.backend.move(path, [self.root])
+        with open(stray, "rb") as handle:
+            self.assertEqual(handle.read(), b"older-file")
+        self.assertTrue(os.path.exists(os.path.join(top, "files", "a.1.png")))
+        self.assertTrue(os.path.exists(os.path.join(top, "info", "a.1.png.trashinfo")))
+        self.assertFalse(os.path.exists(os.path.join(top, "info", "a.png.trashinfo")))
+
+    def test_existing_info_name_is_skipped(self):
+        path = self._file(self.root)
+        self.backend.move(path, [self.root])
+        second = self._file(self.root)
+        self.backend.move(second, [self.root])
+        top = os.path.join(self.drive, f".Trash-{os.getuid()}")
+        self.assertEqual(
+            sorted(os.listdir(os.path.join(top, "files"))), ["a.1.png", "a.png"]
+        )
+
+    def test_missing_no_replace_rename_is_no_trash(self):
+        path = self._file(self.root)
+        for error in (38, 22):  # ENOSYS, EINVAL
+            with self.subTest(error=error):
+                backend = trash.FreedesktopBackend(
+                    data_home=self.data_home,
+                    mounts=lambda: list(self.mounts),
+                    rename=mock.Mock(side_effect=OSError(error, "no")),
+                )
+                with self.assertRaises(trash.TrashRefused) as caught:
+                    backend.move(path, [self.root])
+                self.assertEqual(caught.exception.code, trash.NO_TRASH)
+                self.assertTrue(os.path.exists(path))
+                top = os.path.join(self.drive, f".Trash-{os.getuid()}")
+                self.assertEqual(os.listdir(os.path.join(top, "info")), [])
+
+    def test_cross_device_rename_is_no_trash_and_never_copies(self):
+        path = self._file(self.root)
+        backend = trash.FreedesktopBackend(
+            data_home=self.data_home,
+            mounts=lambda: list(self.mounts),
+            rename=mock.Mock(side_effect=OSError(18, "EXDEV")),
+        )
+        with mock.patch("shutil.move") as move, mock.patch("shutil.copy2") as copy:
+            with self.assertRaises(trash.TrashRefused) as caught:
+                backend.move(path, [self.root])
+        self.assertEqual(caught.exception.code, trash.NO_TRASH)
+        move.assert_not_called()
+        copy.assert_not_called()
+        self.assertTrue(os.path.exists(path))
+
+    def test_other_rename_failure_is_failed_and_removes_only_our_info(self):
+        path = self._file(self.root)
+        backend = trash.FreedesktopBackend(
+            data_home=self.data_home,
+            mounts=lambda: list(self.mounts),
+            rename=mock.Mock(side_effect=OSError(5, "EIO")),
+        )
+        with self.assertRaises(trash.TrashRefused) as caught:
+            backend.move(path, [self.root])
+        self.assertEqual(caught.exception.code, trash.FAILED)
+        top = os.path.join(self.drive, f".Trash-{os.getuid()}")
+        self.assertEqual(os.listdir(os.path.join(top, "info")), [])
+        self.assertTrue(os.path.exists(path))
+
+    def test_never_removes_or_moves_the_source(self):
+        path = self._file(self.root)
+        real_unlink = os.unlink
+
+        def guarded_unlink(target, *args, **kwargs):
+            # Only our own .trashinfo may ever be removed.
+            self.assertTrue(str(target).endswith(".trashinfo"), target)
+            return real_unlink(target, *args, **kwargs)
+
+        backend = trash.FreedesktopBackend(
+            data_home=self.data_home,
+            mounts=lambda: list(self.mounts),
+            rename=mock.Mock(side_effect=OSError(5, "EIO")),
+        )
+        with (
+            mock.patch("os.unlink", guarded_unlink),
+            mock.patch("os.remove", guarded_unlink),
+            mock.patch("shutil.move") as move,
+            mock.patch("os.rename") as rename,
+        ):
+            with self.assertRaises(trash.TrashRefused):
+                backend.move(path, [self.root])
+        move.assert_not_called()
+        rename.assert_not_called()
+        self.assertTrue(os.path.exists(path))
+
+    def test_mountinfo_parsing_decodes_escapes(self):
+        text = (
+            "36 35 98:0 / / rw - ext4 /dev/root rw\n"
+            "37 36 98:1 / /mnt/my\\040drive rw - ext4 /dev/sdb1 rw\n"
+        )
+        self.assertEqual(trash.parse_mountinfo(text), ["/", "/mnt/my drive"])
+
+    def test_mount_of_picks_the_longest_containing_mount(self):
+        mounts = ["/", "/mnt", "/mnt/data"]
+        self.assertEqual(trash.mount_of("/mnt/data/x/y.png", mounts), "/mnt/data")
+        self.assertEqual(trash.mount_of("/mnt/datax/y.png", mounts), "/mnt")
+        self.assertEqual(trash.mount_of("/home/u/y.png", mounts), "/")
+
+
+class UnavailableBackendTests(unittest.TestCase):
+    def test_reports_no_trash_and_leaves_the_file(self):
+        backend = trash.UnavailableBackend()
+        with tempfile.TemporaryDirectory() as folder:
+            path = os.path.join(folder, "a.png")
+            with open(path, "wb") as handle:
+                handle.write(b"x")
+            self.assertEqual(backend.available(path, [folder]), trash.NO_TRASH)
+            with self.assertRaises(trash.TrashRefused) as caught:
+                backend.move(path, [folder])
+            self.assertEqual(caught.exception.code, trash.NO_TRASH)
+            self.assertTrue(os.path.exists(path))
+
+    def test_windows_and_macos_have_no_trash_yet(self):
+        for platform in ("win32", "darwin"):
+            with self.subTest(platform=platform):
+                self.assertIsInstance(
+                    trash.default_backend(platform), trash.UnavailableBackend
+                )
+        self.assertIsInstance(trash.default_backend("linux"), trash.FreedesktopBackend)
+
+    def test_module_functions_use_the_replaceable_backend(self):
+        fake = mock.Mock()
+        fake.available.return_value = "network"
+        fake.move.return_value = "label"
+        with mock.patch.object(trash, "BACKEND", fake):
+            self.assertEqual(trash.trash_available("/x/a.png", ["/x"]), "network")
+            self.assertEqual(trash.move_to_trash("/x/a.png", ["/x"]), "label")
+
+
+if __name__ == "__main__":
+    unittest.main()

tools/vendor/frontend-coverage-baseline.json

@@ -281,6 +281,12 @@
     "functions": 98,
     "statements": 98
   },
+  "workspace/images-delete.mjs": {
+    "lines": 80,
+    "branches": 70,
+    "functions": 80,
+    "statements": 80
+  },
   "workspace/images-export.mjs": {
     "lines": 98,
     "branches": 98,

utils/trash.py

@@ -0,0 +1,364 @@
+"""Move one media file into the system trash, or refuse (#480).
+
+Two rules hold for every backend:
+
+- **Rename or fail.** The only primitive is a no-replace rename within one
+  filesystem. Nothing is ever copied, unlinked or overwritten: a file that
+  cannot be moved recoverably stays where it is, and ``TrashRefused`` says why.
+- **Never into a root.** A trash folder inside an output root would be listed
+  as images, so it is refused (``trash-in-root``).
+
+Linux and BSD use :class:`FreedesktopBackend`, the freedesktop.org Trash
+specification 1.0: the home trash (``$XDG_DATA_HOME/Trash``) for a file on the
+same mount, else the file's mount's ``.Trash/$uid`` (only under a real, sticky
+``.Trash``) or ``.Trash-$uid``. Names are reserved by creating the
+``.trashinfo`` with ``O_EXCL`` and the file is moved with
+``renameat2(RENAME_NOREPLACE)``; without that call the move is refused.
+Windows and macOS have no backend yet (:class:`UnavailableBackend`): their
+own PRs add one with a native proof.
+
+``BACKEND`` is the one replaceable seam; tests install a fake.
+"""
+
+from __future__ import annotations
+
+import ctypes
+import datetime
+import errno
+import os
+import stat
+import sys
+import urllib.parse
+
+NO_TRASH = "no-trash"
+NETWORK = "network"
+TRASH_IN_ROOT = "trash-in-root"
+PATH_TOO_LONG = "path-too-long"
+LOCKED = "locked"
+FAILED = "failed"
+CODES = frozenset({NO_TRASH, NETWORK, TRASH_IN_ROOT, PATH_TOO_LONG, LOCKED, FAILED})
+
+HOME_TRASH_LABEL = "home trash"
+RENAME_NOREPLACE = 1
+MAX_NAME_ATTEMPTS = 1000
+_DIR_FLAGS = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0)
+# rename() errors that mean "this trash cannot take it", never a copy.
+_NO_TRASH_ERRNOS = {
+    errno.ENOSYS,
+    errno.EINVAL,
+    errno.EXDEV,
+    errno.EACCES,
+    errno.EPERM,
+    errno.EROFS,
+    getattr(errno, "ENOTSUP", errno.EOPNOTSUPP),
+    errno.EOPNOTSUPP,
+}
+
+
+class TrashRefused(Exception):
+    """The file was not moved; ``code`` is one of ``CODES``."""
+
+    def __init__(self, code: str, detail: str = ""):
+        super().__init__(detail or code)
+        self.code = code
+
+
+def parse_mountinfo(text: str) -> list[str]:
+    """Mount points of ``/proc/self/mountinfo`` text (octal escapes decoded)."""
+    points = []
+    for line in text.splitlines():
+        fields = line.split(" ")
+        if len(fields) > 4:
+            raw = fields[4]
+            points.append(
+                bytes(raw, "utf-8")
+                .decode("unicode_escape")
+                .encode("latin-1")
+                .decode("utf-8", "surrogateescape")
+            )
+    return points
+
+
+def _proc_mounts() -> list[str]:
+    try:
+        with open(
+            "/proc/self/mountinfo", encoding="utf-8", errors="surrogateescape"
+        ) as f:
+            return parse_mountinfo(f.read())
+    except OSError:
+        return []
+
+
+def _inside(path: str, folder: str) -> bool:
+    folder = folder.rstrip("/") or "/"
+    return path == folder or path.startswith(folder if folder == "/" else folder + "/")
+
+
+def mount_of(path: str, mounts: list[str]) -> str:
+    """The longest mount point holding *path* (both real paths)."""
+    best = "/"
+    for point in mounts:
+        if _inside(path, point) and len(point) > len(best):
+            best = point
+    return best
+
+
+def _ismount_walk(path: str) -> str:
+    """Fallback without mountinfo (BSD): walk up until ``os.path.ismount``."""
+    current = path
+    while not os.path.ismount(current):
+        parent = os.path.dirname(current)
+        if parent == current:
+            break
+        current = parent
+    return current
+
+
+def _libc_renameat2():
+    if os.name != "posix":
+        return None  # Windows: no libc, and no freedesktop backend
+    try:
+        libc = ctypes.CDLL(None, use_errno=True)
+        function = libc.renameat2
+    except (OSError, AttributeError, TypeError):
+        return None
+    function.argtypes = [
+        ctypes.c_int,
+        ctypes.c_char_p,
+        ctypes.c_int,
+        ctypes.c_char_p,
+        ctypes.c_uint,
+    ]
+    function.restype = ctypes.c_int
+    return function
+
+
+_RENAMEAT2 = _libc_renameat2()
+
+
+def rename_noreplace(src_dir_fd: int, src: str, dst_dir_fd: int, dst: str) -> None:
+    """``renameat2(..., RENAME_NOREPLACE)``; raises OSError (ENOSYS when missing)."""
+    if _RENAMEAT2 is None:
+        raise OSError(errno.ENOSYS, "renameat2 is not available")
+    result = _RENAMEAT2(
+        src_dir_fd, os.fsencode(src), dst_dir_fd, os.fsencode(dst), RENAME_NOREPLACE
+    )
+    if result != 0:
+        code = ctypes.get_errno()
+        raise OSError(code, os.strerror(code))
+
+
+def _canonical(path: str) -> str:
+    return os.path.normcase(os.path.realpath(path))
+
+
+def _check_roots(trash_dir: str, roots) -> None:
+    trash_key = _canonical(trash_dir)
+    for root in roots:
+        if _inside(trash_key, _canonical(str(root))):
+            raise TrashRefused(TRASH_IN_ROOT)
+
+
+def _owned_folder(path: str, uid: int, *, create: bool) -> bool:
+    """Whether *path* is (or, with *create*, now is) a real folder we own."""
+    try:
+        info = os.lstat(path)
+    except FileNotFoundError:
+        if not create:
+            return False
+        try:
+            os.mkdir(path, 0o700)
+        except OSError:
+            return False
+        info = os.lstat(path)
+    except OSError:
+        return False
+    return stat.S_ISDIR(info.st_mode) and info.st_uid == uid
+
+
+class UnavailableBackend:
+    """No trash on this system: every move is refused, nothing is touched."""
+
+    def available(self, path: str, roots) -> str | None:
+        return NO_TRASH
+
+    def move(self, path: str, roots) -> str:
+        raise TrashRefused(NO_TRASH)
+
+
+class FreedesktopBackend:
+    """The freedesktop.org Trash specification 1.0 (Linux, BSD)."""
+
+    def __init__(self, data_home=None, mounts=None, rename=None, uid=None, now=None):
+        self._data_home = data_home
+        self._mounts = mounts or _proc_mounts
+        self._rename = rename or rename_noreplace
+        if uid is None:
+            uid = os.getuid() if hasattr(os, "getuid") else -1
+        self._uid = uid
+        self._now = now or datetime.datetime.now
+
+    def _home_data(self) -> str:
+        if self._data_home:
+            return self._data_home
+        configured = os.environ.get("XDG_DATA_HOME")
+        if configured and os.path.isabs(configured):
+            return configured
+        return os.path.join(os.path.expanduser("~"), ".local", "share")
+
+    def _mount(self, real: str, mounts) -> str:
+        return mount_of(real, mounts) if mounts else _ismount_walk(real)
+
+    def _plan(self, path: str, roots, *, create: bool):
+        """``(trash_dir, topdir_or_None, label)`` for *path*; raises TrashRefused."""
+        real = os.path.realpath(path)
+        mounts = [os.path.realpath(point) for point in self._mounts()]
+        home_data = os.path.realpath(self._home_data())
+        if self._mount(real, mounts) == self._mount(home_data, mounts):
+            trash_dir = os.path.join(home_data, "Trash")
+            _check_roots(trash_dir, roots)
+            if create:
+                os.makedirs(trash_dir, mode=0o700, exist_ok=True)
+            return trash_dir, None, HOME_TRASH_LABEL
+        top = self._mount(real, mounts)
+        shared = os.path.join(top, ".Trash")
+        try:
+            info = os.lstat(shared)
+            sticky = stat.S_ISDIR(info.st_mode) and bool(info.st_mode & stat.S_ISVTX)
+        except OSError:
+            sticky = False
+        candidates = []
+        if sticky:
+            candidates.append(os.path.join(shared, str(self._uid)))
+        candidates.append(os.path.join(top, f".Trash-{self._uid}"))
+        for trash_dir in candidates:
+            _check_roots(trash_dir, roots)
+            if self._usable(trash_dir, create=create):
+                label = f"{os.path.relpath(trash_dir, top)} on its drive"
+                return trash_dir, top, label
+        raise TrashRefused(NO_TRASH)
+
+    def _usable(self, trash_dir: str, *, create: bool) -> bool:
+        if os.path.lexists(trash_dir):
+            return _owned_folder(trash_dir, self._uid, create=False)
+        parent = os.path.dirname(trash_dir)
+        if not create:
+            return os.access(parent, os.W_OK | os.X_OK)
+        return _owned_folder(trash_dir, self._uid, create=True)
+
+    def available(self, path: str, roots) -> str | None:
+        """None when *path* can be trashed, else the refusal code (no writes)."""
+        try:
+            self._plan(path, roots, create=False)
+        except TrashRefused as refused:
+            return refused.code
+        except OSError:
+            return NO_TRASH
+        return None
+
+    def move(self, path: str, roots) -> str:
+        """Move *path* into its trash and return the trash's label."""
+        try:
+            trash_dir, top, label = self._plan(path, roots, create=True)
+        except OSError as error:
+            raise TrashRefused(NO_TRASH, str(error)) from error
+        real = os.path.realpath(path)
+        recorded = os.path.relpath(real, top) if top else real
+        opened = []
+        try:
+            trash_fd = self._open_dir(trash_dir, opened)
+            files_fd = self._subfolder(trash_fd, "files", opened)
+            info_fd = self._subfolder(trash_fd, "info", opened)
+            source_fd = self._open_dir(os.path.dirname(real), opened)
+            self._move_named(
+                source_fd, os.path.basename(real), files_fd, info_fd, recorded
+            )
+        finally:
+            for fd in opened:
+                os.close(fd)
+        return label
+
+    def _open_dir(self, path: str, opened: list) -> int:
+        try:
+            fd = os.open(path, _DIR_FLAGS)
+        except OSError as error:
+            raise TrashRefused(NO_TRASH, str(error)) from error
+        opened.append(fd)
+        return fd
+
+    def _subfolder(self, parent_fd: int, name: str, opened: list) -> int:
+        try:
+            os.mkdir(name, 0o700, dir_fd=parent_fd)
+        except FileExistsError:
+            pass
+        except OSError as error:
+            raise TrashRefused(NO_TRASH, str(error)) from error
+        try:
+            info = os.lstat(name, dir_fd=parent_fd)
+        except OSError as error:
+            raise TrashRefused(NO_TRASH, str(error)) from error
+        if not stat.S_ISDIR(info.st_mode) or info.st_uid != self._uid:
+            raise TrashRefused(NO_TRASH, f"{name} is not our folder")
+        try:
+            fd = os.open(name, _DIR_FLAGS, dir_fd=parent_fd)
+        except OSError as error:
+            raise TrashRefused(NO_TRASH, str(error)) from error
+        opened.append(fd)
+        return fd
+
+    def _info_text(self, recorded: str) -> bytes:
+        encoded = urllib.parse.quote(os.fsencode(recorded), safe="/")
+        stamp = self._now().strftime("%Y-%m-%dT%H:%M:%S")
+        return f"[Trash Info]\nPath={encoded}\nDeletionDate={stamp}\n".encode("utf-8")
+
+    def _move_named(self, source_fd, name, files_fd, info_fd, recorded) -> None:
+        stem, extension = os.path.splitext(name)
+        for attempt in range(MAX_NAME_ATTEMPTS):
+            candidate = name if attempt == 0 else f"{stem}.{attempt}{extension}"
+            info_name = candidate + ".trashinfo"
+            try:
+                fd = os.open(
+                    info_name,
+                    os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0),
+                    0o600,
+                    dir_fd=info_fd,
+                )
+            except FileExistsError:
+                continue
+            except OSError as error:
+                raise TrashRefused(NO_TRASH, str(error)) from error
+            try:
+                os.write(fd, self._info_text(recorded))
+            finally:
+                os.close(fd)
+            try:
+                self._rename(source_fd, name, files_fd, candidate)
+                return
+            except OSError as error:
+                os.unlink(info_name, dir_fd=info_fd)
+                if error.errno == errno.EEXIST:
+                    continue  # a stray trashed file has that name: never replace it
+                if error.errno in _NO_TRASH_ERRNOS:
+                    raise TrashRefused(NO_TRASH, str(error)) from error
+                raise TrashRefused(FAILED, str(error)) from error
+        raise TrashRefused(FAILED, "no free name in the trash")
+
+
+def default_backend(platform: str = sys.platform):
+    """The backend for *platform*; Windows and macOS have none yet (#480)."""
+    if platform in ("win32", "darwin") or platform.startswith("cygwin"):
+        return UnavailableBackend()
+    return FreedesktopBackend()
+
+
+BACKEND = default_backend()
+
+
+def trash_available(path: str, roots) -> str | None:
+    """None when *path* can be trashed, else the refusal code (blocking, no writes)."""
+    return BACKEND.available(path, roots)
+
+
+def move_to_trash(path: str, roots) -> str:
+    """Move one regular file into the trash; returns where. Raises TrashRefused."""
+    return BACKEND.move(path, roots)

web/js/workspace/file-inspector.mjs

@@ -234,7 +234,7 @@ export const DEFERRED_TEXT = 'Metadata loads when you ask for it.';
  * collections checklist; `onCollectionsChanged({added, removed})` hears the confirmed changes.
  */
 export function mountFileInspector(container, {api, sender, parentScope, variant, inline = false, promptContext = '', onTagFilter, onRetry, onRefresh, onLoad, onInteraction = () => {}, copyFallback = openCopyFallback,
-    collections = null, onCollectionsChanged = null}) {
+    collections = null, onCollectionsChanged = null, onDelete = null}) {
     if (parentScope.signal.aborted) return {setState() {}, dispose() {}};
     const scope = childScope(parentScope);
     let contentScope;
@@ -339,6 +339,9 @@ export function mountFileInspector(container, {api, sender, parentScope, variant
                 onChanged:onCollectionsChanged, focusFallback:fallback}));
             footer.append(button);
         }
+        // #480: Delete… for a listed Images file only, last in the actions.
+        const removable = onDelete ? revealTarget(record.identity) : null;
+        if (removable) footer.append(deleteButton(removable));
         if (!full && footer.childNodes.length) container.append(footer);
     }
     /**
@@ -456,6 +459,12 @@ export function mountFileInspector(container, {api, sender, parentScope, variant
         });
         return button;
     }
+    /** #480: "Delete…" opens the Images delete review for this one file. */
+    function deleteButton(target) {
+        const button = createButton('Delete…', 'danger');
+        contentScope.listen(button, 'click', () => onDelete(target, button));
+        return button;
+    }
     /** "Workflow & raw metadata": each supported record (`pre[tabindex=0] > code`), then the two downloads. */
     function rawMetadata(record) {
         const {metadata} = record, section = disclosure('Workflow & raw metadata');

web/js/workspace/images-delete.mjs

@@ -0,0 +1,290 @@
+/**
+ * Images › Delete… (#480, D159): the review dialog for the checked images (or "all N matching", or the one image of
+ * the inspector or viewer) and the apply.
+ *
+ * - The concept's selection bar has "Remove links…" (`danger`) opening its `maintenancePlan` review (`dl.meta-grid`
+ *   Scope / Protected / Backup / Media on disk, Cancel and a primary). Production names it "Delete…" and asks each
+ *   time (maintainer ruling): a radio group "What should happen to the files?" with nothing checked, "Remove from the
+ *   library only" (records go, the file stays on disk, hidden from Images) or "Also move the files to the trash".
+ * - One preview (`POST /images/delete/preview`) answers both modes; choosing a mode only redraws the rows. The primary
+ *   (`danger`) is enabled once a mode with images to act on is chosen. Apply sends `{token, mode}` to the reviewed-plan
+ *   apply; the job is followed in the Activity store and `onDeleted(job)` runs when it ends.
+ * - Nothing here deletes a file: the server never unlinks one (it renames into the system trash or leaves it).
+ */
+import { createButton } from '../shared/toolkit/core.js';
+import { closeWorkspaceDialog, openCloseOnlyReview, openContentDialog } from './workspace-dialogs.mjs';
+import { followJobEnd } from './images-tagging-bulk.mjs';
+import { applyRefusal, APPLY_URL } from './maintenance-review.mjs';
+import { showWorkspaceToast } from './toast.mjs';
+import { countOf, noun } from './plural.mjs';
+
+export const PREVIEW_URL = '/prompt_manager/images/delete/preview';
+export const MAX_DELETE = 500;
+export const TOO_MANY = 'Select at most 500 images.';
+export const QUESTION = 'What should happen to the files?';
+export const MODES = Object.freeze([
+    Object.freeze({value: 'library', label: 'Remove from the library only', hint: 'Files stay on disk, hidden from Images.'}),
+    Object.freeze({value: 'trash', label: 'Also move the files to the trash',
+        hint: 'You can restore them from the trash until it is emptied.'}),
+]);
+export const NONE_DELETABLE = 'None of the selected images can be deleted.';
+export const REVIEW_FAILED = 'The review failed. Nothing was changed.';
+export const SOURCE_CHANGED = 'The sources changed. Refresh the images and try again.';
+export const STORAGE_CHANGED = 'Prompt storage was restored. Reopen Delete.';
+export const USE_LIBRARY_ONLY = 'To remove these files from the library anyway, choose “Remove from the library only”.';
+export const NO_TRASH_HERE = 'Moving to the trash is not available on this system yet.';
+const SENTENCES = Object.freeze({
+    libraryBackup: 'Removed images stay hidden unless the file is modified or replaced. Their tags, category, collections '
+        + 'and favorite can only be recovered by restoring the safety backup, which also undoes every later change.',
+    libraryNoBackup: 'Removed images stay hidden unless the file is modified or replaced. Their tags, category, '
+        + 'collections and favorite cannot be recovered.',
+    trash: 'A file restored from the trash shows again without its tags, category, collections or favorite; a scan '
+        + 'links it to its prompt again.',
+});
+export { SENTENCES };
+
+const RECORD_WORDS = [
+    ['image_tags', 'with image tags', 'with image tags'],
+    ['auto_tagged', 'auto-tagged', 'auto-tagged'],
+    ['categories', 'category', 'categories'],
+    ['collections', 'collection membership', 'collection memberships'],
+    ['favorites', 'favorite', 'favorites'],
+    ['review_items', 'Needs review item', 'Needs review items'],
+];
+
+function el(tag, text, className) {
+    const node = document.createElement(tag);
+    if (text !== undefined) node.textContent = text;
+    if (className) node.className = className;
+    return node;
+}
+
+function number(value) {
+    return Number(value ?? 0).toLocaleString('en-US');
+}
+
+/** The images a mode acts on. */
+export function modeCount(plan, mode) {
+    return Number(plan?.[mode]?.count ?? 0);
+}
+
+/** "Name, name, name and K more" for up to three refused rows, grouped by their sentence. */
+export function refusedValue(summary) {
+    const count = Number(summary?.count ?? 0);
+    if (!count) return null;
+    const rows = Array.isArray(summary.rows) ? summary.rows : [];
+    const names = rows.slice(0, 3).map(row => String(row.path).split('/').at(-1));
+    const more = count - names.length;
+    const reasons = [...new Set(rows.map(row => row.detail).filter(Boolean))];
+    return `${number(count)}: ${names.join(', ')}${more > 0 ? ` and ${number(more)} more` : ''}`
+        + (reasons.length ? `. ${reasons.join(' ')}` : '');
+}
+
+/** The `dl.meta-grid` rows for `mode` (design §9.3). */
+export function deleteRows(plan, mode) {
+    const rows = [['Scope', countOf(modeCount(plan, mode), 'image')]];
+    const refused = {count: 0, rows: []};
+    for (const part of mode === 'trash' ? [plan.refused, plan.trash_refused] : [plan.refused]) {
+        refused.count += Number(part?.count ?? 0);
+        refused.rows.push(...(part?.rows ?? []));
+    }
+    const notChanged = refusedValue(refused);
+    if (notChanged) rows.push(['Not changed', notChanged]);
+    const records = plan.records ?? {};
+    const links = Number(records.prompt_links ?? 0);
+    if (links) {
+        const orphans = plan.orphaned_prompts ?? {count: 0, titles: []};
+        let text = `${countOf(links, 'link')}.`;
+        if (orphans.count > 0) {
+            const titles = (orphans.titles ?? []).filter(Boolean);
+            const more = orphans.count - titles.length;
+            text += ` ${countOf(orphans.count, 'prompt')} ${orphans.count === 1 ? 'loses its' : 'lose their'} last image`
+                + (titles.length ? `: ${titles.join(', ')}${more > 0 ? ` and ${number(more)} more` : ''}` : '')
+                + `. The ${noun(orphans.count, 'prompt')} ${orphans.count === 1 ? 'stays' : 'stay'} on the Library page.`;
+        }
+        rows.push(['Prompt links', text]);
+    }
+    const details = RECORD_WORDS.filter(([key]) => Number(records[key] ?? 0) > 0)
+        .map(([key, one, many]) => `${number(records[key])} ${Number(records[key]) === 1 ? one : many}`);
+    if (details.length) rows.push(['Image details', details.join(', ')]);
+    rows.push(['Backup', backupOn(plan) ? 'Create and verify before changes (the database, not the files)'
+        : 'Off: no safety backup (System › Backup & export › Recovery policy)']);
+    rows.push(['Media on disk', mode === 'trash'
+        ? 'Files move to the trash. Nothing is deleted permanently. Disk space is freed when you empty the trash.'
+        : 'Files stay on disk, hidden from Images.']);
+    return rows;
+}
+
+function backupOn(plan) {
+    return plan?.backup?.policy !== 'off';
+}
+
+/** The `p.small` sentences for `mode`. */
+export function deleteSentences(plan, mode) {
+    if (mode === 'library') return [backupOn(plan) ? SENTENCES.libraryBackup : SENTENCES.libraryNoBackup];
+    const sentences = [SENTENCES.trash];
+    const trashRefused = Number(plan?.trash_refused?.count ?? 0);
+    if (trashRefused && trashRefused >= modeCount(plan, 'library')) sentences.push(NO_TRASH_HERE);
+    if (trashRefused) sentences.push(USE_LIBRARY_ONLY);
+    return sentences;
+}
+
+/** The primary's label: "Remove N images from the library" or "Move N images to the trash". */
+export function primaryLabel(mode, count) {
+    return mode === 'trash' ? `Move ${countOf(count, 'image')} to the trash` : `Remove ${countOf(count, 'image')} from the library`;
+}
+
+/** Why the review cannot start at all, or null. */
+export function closeOnlyReason(plan) {
+    if (modeCount(plan, 'library') === 0 && modeCount(plan, 'trash') === 0) return NONE_DELETABLE;
+    if (plan?.disk && plan.disk.sufficient === false) {
+        return `There is not enough free space for the safety backup (needs ${number(plan.disk.required_bytes)} bytes, `
+            + `has ${number(plan.disk.free_bytes)}).`;
+    }
+    return null;
+}
+
+/** The toast for a review that could not be read. */
+export function reviewFailure(error) {
+    if (error?.code === 'source-changed') return SOURCE_CHANGED;
+    if (error?.code === 'storage-changed') return STORAGE_CHANGED;
+    if (error?.code === 'too-many-images') return TOO_MANY;
+    if (error?.code === 'busy' && error.message) return error.message;
+    return REVIEW_FAILED;
+}
+
+/** The toast when the job ends (design §9.4). */
+export function outcomeToast(job) {
+    const counters = job?.counters ?? {};
+    const mode = job?.scope?.mode;
+    if (job?.scope?.failure === 'backup-failed') return 'The safety backup failed; nothing was changed.';
+    if (job?.scope?.failure === 'sources-changed') return 'The sources changed; nothing was changed.';
+    const done = Number(mode === 'trash' ? counters.trashed : counters.removed) || 0;
+    const total = Number(job?.total ?? 0);
+    if (job?.scope?.failure === 'records-failed') {
+        return `Stopped after ${number(done)} of ${countOf(total, 'image')}; the rest were not changed. See Activity.`;
+    }
+    const head = mode === 'trash' ? `Moved ${countOf(done, 'image')} to the trash.` : `Removed ${countOf(done, 'image')} from the library.`;
+    const unchanged = Number(counters.not_changed ?? 0);
+    return unchanged ? `${head} ${number(unchanged)} ${unchanged === 1 ? 'was' : 'were'} not changed (see Activity).` : head;
+}
+
+function modeGroup(onChange) {
+    const group = el('div', undefined, 'field');
+    group.setAttribute('role', 'radiogroup');
+    const caption = el('span', QUESTION);
+    caption.id = 'image-delete-mode-label';
+    group.setAttribute('aria-labelledby', caption.id);
+    group.append(caption);
+    for (const choice of MODES) {
+        const label = el('label');
+        const input = el('input');
+        Object.assign(input, {type: 'radio', name: 'image-delete-mode', value: choice.value, checked: false});
+        input.addEventListener('change', () => { if (input.checked) onChange(choice.value); });
+        label.append(input, ` ${choice.label}`, el('span', ` ${choice.hint}`, 'small muted'));
+        group.append(label);
+    }
+    return group;
+}
+
+/** The review for an answered preview `plan`; `onApply(mode)` runs on the primary. */
+export function showDeleteReview(plan, {scope, count, trigger = null, onApply}) {
+    const title = `Delete ${countOf(count, 'image')}`;
+    const reason = closeOnlyReason(plan);
+    if (reason) {
+        openCloseOnlyReview({scope, title, grid: deleteRows(plan, 'library'), sentence: reason, trigger});
+        return null;
+    }
+    let mode = null, started = false;
+    const list = el('dl', undefined, 'meta-grid');
+    list.hidden = true;
+    const notes = el('div', undefined, 'image-delete-notes');
+    const cancel = createButton('Cancel');
+    const start = createButton('Choose what happens to the files', 'danger image-delete-start');
+    start.setAttribute('aria-disabled', 'true');
+    function render() {
+        list.replaceChildren();
+        notes.replaceChildren();
+        if (!mode) return;
+        for (const [term, value] of deleteRows(plan, mode)) {
+            const entry = el('div');
+            entry.append(el('dt', term), el('dd', value));
+            list.append(entry);
+        }
+        list.hidden = false;
+        for (const sentence of deleteSentences(plan, mode)) notes.append(el('p', sentence, 'small'));
+        const acting = modeCount(plan, mode);
+        start.textContent = primaryLabel(mode, acting);
+        start.setAttribute('aria-disabled', String(acting === 0));
+    }
+    const group = modeGroup(value => { mode = value; render(); });
+    cancel.addEventListener('click', () => closeWorkspaceDialog());
+    // Nothing starts without an explicit mode: an aria-disabled primary ignores every activation (click, Enter, Space),
+    // and a chosen one starts once however often it is pressed. The dialog has no form, so Enter elsewhere submits nothing.
+    start.addEventListener('click', () => {
+        if (started || !MODES.some(choice => choice.value === mode) || start.getAttribute('aria-disabled') === 'true') return;
+        started = true;
+        start.setAttribute('aria-disabled', 'true');
+        const chosen = mode;
+        closeWorkspaceDialog();
+        onApply(chosen);
+    });
+    const actions = el('div', undefined, 'dialog-actions');
+    actions.append(cancel, start);
+    openContentDialog({scope, title, nodes: [group, list, notes, actions], trigger});
+    return {group, list, start};
+}
+
+/** Apply `plan` in `mode`; the started job, or null (a toast said why). */
+export async function applyDelete({api, store, plan, mode, notify = showWorkspaceToast}) {
+    if (!MODES.some(choice => choice.value === mode)) throw new TypeError('A delete needs an explicit mode');
+    let result;
+    try {
+        result = await api.post(APPLY_URL, {token: plan.token, mode});
+    } catch (error) {
+        const refusal = applyRefusal(error);
+        notify(refusal.toast);
+        if (refusal.lost) store?.load?.();
+        return null;
+    }
+    try {
+        const job = store ? store.track(result?.job) : result?.job;
+        notify(`${job.title} started.`);
+        return job;
+    } catch {
+        store?.load?.();
+        return null;
+    }
+}
+
+/**
+ * The whole flow for `images` (the checked ones) or `filter` ("all N matching", `count` of them): the preview, the
+ * review and the apply. `trigger` reads "Reviewing…" and is aria-disabled until the review opens; a second press is
+ * ignored. `onDeleted(job)` runs when the started job ends.
+ */
+export async function openDeleteImages({api, scope, store, images = [], filter = null, count = images.length, trigger = null,
+    onDeleted = () => {}, notify = showWorkspaceToast}) {
+    if (count > MAX_DELETE) { notify(TOO_MANY); return; }
+    if (!filter && !images.length) return;
+    if (trigger?.getAttribute('aria-disabled') === 'true') return;
+    const label = trigger?.textContent;
+    trigger?.setAttribute('aria-disabled', 'true');
+    if (trigger) trigger.textContent = 'Reviewing…';
+    let plan;
+    try {
+        const body = filter ? {filter} : {items: images.map(({root_index, root_scope, path}) => ({root_index, root_scope, path}))};
+        plan = await api.post(PREVIEW_URL, body, {signal: scope.signal});
+    } catch (error) {
+        if (!scope.signal.aborted) notify(reviewFailure(error));
+        return;
+    } finally {
+        if (trigger) { trigger.textContent = label; trigger.removeAttribute('aria-disabled'); }
+    }
+    if (scope.signal.aborted) return;
+    showDeleteReview(plan, {scope, count, trigger, onApply: async mode => {
+        const job = await applyDelete({api, store, plan, mode, notify});
+        if (!job) return;
+        if (store) followJobEnd({store, scope, id: job.id, onEnd: ended => { notify(outcomeToast(ended)); onDeleted(ended); }});
+        else onDeleted(job);
+    }});
+}

web/js/workspace/images-selection.mjs

@@ -9,6 +9,7 @@ import { TOO_MANY, bulkTargets, openAddTags, openAutoTag } from './images-taggin
 import { showWorkspaceToast } from './toast.mjs';
 import { applyKeys, createLongPress, matchingLabel, matchingOffer, offersAllMatching, rangeKeys } from './images-multiselect.mjs';
 import { openAddToCollection, removeFromCollection } from './images-collections.mjs';
+import { openDeleteImages } from './images-delete.mjs';
 
 const CHECKBOX = '.image-overlays input[type="checkbox"]';
 const CHECKBOX_LABEL = '.image-overlays > label';
@@ -50,7 +51,7 @@ function el(tag, text, className) {
  */
 export function mountSelectionBar({ scope, api, grid, getItems, focusFallback, onCategorySet, activity = null, onTagged,
     getListing, getFilter, getCategories, checked: initial = Object.freeze(new Set()), collections = null,
-    collection = null, onCollectionsChanged = null, onSourceChanged = null }) {
+    collection = null, onCollectionsChanged = null, onSourceChanged = null, onDeleted = null }) {
     let checked = initial;
     let exporting = false, categorizing = false;
     // D135: the range anchor, "all N matching" mode, touch selection mode and a pending Shift+click on a checkbox.
@@ -93,7 +94,15 @@ export function mountSelectionBar({ scope, api, grid, getItems, focusFallback, o
         scope.listen(removeButton, 'click', () => removeChecked(removeButton));
         bulk.append(removeButton);
     }
-    bulk.append(exportMetadata, clearButton);
+    bulk.append(exportMetadata);
+    // #480 (D159): the concept's "Remove links…" (`danger`) place, after Export metadata and before Clear.
+    if (onDeleted) {
+        const deleteButton = createButton('Delete…', 'danger');
+        scope.listen(deleteButton, 'click', () => openDeleteImages({ api, scope, store: activity, images: checkedImages(),
+            ...matchingTarget(), trigger: deleteButton, onDeleted }));
+        bulk.append(deleteButton);
+    }
+    bulk.append(clearButton);
 
     /** The count updates in place: the same bar and `strong`, its text written only when it changes. */
     function render() {

web/js/workspace/images.mjs

@@ -9,6 +9,7 @@ import { createPromptSender } from './prompt-sender.mjs';
 import { promptRoles } from './prompt-blocks.mjs';
 import { createWorkspaceIcon } from './icons.mjs';
 import { createMediaViewer } from './media-viewer.mjs';
+import { openDeleteImages } from './images-delete.mjs';
 import { noun } from './plural.mjs';
 import { pageHeading } from './heading.mjs';
 import { mountToolbar } from './toolbar.mjs';
@@ -299,6 +300,7 @@ export function mountImages(main, api, parentScope, {embedded = false, fetchPage
     // selection is kept across appended batches and cleared with the collection; the Library-embedded gallery has none.
     const bar = embedded ? null : mountSelectionBar({scope, api, grid, getItems:() => state?.items ?? [], focusFallback,
         checked:visit?.checked, onCategorySet:() => { loadCategories(); reset(); }, activity, onTagged:keys => refreshTagged(keys),
+        onDeleted:deleted,
         // The Set category field suggests the toolbar's categories (#332 MC13).
         getCategories:() => categoryNames,
         // #392 "Select all N matching": the listing's total and its filter, as the server lists them.
@@ -321,6 +323,7 @@ export function mountImages(main, api, parentScope, {embedded = false, fetchPage
     };
     const mediaViewer = createMediaViewer({api,container:main,parentScope:scope,inspectionController:inspection,sender,
         collections:inspectorCollections,onCollectionsChanged:collectionsChangedBy,
+        onDelete:embedded ? null : (item,button) => deleteOne(item,button,() => mediaViewer.close()),
         onSelect:image => {if (selection !== imageKey(image)) inspect(image);else inspection.select(image);},
         onRefresh:refreshInspectionCollection,onClose:scheduleLoad,onMediaError:image => recovery.recover(image),onRestoreFocus:image => {
             const index = state?.items.findIndex(entry => imageKey(entry) === imageKey(image));
@@ -338,7 +341,8 @@ export function mountImages(main, api, parentScope, {embedded = false, fetchPage
         for (const video of details.querySelectorAll('video')) releaseVideo(video);
         detailInspector = mountFileInspector(details, {sender, parentScope:scope, variant:'panel',
             onTagFilter:embedded ? undefined : filterByTag, onRetry:() => inspection.retry(), onLoad:() => inspection.load(), onRefresh:refreshInspectionCollection,
-            collections:inspectorCollections, onCollectionsChanged:collectionsChangedBy});
+            collections:inspectorCollections, onCollectionsChanged:collectionsChangedBy,
+            onDelete:embedded ? null : (target, button) => deleteOne(target, button)});
         detailInspector.setState(inspection.state);
         inspection.select(image);
         for (const [key, card] of rendered) markSelected(card, key === selection);
@@ -647,6 +651,15 @@ export function mountImages(main, api, parentScope, {embedded = false, fetchPage
         clearChecks:() => bar?.clear(), listsNow:() => !recovering,
         current:started => !scope.signal.aborted && started === generation,
         rediscover:() => loadOptions().then(found => { if (found) renderLoadStatus(); })})});
+    /** #480: a finished delete job reloads the listing (the deleted files leave it, and the selection clears). */
+    function deleted() {
+        loadCategories();
+        reset();
+    }
+    /** #480: Delete… for one file, from the inspector or the viewer (`after` closes the viewer first). */
+    function deleteOne(target, trigger, after = null) {
+        openDeleteImages({api, scope, store:activity, images:[target], trigger, onDeleted:() => { after?.(); deleted(); }});
+    }
     function reset() {
         generation += 1;
         mediaViewer.stopSlideshow();

web/js/workspace/inspect.mjs

@@ -115,9 +115,39 @@ export function mountInspect(main, { parentScope, api, sender }) {
         retiring.delete(url);
         entry.waiting.abort(); clearTimeout(entry.timer);
         releaseMedia(entry.preview);
-        if (entry.revoke) URL.revokeObjectURL(url);
+        if (entry.revoke) revokeObjectURL(url);
         entry.done();
     }
+    // #483 (D62): the viewer's own video, when a new selection closes the viewer while that video is still starting.
+    // The viewer stays open until the video reports metadata or an error (or D62's 15 s), then closes; its URL, which
+    // the session asked to revoke, goes after that. A viewer asked for meanwhile opens then: only the latest request,
+    // and only if its selection is still current (an automatic open waits for its own inline video again).
+    let held = null;
+    function closeViewer() {
+        const node = document.querySelector('#viewer-dialog video');
+        if (!node || !isStarting(node)) { viewer.close(); return; }
+        const waiting = new AbortController();
+        held = { url: node.getAttribute('src'), revoke: false, waiting, pending: null, timer: setTimeout(finishHeld, STARTING_VIDEO_MS) };
+        for (const type of ['loadedmetadata', 'error']) node.addEventListener(type, finishHeld, { once: true, signal: waiting.signal });
+    }
+    function endHeld() {
+        const entry = held;
+        if (!entry) return null;
+        held = null;
+        entry.waiting.abort(); clearTimeout(entry.timer);
+        return entry;
+    }
+    function finishHeld() {
+        const entry = endHeld();
+        if (!entry) return;
+        viewer.close();
+        // Through the session's hook, so an inline video still starting on the same URL keeps it until it settles.
+        if (entry.revoke) revokeObjectURL(entry.url);
+        const request = entry.pending;
+        if (!request || scope.signal.aborted || session.entry !== request.entry || session.state.phase !== 'ready') return;
+        if (request.auto) autoOpen(request.entry);
+        else openViewer(request.trigger);
+    }
     function retire(preview, url) {
         const waiting = new AbortController();
         const entry = { preview, waiting, revoke: false, timer: setTimeout(() => finishRetiring(url), STARTING_VIDEO_MS) };
@@ -129,6 +159,7 @@ export function mountInspect(main, { parentScope, api, sender }) {
     }
     const finishAllRetiring = () => { for (const url of [...retiring.keys()]) finishRetiring(url); };
     const revokeObjectURL = url => {
+        if (held?.url === url) { held.revoke = true; return; }
         const entry = retiring.get(url);
         if (entry) entry.revoke = true;
         else URL.revokeObjectURL(url);
@@ -136,7 +167,8 @@ export function mountInspect(main, { parentScope, api, sender }) {
     const session = createLocalSession({ parentScope: sessionScope, pageTarget: null, revokeObjectURL });
     const viewer = createMediaViewer({ api, container: main, parentScope: scope, inspectionController: session.controller, sender,
         adapter: localFileAdapter, onRestoreFocus: () => { if (open.isConnected) open.focus({ preventScroll: true }); },
-        onClose: () => { viewerKey = null; } });
+        // A held viewer closed by hand (Escape, Close) ends its hold at once: its video is gone.
+        onClose: () => { viewerKey = null; if (held) finishHeld(); } });
 
     let viewerKey = null, media = null, mediaUrl = null, statusLine = null, rejectionLine = null, rejectionTimer = 0, announced = null, chosen = null;
     function statusElement(message) {
@@ -209,7 +241,7 @@ export function mountInspect(main, { parentScope, api, sender }) {
     scope.cleanup(session.subscribe(state => {
         // D155: a new selection can start while the viewer shows the previous one; it closes first, so no element holds
         // the old URL when the session revokes it (the new file's viewer opens once it is ready).
-        if (viewerKey !== null && (state.phase !== 'ready' || session.entry?.key !== viewerKey)) { viewer.close(); viewerKey = null; }
+        if (viewerKey !== null && (state.phase !== 'ready' || session.entry?.key !== viewerKey)) { closeViewer(); viewerKey = null; }
         showMedia(state);
         showStatus(state.message);
         const entry = session.entry;
@@ -220,17 +252,14 @@ export function mountInspect(main, { parentScope, api, sender }) {
             // while this selection is still the current one and no other dialog is open (the viewer is replaced). D62:
             // a video opens it once the inline preview has settled, so a replacement while it starts never tears down a
             // second starting video in the viewer (that deadlocked WebKit).
-            if (AUTO_OPEN.has(state.message)) whenSettled(media, () => {
-                if (scope.signal.aborted || session.entry !== entry || session.state.phase !== 'ready' || document.querySelector('dialog[open]:not(#viewer-dialog)')) return;
-                openViewer(input);
-            });
+            if (AUTO_OPEN.has(state.message)) autoOpen(entry);
             // An unreadable or failed preview says so in its alert; the toast never claims metadata is available.
             else if (TOASTABLE.has(state.message)) showWorkspaceToast(INSPECT_COPY.toast);
         }
     }));
     // A replacement or page leave revokes the URL after this subscriber dropped the media; a dispose closes the
     // viewer (on abort), drops the media, then ends the session.
-    scope.cleanup(() => viewer.dispose());
+    scope.cleanup(() => { const entry = endHeld(); viewer.dispose(); if (entry?.revoke) revokeObjectURL(entry.url); });
     scope.cleanup(() => { releaseMedia(media); media = null; mediaUrl = null; });
     scope.cleanup(finishAllRetiring);
     scope.cleanup(() => sessionScope.dispose());
@@ -239,9 +268,11 @@ export function mountInspect(main, { parentScope, api, sender }) {
     // cleared, so a page restored from the back-forward cache never names a file beside the empty selection.
     function leave() {
         clearRejection();
+        const entry = endHeld();
         viewer.close();
         releaseMedia(media); media = null; mediaUrl = null;
         finishAllRetiring();
+        if (entry?.revoke) revokeObjectURL(entry.url);
         session.clear();
         chosen = null; input.value = '';
     }
@@ -287,10 +318,20 @@ export function mountInspect(main, { parentScope, api, sender }) {
         for (const type of ['loadedmetadata', 'error']) preview.node.addEventListener(type, done, { once: true, signal: settled.signal });
         scope.cleanup(done);
     }
+    /** D155: the viewer opens on `entry` once its inline preview has settled, while it is still the current selection. */
+    function autoOpen(entry) {
+        whenSettled(media, () => {
+            if (scope.signal.aborted || session.entry !== entry || session.state.phase !== 'ready' || document.querySelector('dialog[open]:not(#viewer-dialog)')) return;
+            // #483: the previous viewer is still closing around its starting video; this one opens once it has.
+            if (held) { held.pending = { entry, auto: true }; return; }
+            openViewer(input);
+        });
+    }
     /** The viewer on the current selection, details showing (D155 amends D97); focus returns to `trigger` on close. */
     function openViewer(trigger) {
         const entry = session.entry;
         if (!entry || session.state.phase !== 'ready') return;
+        if (held) { held.pending = { entry, trigger, auto: false }; return; }
         viewer.open({ items: [entry], selectedKey: entry.key, trigger, details: 'shown' });
         viewerKey = entry.key;
     }

web/js/workspace/media-viewer.mjs

@@ -25,7 +25,7 @@ function element(tag, className, text) {
  * reach `onCollectionsChanged({added, removed})` once the viewer closes, so the viewer keeps its place while open.
  */
 export function createMediaViewer({api,container,parentScope,inspectionController,sender,onSelect,onRefresh,onRestoreFocus,onClose,onMediaError,adapter = rootedFileAdapter,
-    collections = null,onCollectionsChanged = null}) {
+    collections = null,onCollectionsChanged = null,onDelete = null}) {
     const scope = createScope();
     let active = null, retryVersion = 0;
     const dispose = () => scope.dispose();
@@ -82,6 +82,16 @@ export function createMediaViewer({api,container,parentScope,inspectionControlle
         const download = element('a','media-download','Download');download.setAttribute('aria-label','Download original');download.setAttribute('download','');
         local.listen(download,'click',event => {session.stop();if (download.getAttribute('aria-disabled') === 'true') event.preventDefault();});
         controls.append(download);
+        // #480: Delete… (the Images viewer only) pauses the media, then opens the delete review for this file.
+        if (onDelete) {
+            const remove = createButton('Delete…','danger');
+            local.listen(remove,'click',() => {
+                session.stop();
+                for (const video of dialog.querySelectorAll('video')) video.pause();
+                if (entry) onDelete(entry,remove);
+            });
+            controls.append(remove);
+        }
         const next = button('Next media','Next →',() => session.navigate(1));
         // D97: "Collapse viewer details by default" opens the viewer with its details hidden behind one toggle. D155
         // (#448): `details: 'shown'` (Inspect a file) keeps the toggle but opens with the details showing. D156 (#450):

web/js/workspace/workspace.css

@@ -478,6 +478,7 @@ pre.prompt-text { font: inherit; font-size: 12px; line-height: 1.75; margin-bloc
 /* D147 (#430): a state that blocks Save shows its reason on its own row above the sticky actions, with Refresh for a
    changed source; the disabled actions look disabled; a refused capture's empty preview says so. */
 .tagging-modal .dialog-actions [aria-disabled="true"], .tagging-modal .tag-model-panel [aria-disabled="true"] { opacity: .55; cursor: not-allowed; /* D147 */ }
+#modal .dialog-actions .image-delete-start[aria-disabled="true"] { opacity: .55; cursor: not-allowed; /* #480 D159: no mode chosen yet */ }
 .tagging-modal .tag-action-notice { flex: 1 0 100%; display: flex; flex-wrap: wrap; align-items: center; justify-content: space-between; gap: 8px; margin: 0; color: var(--ca-text); font-size: 12px; /* D147 */ }
 .tagging-modal .tag-action-notice > span { flex: 1 1 16em; }
 .tag-image-preview.tag-preview-empty { display: grid; place-items: center; padding: 8px; text-align: center; font-size: 12px; color: var(--ca-muted); background: var(--ca-field); box-sizing: border-box; /* D147 */ }