diff --git a/TESTING.md b/TESTING.md
index f81916ab..4d585eca 100644
--- a/TESTING.md
+++ b/TESTING.md
@@ -2460,13 +2460,40 @@ files to the system trash. **No code path unlinks or overwrites a media file.**
   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.
+  Since #501 the thumbnail jobs (System › Thumbnails' counts and its missing, rebuild and
+  convert runs, and the previews after a scan) and the legacy `GET /scan_output_dir`
+  listing skip hidden files too; an on-demand preview the legacy gallery asks for is still
+  made, and a one-image auto-tag (`/autotag/single` + `/autotag/apply`) is an explicit
+  write that shows the file again. System › Maintenance **Missing-file links** also
+  forgets the marks that hide nothing any more (the file is gone from a folder that still
+  exists, or its bytes changed), counted in its scope ("K removed images whose files are
+  gone or changed") and job line, judged only under a source that is readable and lists
+  something (an unplugged drive's empty mount point keeps its marks), never for a source
+  no longer configured, and re-checked at apply (`plan-changed`). The sidebar's Images,
+  Favorites and Needs review counts and the collection counts are read again when an
+  `image_delete` job ends in the tab that followed it (`JOB_END_COUNTS`, `state.mjs`).
+  Tests: `tests/test_image_removal_gaps.py`, the sidebar case in
+  `tests/browser/workspace-images-delete.spec.mjs`.
+- **Trash folders inside an output root** (#501, `utils/media_folders.py`): no walker
+  descends a folder below a root named `.Trash`, `.Trash-<digits>`, `.Trashes` or
+  `$RECYCLE.BIN` (any case), as none descends `thumbnails`: the workspace and legacy
+  listings (`_walk_media`, `_settled_entry`), the metadata scan and duplicate find
+  (`_collect_media_files`), the legacy auto-tag listing (`_collect_scan_files`) and the
+  delete review (`trash-folder`). The image monitor tests every folder of the new file's
+  absolute path, so it also ignores outputs written into a source that itself lies
+  inside such a folder. Tests: `tests/test_media_folders.py`,
+  `tests/test_image_removal_gaps.py`.
 - **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
+  never uses the Recycle Bin (#500, maintainer ruling (a)): files larger than the bin's
+  capacity are deleted permanently and IFileOperation's `PreDeleteItem` guard cannot detect
+  it (native probe runs 40951037575, 40953347904), so every file answers `windows-recycle`
+  ("Moving files to the Recycle Bin isn't available on Windows: files larger than the bin's
+  capacity would be deleted permanently."). macOS reports `no-trash` until #499 lands;
+  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)
diff --git a/database/image_removal.py b/database/image_removal.py
index 3360433c..58099e03 100644
--- a/database/image_removal.py
+++ b/database/image_removal.py
@@ -144,6 +144,23 @@ def removed_from_library(conn, path) -> bool:
     return False
 
 
+def mark_hides(marks: dict, locator: str, path) -> bool:
+    """Whether a removal mark in *marks* hides the file at *path* (blocking).
+
+    *marks* is ``{locator: (size, mtime_ns)}``. The mark of *locator* hides
+    the file while ``os.lstat(path)`` still has the size and ``mtime_ns`` it
+    was removed with; a file that cannot be looked at is not hidden.
+    """
+    mark = marks.get(locator)
+    if mark is None:
+        return False
+    try:
+        info = os.lstat(path)
+    except OSError:
+        return False
+    return (info.st_size, info.st_mtime_ns) == tuple(mark)
+
+
 def write_mark(conn, locator, size, mtime_ns):
     _write_mark(conn, locator, size, mtime_ns, utc_now_iso())
 
diff --git a/database/maintenance.py b/database/maintenance.py
index c2a5cdd8..27002761 100644
--- a/database/maintenance.py
+++ b/database/maintenance.py
@@ -15,6 +15,8 @@ change:
   ``cleanup_orphaned_images``), counted only under a source that can be read
   now, as the health check's ``missing_files`` judges them
   (``health.link_states``). It deletes database rows only, never a file.
+  It also forgets the removal marks (#480) that hide nothing any more: the
+  file is gone or changed, under a readable source (#501, ``_stale_marks``).
 
 ``preview_maintenance`` reads one snapshot on its own ``mode=ro`` connection
 (``health.readonly_uri``) and closes it in ``finally``; files are looked at
@@ -41,6 +43,8 @@ from pathlib import Path
 
 from .execution_contexts import LIVE_HINT_CTES
 from .health import BATCH_ROWS, link_states, readable_roots, readonly_uri
+from .image_removal_schema import TABLE as MARKS_TABLE
+from .image_removal_schema import clear_marks, has_marks_table
 from .pairs import PairDependencyError
 from .workflow_documents import documents_of_rows, release_documents
 
@@ -194,6 +198,8 @@ class MaintenanceDraft:
     protected: dict
     media: dict
     details: list = field(default_factory=list)
+    # missing_links only: locators of removal marks whose file is gone (#501).
+    marks: list = field(default_factory=list)
 
     def public(self):
         """The plan fields a response may carry (no fingerprint, no targets)."""
@@ -296,6 +302,90 @@ def _delete_ids(conn, table, ids):
     return removed
 
 
+def _mark_rows(conn, only=None):
+    """``[(locator, size, mtime_ns)]`` of the removal marks (all, or those
+    among *only*), sorted by locator."""
+    if not has_marks_table(conn):
+        return []
+    sql = f"SELECT locator, size, mtime_ns FROM {MARKS_TABLE}"  # nosec B608
+    if only is None:
+        return sorted(tuple(row) for row in conn.execute(sql))
+    found = []
+    for batch in _batches(sorted(only)):
+        found.extend(
+            tuple(row)
+            for row in conn.execute(
+                f"{sql} WHERE locator IN ({_marks(batch)})", batch  # nosec B608
+            )
+        )
+    return sorted(found)
+
+
+def _lists_something(folder):
+    """Whether *folder* holds at least one entry (an unplugged drive's mount
+    point, or an unbound volume, is an empty folder that reads fine)."""
+    try:
+        with os.scandir(folder) as entries:
+            return next(entries, None) is not None
+    except OSError:
+        return False
+
+
+def _name_listed(path):
+    """Whether *path*'s folder lists its name, compared under ``normcase``.
+
+    A locator is ``normcase``d; on a case-sensitive NTFS folder ``lstat`` of
+    the lower-cased spelling fails for a file that is there.
+    """
+    folder, name = os.path.split(path)
+    wanted = os.path.normcase(name)
+    try:
+        with os.scandir(folder) as entries:
+            return any(os.path.normcase(entry.name) == wanted for entry in entries)
+    except OSError:
+        return True  # cannot tell: keep the mark
+
+
+def _stale_mark(locator, size, mtime_ns):
+    """Whether the file of a mark is gone or has other bytes (blocking)."""
+    try:
+        info = os.lstat(locator)
+    except (FileNotFoundError, NotADirectoryError):
+        return os.path.isdir(os.path.dirname(locator)) and not _name_listed(locator)
+    except OSError:
+        return False
+    return (info.st_size, info.st_mtime_ns) != (size, mtime_ns)
+
+
+def _stale_marks(marks, roots):
+    """``[(locator, owner)]`` of the marks that hide nothing any more (#501).
+
+    *marks* are ``(locator, size, mtime_ns)`` rows. A mark is stale when its
+    file was changed, or is gone from a folder that still exists. Only marks
+    under a source that can be read now and lists something are judged (an
+    unplugged drive's marks, or its empty mount point's, stay); any doubt
+    keeps the mark. *owner* is the index of the longest source holding it.
+    Locators are ``normcase(realpath)`` paths, so the sources are too.
+    """
+    readable = readable_roots(roots)
+    prefixes = [
+        os.path.normcase(os.path.realpath(root)).rstrip("\\/") + os.sep
+        for root in roots
+    ]
+    judged = {}
+    stale = []
+    for locator, size, mtime_ns in marks:
+        owners = [i for i, prefix in enumerate(prefixes) if locator.startswith(prefix)]
+        if not owners:
+            continue  # under no source now (a source removed from the settings)
+        owner = max(owners, key=lambda i: len(prefixes[i]))
+        if owner not in judged:
+            judged[owner] = readable[owner] and _lists_something(roots[owner])
+        if judged[owner] and _stale_mark(locator, size, mtime_ns):
+            stale.append((locator, owner))
+    return stale
+
+
 def _read_snapshot(db_path, read):
     """``read(conn)`` on one ``mode=ro`` snapshot; the connection always closes."""
     conn = sqlite3.connect(readonly_uri(db_path), uri=True)
@@ -438,8 +528,11 @@ class MaintenanceOperations:
         if operation == "workflow_storage":
             return self._plan_workflow_storage()
         if operation == "missing_links":
-            rows = _read_snapshot(self.model.db_path, self._link_rows)
-            return self._plan_missing_links(rows, roots)
+            rows, marks = _read_snapshot(
+                self.model.db_path,
+                lambda conn: (self._link_rows(conn), _mark_rows(conn)),
+            )
+            return self._plan_missing_links(rows, roots, _stale_marks(marks, roots))
         reader = (
             self._plan_duplicate_prompts
             if operation == "duplicate_prompts"
@@ -570,13 +663,15 @@ class MaintenanceOperations:
         )
 
     @staticmethod
-    def _links_fingerprint(candidates):
+    def _links_fingerprint(candidates, stale=()):
         fingerprint = _Fingerprint("missing_links")
         for row, owner in candidates:
             fingerprint.add([row[0], row[1], row[3], owner])
+        for locator, owner in stale:
+            fingerprint.add(["mark", locator, owner])
         return fingerprint.hexdigest()
 
-    def _plan_missing_links(self, rows, roots):
+    def _plan_missing_links(self, rows, roots, stale=()):
         readable = readable_roots(roots)
         candidates, unavailable, outside = [], 0, 0
         for row, owner, state in link_states(rows, roots, readable):
@@ -604,29 +699,47 @@ class MaintenanceOperations:
                 )
             )
         count = len(candidates)
+        parts = []
+        if count:
+            parts.append(
+                _plural(count, "link to a missing file", "links to missing files")
+            )
+        if stale:
+            parts.append(
+                _plural(
+                    len(stale),
+                    "removed image whose file is gone or changed",
+                    "removed images whose files are gone or changed",
+                )
+            )
+        details = [
+            {
+                "label": _relative(row[1], roots[owner]),
+                "detail": f"{_source_name(roots[owner], owner)} · the file is missing",
+            }
+            for row, owner in candidates
+        ] + [
+            {
+                "label": _relative(locator, os.path.realpath(roots[owner])),
+                "detail": (
+                    f"{_source_name(roots[owner], owner)} · removed image, "
+                    "the file is gone or changed"
+                ),
+            }
+            for locator, owner in stale
+        ]
         return MaintenanceDraft(
             operation="missing_links",
-            fingerprint=self._links_fingerprint(candidates),
+            fingerprint=self._links_fingerprint(candidates, stale),
             targets=[row[0] for row, _owner in candidates],
             scope={
-                "count": count,
-                "label": (
-                    _plural(count, "link to a missing file", "links to missing files")
-                    if count
-                    else "No links to missing files"
-                ),
+                "count": count + len(stale),
+                "label": "; ".join(parts) or "No links to missing files",
             },
             protected={"count": unavailable + outside, "reasons": reasons},
             media={"deletes_files": False, "label": LINKS_ONLY},
-            details=[
-                {
-                    "label": _relative(row[1], roots[owner]),
-                    "detail": (
-                        f"{_source_name(roots[owner], owner)} · the file is missing"
-                    ),
-                }
-                for row, owner in candidates[:MAX_DETAILS]
-            ],
+            details=details[:MAX_DETAILS],
+            marks=[locator for locator, _owner in stale],
         )
 
     def _plan_optimize(self):
@@ -676,14 +789,18 @@ class MaintenanceOperations:
 
     # ── apply ─────────────────────────────────────────────────────────
 
-    def apply_maintenance(self, operation, fingerprint, targets, roots=()) -> dict:
+    def apply_maintenance(
+        self, operation, fingerprint, targets, roots=(), marks=()
+    ) -> dict:
         """Apply a reviewed plan exactly, or raise PlanChanged (blocking).
 
         Inside ``BEGIN IMMEDIATE`` the fingerprint is computed again on the
         write connection; when it differs nothing is written. ``targets`` and
         ``roots`` are the plan's. Returns ``{changed, rows}``: how many
         records changed and up to ``MAX_RESULT_ROWS`` descriptions of them
-        (``{prompt_id[, kept_id]}`` or ``{image_path}``).
+        (``{prompt_id[, kept_id]}`` or ``{image_path}``). ``marks`` are a
+        missing-links plan's stale removal marks; that apply also returns
+        ``marks``, how many it forgot.
         """
         if operation not in MAINTENANCE_OPERATIONS or operation == "workflow_storage":
             raise ValueError(f"Unknown maintenance operation {operation!r}")
@@ -696,9 +813,12 @@ class MaintenanceOperations:
         appliers = {
             "duplicate_prompts": self._apply_duplicate_prompts,
             "unused_components": self._apply_unused_components,
-            "missing_links": self._apply_missing_links,
         }
         with self._pair_transaction() as conn:
+            if operation == "missing_links":
+                return self._apply_missing_links(
+                    conn, fingerprint, targets, roots, list(marks)
+                )
             return appliers[operation](conn, fingerprint, targets, roots)
 
     def _apply_duplicate_prompts(self, conn, fingerprint, targets, _roots):
@@ -727,7 +847,7 @@ class MaintenanceOperations:
         ]
         return {"changed": changed, "rows": rows}
 
-    def _apply_missing_links(self, conn, fingerprint, targets, roots):
+    def _apply_missing_links(self, conn, fingerprint, targets, roots, marks):
         rows = []
         for batch in _batches(sorted(targets)):
             rows.extend(
@@ -744,10 +864,14 @@ class MaintenanceOperations:
             if state != "missing":  # back, or its source went away
                 raise PlanChanged("missing_links")
             candidates.append((row, owner))
-        if self._links_fingerprint(candidates) != fingerprint:
+        stale = _stale_marks(_mark_rows(conn, marks) if marks else [], roots)
+        if [locator for locator, _owner in stale] != marks:  # a mark went or is valid
+            raise PlanChanged("missing_links")
+        if self._links_fingerprint(candidates, stale) != fingerprint:
             raise PlanChanged("missing_links")
         changed = _delete_ids(conn, "generated_images", [row[0] for row in rows])
         return {
             "changed": changed,
+            "marks": clear_marks(conn, marks),
             "rows": [{"image_path": row[1]} for row in rows[:MAX_RESULT_ROWS]],
         }
diff --git a/py/api/admin.py b/py/api/admin.py
index a6b123d2..a9f820e2 100644
--- a/py/api/admin.py
+++ b/py/api/admin.py
@@ -36,10 +36,13 @@ except ImportError:
 
 try:
     from ...utils import exit_signal
+    from ...utils.media_folders import is_skipped_folder
 except ImportError:
     from utils import exit_signal
+    from utils.media_folders import is_skipped_folder
 
 try:
+    from ...database.image_removal import mark_hides
     from ...database.pair_ingestion import SCAN_MARKS, ingest_resolved_pair
     from ...database.run_attribution import (
         RecordedOutput,
@@ -60,6 +63,7 @@ try:
         decide_file_prompt,
     )
 except ImportError:
+    from database.image_removal import mark_hides
     from database.pair_ingestion import SCAN_MARKS, ingest_resolved_pair
     from database.run_attribution import (
         RecordedOutput,
@@ -118,10 +122,11 @@ DUPLICATE_FIND_LEGACY_TITLE = "Find duplicate image files (gallery page)"
 def _collect_media_files(output_dirs, extensions):
     """Media files under ``output_dirs`` (blocking), bounded and symlink-free.
 
-    Walks each root without following symlinks, skips ``thumbnails``
-    directories and symlinked files, never descends more than MAX_SCAN_DEPTH
-    levels and stops after MAX_SCAN_FILES files in total. Extensions match
-    case-insensitively and each path is listed once.
+    Walks each root without following symlinks, skips ``thumbnails`` and
+    trash folders (``utils.media_folders``) and symlinked files, never
+    descends more than MAX_SCAN_DEPTH levels and stops after MAX_SCAN_FILES
+    files in total. Extensions match case-insensitively and each path is
+    listed once.
     """
     wanted = {ext.lower() for ext in extensions}
     found = []
@@ -136,7 +141,7 @@ def _collect_media_files(output_dirs, extensions):
             dirnames[:] = sorted(
                 d
                 for d in dirnames
-                if d != "thumbnails"
+                if not is_skipped_folder(d)
                 and depth < MAX_SCAN_DEPTH
                 and not os.path.islink(os.path.join(current, d))
             )
@@ -893,11 +898,8 @@ class AdminRoutesMixin:
                     stat = os.stat(path)
                 except OSError:
                     continue
-                mark = marks.get(os.path.normcase(os.path.realpath(str(path))))
-                if mark is not None and tuple(mark) == (
-                    stat.st_size,
-                    stat.st_mtime_ns,
-                ):
+                locator = os.path.normcase(os.path.realpath(str(path)))
+                if locator in marks and mark_hides(marks, locator, path):
                     continue
                 key = _physical_file_key(path, stat)
                 if key not in seen:
diff --git a/py/api/autotag_routes.py b/py/api/autotag_routes.py
index edc8ca82..d3ff1629 100644
--- a/py/api/autotag_routes.py
+++ b/py/api/autotag_routes.py
@@ -23,6 +23,11 @@ from .prompts import thumbnail_url_for
 from . import thumbnail_settings
 from .. import model_gate
 
+try:
+    from ...utils.media_folders import is_skipped_folder
+except ImportError:
+    from utils.media_folders import is_skipped_folder
+
 from aiohttp import web
 
 _INTERNAL_ERROR = "An internal error occurred. Check server logs for details."
@@ -1319,8 +1324,9 @@ class AutotagRoutesMixin:
     def _collect_scan_files(output_path: Path):
         """Media files under ``output_path`` from a bounded, symlink-free walk.
 
-        Descends at most ``_SCAN_MAX_DEPTH`` levels, skips ``thumbnails``
-        trees and symlinked files, and stops after ``_SCAN_MAX_FILES`` files.
+        Descends at most ``_SCAN_MAX_DEPTH`` levels, skips ``thumbnails`` and
+        trash folders (``utils.media_folders``) and symlinked files, and stops
+        after ``_SCAN_MAX_FILES`` files.
         Returns ``(paths, truncated)``.
         """
         found = []
@@ -1330,7 +1336,7 @@ class AutotagRoutesMixin:
             dirnames[:] = sorted(
                 name
                 for name in dirnames
-                if name != "thumbnails" and depth + 1 <= _SCAN_MAX_DEPTH
+                if not is_skipped_folder(name) and depth + 1 <= _SCAN_MAX_DEPTH
             )
             for name in sorted(filenames):
                 if Path(name).suffix.lower() not in _MEDIA_EXTENSIONS:
@@ -1357,6 +1363,10 @@ class AutotagRoutesMixin:
             files, truncated = await self._run_in_executor(
                 self._collect_scan_files, output_path
             )
+            # #501: a file removed from the library is not offered for tagging.
+            files = await self._run_in_executor(
+                self._without_removed, output_path, files
+            )
             has_thumbnails = (output_path / "thumbnails").is_dir()
 
             images = []
diff --git a/py/api/image_delete.py b/py/api/image_delete.py
index fe011149..5227f966 100644
--- a/py/api/image_delete.py
+++ b/py/api/image_delete.py
@@ -47,12 +47,16 @@ from .maintenance import MAINTENANCE_PREVIEW_LEASE, _fail_job, _failure, backup_
 
 try:
     from ...database import image_removal_schema
+    from ...database.image_removal import mark_hides
     from ...database.annotation_schema import MAX_LOCATOR_LENGTH
     from ...utils import trash
+    from ...utils.media_folders import is_trash_folder
 except ImportError:
     from database import image_removal_schema
+    from database.image_removal import mark_hides
     from database.annotation_schema import MAX_LOCATOR_LENGTH
     from utils import trash
+    from utils.media_folders import is_trash_folder
 
 IMAGE_DELETE = "delete_images"
 MODES = ("library", "trash")
@@ -61,7 +65,6 @@ TITLES = {
     "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 = {
@@ -80,6 +83,10 @@ REASONS = {
     trash.TRASH_IN_ROOT: "This drive's trash is inside an output folder.",
     trash.LOCKED: "The file is open in another program. Close it and try again.",
     trash.FAILED: "The file could not be moved to the trash.",
+    trash.WINDOWS_RECYCLE: (
+        "Moving files to the Recycle Bin isn't available on Windows: files larger "
+        "than the bin's capacity would be deleted permanently."
+    ),
     "not-attempted": "Not attempted: the job stopped after an error.",
 }
 _PROBLEM_CODES = {
@@ -93,19 +100,12 @@ SOURCES_CHANGED = "The sources changed since the review; nothing was changed."
 RECORDS_FAILED = "The image records could not be removed; the job stopped."
 
 
-def _is_trash_folder(name: str) -> bool:
-    folded = name.casefold()
-    return folded in TRASH_FOLDERS or (
-        folded.startswith(".trash-") and folded[7:].isdigit()
-    )
-
-
 def lexical_problem(relative: str) -> str | None:
     """The refusal code a root-relative POSIX path earns before any file access."""
     parts = relative.split("/")
     if "thumbnails" in parts[:-1]:
         return "thumbnail"
-    if any(_is_trash_folder(part) for part in parts[:-1]):
+    if any(is_trash_folder(part) for part in parts[:-1]):
         return "trash-folder"
     if os.path.splitext(parts[-1])[1].lower() not in DELETABLE_MEDIA_EXTENSIONS:
         return "not-media"
@@ -443,37 +443,49 @@ class ImageDeleteMixin:
         if not marks:
             return set()
         located = self._workspace_locators(listings)
-        hidden = set()
-        for locator, (size, mtime_ns) in marks.items():
-            entry = located.get(locator)
-            if entry is None:
-                continue
-            try:
-                info = os.lstat(entry.path)
-            except OSError:
-                continue
-            if (info.st_size, info.st_mtime_ns) == (size, mtime_ns):
-                hidden.add(entry.normalized)
-        return hidden
+        return {
+            entry.normalized
+            for locator, entry in ((locator, located.get(locator)) for locator in marks)
+            if entry is not None and mark_hides(marks, locator, entry.path)
+        }
 
     def _removed_paths(self, paths) -> set:
         """The absolute paths among *paths* a removal mark hides (blocking)."""
         marks = self._removal_marks()
         if not marks:
             return set()
-        hidden = set()
-        for path in paths:
-            mark = marks.get(os.path.normcase(os.path.realpath(str(path))))
-            if mark is None:
-                continue
-            try:
-                info = os.lstat(path)
-            except OSError:
-                continue
-            if (info.st_size, info.st_mtime_ns) == tuple(mark):
-                hidden.add(str(path))
+        return {
+            str(path)
+            for path in paths
+            if mark_hides(marks, os.path.normcase(os.path.realpath(str(path))), path)
+        }
+
+    def _removal_check_under(self, root):
+        """``hidden(path) -> bool`` for files a walk of *root* found (blocking).
+
+        A walk never follows a link below *root*, so a file's locator is
+        ``normcase(join(realpath(root), relative))``: no ``realpath`` per
+        file, and an ``lstat`` only for a marked one. Without marks it
+        answers False at once.
+        """
+        marks = self._removal_marks()
+        if not marks:
+            return lambda _path: False
+        root = os.fspath(root)
+        real = os.path.realpath(root)
+
+        def hidden(path):
+            relative = os.path.relpath(os.fspath(path), root)
+            locator = os.path.normcase(os.path.join(real, relative))
+            return mark_hides(marks, locator, path)
+
         return hidden
 
+    def _without_removed(self, root, paths) -> list:
+        """*paths* found by a walk of *root*, less the hidden ones (blocking)."""
+        removed = self._removal_check_under(root)
+        return [path for path in paths if not removed(path)]
+
 
 def parse_mode(value):
     """The apply's ``mode``, or None when it is not one of MODES."""
diff --git a/py/api/images.py b/py/api/images.py
index 38286e2e..eb5636d9 100644
--- a/py/api/images.py
+++ b/py/api/images.py
@@ -66,6 +66,7 @@ try:
     from ...database.models import normalize_image_path
     from ...database.run_attribution import recorded_output_lookup
     from ...utils.card_seed import card_seed, header_params, parsed_positive
+    from ...utils.media_folders import in_skipped_folder, is_skipped_folder
     from ...utils.parallel import map_parallel
     from ...utils.json_strict import dumps_strict
     from ...utils.inspection_response import (
@@ -91,6 +92,7 @@ except ImportError:
     from database.models import normalize_image_path
     from database.run_attribution import recorded_output_lookup
     from utils.card_seed import card_seed, header_params, parsed_positive
+    from utils.media_folders import in_skipped_folder, is_skipped_folder
     from utils.parallel import map_parallel
     from utils.json_strict import dumps_strict
     from utils.inspection_response import (
@@ -915,7 +917,8 @@ def _name_suffix(name):
 def _walk_media(output_path, max_files):
     """Yield the ``os.DirEntry`` of each media file under *output_path*.
 
-    Symlinks are never followed, ``thumbnails/`` is skipped, descent stops
+    Symlinks are never followed, ``thumbnails/`` and trash folders are
+    skipped (``utils.media_folders``, #501), descent stops
     at MAX_SCAN_DEPTH and at most *max_files* entries are produced. The order
     is ``os.walk``'s (top-down, subfolders in ``os.scandir`` order, files by
     name), which the gallery's ties in mtime keep. Each entry's type comes
@@ -927,8 +930,6 @@ def _walk_media(output_path, max_files):
     produced = 0
     while stack:
         top, rel_parts = stack.pop()
-        if "thumbnails" in rel_parts:
-            continue
         dirs, files = [], []
         try:
             with os.scandir(top) as entries:
@@ -952,7 +953,7 @@ def _walk_media(output_path, max_files):
         if len(rel_parts) >= MAX_SCAN_DEPTH:
             continue
         for entry in reversed(dirs):
-            if not entry.is_symlink():
+            if not entry.is_symlink() and not is_skipped_folder(entry.name):
                 stack.append((entry.path, rel_parts + (entry.name,)))
 
 
@@ -976,7 +977,7 @@ def _settled_entry(root, relative):
     *folders, name = relative.split(os.sep)
     if _name_suffix(name).lower() not in MEDIA_EXTENSIONS:
         return None
-    if "thumbnails" in folders or len(folders) > MAX_SCAN_DEPTH:
+    if in_skipped_folder(folders) or len(folders) > MAX_SCAN_DEPTH:
         return None
     folder = os.fspath(root)
     for part in ("", *folders):
@@ -3075,7 +3076,9 @@ class ImageRoutesMixin:
         no symlinks), each preview folder is listed once, and at most
         MAX_THUMBNAIL_FILES targets are queued per batch (a job plans again
         after each batch); ``remaining`` counts the targets beyond that cap.
-        Sources in *exclude* (worked on by an earlier batch) are left out.
+        Sources in *exclude* (worked on by an earlier batch) are left out, and
+        so are files removed from the library (#501: a removal mark hides
+        them, and their previews went with their records).
         ``stop()`` true ends the walk before the next file.
         """
         settings = writer_settings(settings)
@@ -3084,11 +3087,14 @@ class ImageRoutesMixin:
         plan = {key: 0 for key in _PLAN_COUNTS}
         plan.update(targets=[], fresh_files=[])
         folders = {}
+        removed = self._removal_check_under(output_path)
         for src in _iter_media_files(output_path, MAX_GALLERY_FILES):
             if stop is not None and stop():
                 break
             if exclude and src in exclude:
                 continue
+            if removed(src):
+                continue
             rel_no_ext = src.relative_to(output_path).with_suffix("")
             stem = f"{rel_no_ext.name}{thumbnail_settings.THUMB_SUFFIX}"
             dst = thumbnails_dir / rel_no_ext.parent / (stem + settings.extension)
diff --git a/py/api/maintenance.py b/py/api/maintenance.py
index 4a518efd..e7d34020 100644
--- a/py/api/maintenance.py
+++ b/py/api/maintenance.py
@@ -281,7 +281,7 @@ def short_space_message(short):
     )
 
 
-def _changed_line(operation, count):
+def _changed_line(operation, count, marks=0):
     if operation == "optimize":
         return "Optimized the database"
     if operation == "duplicate_prompts":
@@ -291,7 +291,11 @@ def _changed_line(operation, count):
         noun = "unused prompt component" if count == 1 else "unused prompt components"
         return f"Removed {count:,} {noun}"
     noun = "link to a missing file" if count == 1 else "links to missing files"
-    return f"Removed {count:,} {noun}"
+    line = f"Removed {count:,} {noun}"
+    if marks:
+        gone = "removed image" if marks == 1 else "removed images"
+        line += f"; forgot {marks:,} {gone} whose files are gone or changed"
+    return line
 
 
 def take_safety_backup(model, label="backup"):
@@ -419,6 +423,7 @@ class MaintenanceRoutesMixin:
             draft.public(),
             {
                 "targets": draft.targets,
+                "marks": draft.marks,
                 "roots": roots,
                 "protected": draft.protected["count"],
                 "rows": draft.scope["count"],
@@ -653,6 +658,7 @@ class MaintenanceRoutesMixin:
                 plan.fingerprint,
                 payload["targets"],
                 payload["roots"],
+                payload.get("marks", ()),
             )
         except PlanChanged:
             return _fail_job(job, "plan-changed", PLAN_CHANGED)
@@ -664,7 +670,7 @@ class MaintenanceRoutesMixin:
         )
         job.record_counted("changed", result["changed"], rows)
         job.record_counted("protected", payload["protected"])
-        line = _changed_line(plan.operation, result["changed"])
+        line = _changed_line(plan.operation, result["changed"], result.get("marks", 0))
         job.log_line(line)
         job.update(progress=100, status_line=f"{line} · {backup_note(backup)}")
 
diff --git a/tests/browser/workspace-images-delete.spec.mjs b/tests/browser/workspace-images-delete.spec.mjs
index bf727731..7051166b 100644
--- a/tests/browser/workspace-images-delete.spec.mjs
+++ b/tests/browser/workspace-images-delete.spec.mjs
@@ -73,6 +73,19 @@ for (const mode of ['library', 'trash']) {
     });
 }
 
+// #501: the sidebar's Images count (read once at mount) follows a finished delete without a reload.
+test('a finished delete re-reads the sidebar Images count', async ({ page }) => {
+    await openImages(page);
+    const badge = page.locator('#primary-nav a[data-route="images"] .nav-count');
+    await expect(badge).toHaveText(/^\d[\d,]*$/);
+    const before = Number((await badge.textContent()).replace(/,/g, ''));
+    await check(page, 0);
+    const { primary } = await deleteChecked(page, 'library');
+    await primary.click();
+    await expect(toast(page)).toContainText('Removed 1 image from the library.');
+    await expect(badge).toHaveText((before - 1).toLocaleString('en-US'));
+});
+
 test('Cancel changes nothing', async ({ page }) => {
     const before = await openImages(page);
     await check(page, 0);
diff --git a/tests/browser/workspace-library-editor.spec.mjs b/tests/browser/workspace-library-editor.spec.mjs
index 2a17b3e3..725b5427 100644
--- a/tests/browser/workspace-library-editor.spec.mjs
+++ b/tests/browser/workspace-library-editor.spec.mjs
@@ -635,7 +635,11 @@ test('restored storage after a confirmed save or before an archive offers an exp
         const response=await route.fetch();await expect(await restore()).toBeOK();return route.fulfill({response});
     });
     await page.getByRole('button',{name:'Edit metadata',exact:true}).click();let dialog=editorDialog(page);await ready(dialog);
-    await field(dialog,'Notes').fill('Lost to a restore');await button(dialog,'Save').click();
+    await field(dialog,'Notes').fill('Lost to a restore');
+    // The save's answer waits for a whole restore in the route above; wait for that answer, not the 5 s expect
+    // default (#501 merge queue: under load the restore outlasted it on WebKit).
+    const saved=page.waitForResponse(item=>item.request().method()==='PATCH'&&new URL(item.url()).pathname===`/prompt_manager/library/${f.positive_id}`);
+    await button(dialog,'Save').click();await saved;
     await expect(note(dialog)).toHaveText('Saved; Library refresh failed because prompt storage was restored. Reload the Library to see current data.');
     expect(await actionNames(dialog)).toEqual(['Reload Library','Create text revision','Archive…','Cancel','Save']);
     for(const name of ['Create text revision','Archive…'])await expect(button(dialog,name)).toBeDisabled();
diff --git a/tests/browser/workspace-review.spec.mjs b/tests/browser/workspace-review.spec.mjs
index ade5f32a..96f4b638 100644
--- a/tests/browser/workspace-review.spec.mjs
+++ b/tests/browser/workspace-review.spec.mjs
@@ -221,6 +221,33 @@ test('tag image from review saves tags and the item remains pending', async ({ p
     await expect(dialog).toHaveCount(0);
 });
 
+test('#515: tags typed before the image observation answers are saved', async ({ page, request }) => {
+    await seed(request, 1);
+    await open(page);
+    // Hold the dialog's observation so the typing lands while it is pending (a slow host or WebKit in CI).
+    let release;
+    const held = new Promise(resolve => { release = resolve; });
+    const matcher = '**/prompt_manager/image-annotations/observe';
+    let reachedObserve;
+    const reached = new Promise(resolve => { reachedObserve = resolve; });
+    await page.route(matcher, async route => { reachedObserve(); await held; await route.continue(); });
+    await panel(page).getByRole('button', { name: 'Tag image', exact: true }).click();
+    const dialog = page.getByRole('dialog', { name: 'Tag image', exact: true });
+    await reached;
+    const field = dialog.getByRole('textbox', { name: 'Image tags · edit manually', exact: true });
+    await field.fill('dunes');
+    const save = dialog.getByRole('button', { name: 'Save manual tags', exact: true });
+    await expect(save).toHaveAttribute('aria-disabled', 'true');
+    const answered = page.waitForResponse(item => item.url().endsWith('/image-annotations/observe'));
+    release();
+    await answered;
+    await expect(save).not.toHaveAttribute('aria-disabled', 'true');
+    await expect(field).toHaveValue('dunes');
+    await save.click();
+    await expect(toast(page)).toHaveText('1 tags saved for review_0000.png. Prompt association still awaits review.');
+    await page.unroute(matcher);
+});
+
 test('an empty queue shows the empty notice only', async ({ page, request }) => {
     await seed(request, 0);
     await page.goto(review);
diff --git a/tests/browser/workspace-system-diagnostics.spec.mjs b/tests/browser/workspace-system-diagnostics.spec.mjs
index 9f811362..2f4cc07a 100644
--- a/tests/browser/workspace-system-diagnostics.spec.mjs
+++ b/tests/browser/workspace-system-diagnostics.spec.mjs
@@ -274,11 +274,18 @@ test('level filter is minimum-level over all six choices', async ({ page, reques
     for (const line of (await lines(page).textContent()).split('\n')) expect(line).toMatch(/^\d\d:\d\d:\d\d [A-Z]+ +\S/);
     expect(urls.at(-1)).toBe('?limit=100');
     for (const minimum of LEVELS) {
+        // Wait for this level's answer: the previous level's lines also hold the CRITICAL line, so seeing it proved
+        // nothing, and under load the filtered answer painted after the check (#501 merge queue, firefox).
+        const answered = page.waitForResponse(item => new URL(item.url()).pathname === LOGS_URL
+            && new URL(item.url()).search === `?limit=100&level=${minimum}`);
         await level.selectOption(minimum);
+        await answered;
         await expect.poll(() => urls.at(-1)).toBe(`?limit=100&level=${minimum}`);
         await expect(lines(page)).toContainText('CRITICAL The database could not be opened');
-        const shown = await shownLevels(page);
-        expect(shown.every(name => LEVELS.indexOf(name) >= LEVELS.indexOf(minimum)), `${minimum}: ${shown}`).toBe(true);
+        await expect.poll(async () => {
+            const shown = await shownLevels(page);
+            return shown.every(name => LEVELS.indexOf(name) >= LEVELS.indexOf(minimum)) || `${minimum}: ${shown}`;
+        }).toBe(true);
         if (minimum === 'DEBUG') await expect(lines(page)).toContainText('DEBUG Thumbnail cache lookup');
         if (minimum === 'WARNING') {
             await expect(lines(page)).not.toContainText('Database backup verified');
diff --git a/tests/js/workspace-image-tagging-state.test.js b/tests/js/workspace-image-tagging-state.test.js
index 4b7ea2ac..0f4a033a 100644
--- a/tests/js/workspace-image-tagging-state.test.js
+++ b/tests/js/workspace-image-tagging-state.test.js
@@ -876,3 +876,63 @@ test('retry after a changed storage, file or expired capture observes again and
     await settle();
     assert.equal(c.state.manual, 'Minimal, warm_light');
 });
+
+test('#515: text typed while the first observation is pending is kept, after the saved tags', async () => {
+    const { createImageTaggingController } = await load();
+    const answers = [deferred(), deferred()];
+    let n = 0;
+    const h = harness(defaultRoutes({ [`POST ${OBSERVE}`]: () => answers[n++].promise }));
+    // An untagged file: the typed text is the whole draft and Save sends it.
+    const bare = createImageTaggingController({ api: h.api, target: TARGET, source: 'review' });
+    const first = bare.open();
+    bare.setManual('dunes');
+    answers[0].resolve(response(200, observed({ tags: [], tag_revision: 0 })));
+    await first;
+    await settle();
+    assert.equal(bare.state.phase, 'ready');
+    assert.equal(bare.state.manual, 'dunes');
+    assert.equal(bare.state.statusText, '', 'nothing changed elsewhere');
+    await bare.save();
+    await settle();
+    assert.deepEqual(h.of('PUT', TAGS).at(-1).body.tags, ['dunes']);
+    // A tagged file: its saved tags lead, so a Save never drops tags the user had not seen yet.
+    const tagged = createImageTaggingController({ api: h.api, target: TARGET });
+    const second = tagged.open();
+    tagged.setManual('dunes, minimal');
+    answers[1].resolve(response(200, observed()));
+    await second;
+    await settle();
+    assert.equal(tagged.state.manual, 'Minimal, warm_light, dunes');
+    assert.equal(tagged.state.statusText, '');
+});
+
+test('#515: a kept draft takes the text typed while the reopen observes', async () => {
+    const { c } = await opened({ [`PUT ${TAGS}`]: () => fail(409, 'file-changed', 'The file changed.') });
+    c.setManual('kept draft');
+    await c.save();
+    await settle();
+    const reopened = c.retry();
+    c.setManual('kept draft, later');
+    await reopened;
+    await settle();
+    assert.equal(c.state.phase, 'ready');
+    assert.equal(c.state.manual, 'kept draft, later');
+});
+
+test('#515: text typed before Retry after a failed first observation follows the saved tags', async () => {
+    const { createImageTaggingController } = await load();
+    let n = 0;
+    const h = harness(defaultRoutes({
+        [`POST ${OBSERVE}`]: () => (n++ === 0 ? fail(500, 'internal', 'boom') : response(200, observed())),
+    }));
+    const c = createImageTaggingController({ api: h.api, target: TARGET });
+    await c.open();
+    await settle();
+    assert.equal(c.state.phase, 'observe-error');
+    c.setManual('dunes');
+    await c.retry();
+    await settle();
+    assert.equal(c.state.phase, 'ready');
+    assert.equal(c.state.manual, 'Minimal, warm_light, dunes');
+    assert.equal(c.state.statusText, '');
+});
diff --git a/tests/js/workspace-images-delete.test.js b/tests/js/workspace-images-delete.test.js
index 90782294..d03e8c81 100644
--- a/tests/js/workspace-images-delete.test.js
+++ b/tests/js/workspace-images-delete.test.js
@@ -65,6 +65,8 @@ test('sentences follow the mode and the backup policy', () => {
     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]);
+    // #500: no "yet" (Windows is unavailable by design); each file's reason says why.
+    assert.equal(NO_TRASH_HERE, 'Moving to the trash is not available for these files; each one lists why.');
     assert.deepEqual(deleteSentences(plan({ trash_refused: { count: 0 } }), 'trash'), [SENTENCES.trash]);
 });
 
diff --git a/tests/test_autotag_engine.py b/tests/test_autotag_engine.py
index ec3e468e..5f8b2bde 100644
--- a/tests/test_autotag_engine.py
+++ b/tests/test_autotag_engine.py
@@ -8,6 +8,7 @@ import contextlib
 import csv
 import os
 import shutil
+import subprocess
 import sys
 import tempfile
 import types
@@ -15,7 +16,18 @@ import unittest
 from pathlib import Path
 from unittest.mock import MagicMock, patch
 
-sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+_REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+sys.path.insert(0, _REPO_ROOT)
+
+# Run in a child interpreter: import py.autotag with a stub ComfyUI ``server``
+# and print every ML backend that the import pulled in.
+_IMPORT_AUTOTAG_REPORT_BACKENDS = """
+import sys
+from unittest.mock import MagicMock
+sys.modules["server"] = MagicMock()
+import py.autotag
+print(*sorted({"onnxruntime", "llama_cpp"} & set(sys.modules)))
+"""
 
 from PIL import Image
 
@@ -142,17 +154,26 @@ class TestModelInjection(EngineTestCase):
         self.assertEqual(service._model_loader, service._load_model)
 
     def test_importing_autotag_does_not_import_ml_backends(self):
-        for heavy in ("onnxruntime", "llama_cpp"):
-            self.assertNotIn(heavy, sys.modules)
+        # A fresh interpreter: earlier tests in this process may have imported
+        # an installed backend, so only a clean import can prove this (#472).
+        result = subprocess.run(
+            [sys.executable, "-c", _IMPORT_AUTOTAG_REPORT_BACKENDS],
+            cwd=_REPO_ROOT,
+            capture_output=True,
+            text=True,
+            timeout=120,
+        )
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual(result.stdout.split(), [])
 
     def test_default_wd14_loader_reports_missing_onnxruntime(self):
         _touch_model_files(self.models_dir, "wd14-vit")
         service = AutoTagService(models_dir=self.models_dir)
-        if "onnxruntime" in sys.modules:
-            self.skipTest("onnxruntime is installed in this environment")
 
-        with self.assertRaises(RuntimeError) as ctx:
-            service.load_model("wd14-vit")
+        # A None entry makes onnxruntime unimportable even where it is installed.
+        with _patch_modules({"onnxruntime": None}):
+            with self.assertRaises(RuntimeError) as ctx:
+                service.load_model("wd14-vit")
 
         self.assertIn("onnxruntime", str(ctx.exception))
         self.assertFalse(service.is_model_loaded())
diff --git a/tests/test_autotag_packages.py b/tests/test_autotag_packages.py
index fb910ce8..c5336439 100644
--- a/tests/test_autotag_packages.py
+++ b/tests/test_autotag_packages.py
@@ -1,6 +1,7 @@
 """#424: each tagging backend reports the Python packages it cannot import.
 
-``find_spec`` is patched throughout: no ML backend is installed or imported.
+``find_spec`` is patched throughout, and a backend an earlier test imported is
+hidden where it matters, so these pass with or without the ML packages (#472).
 """
 
 import os
@@ -75,8 +76,13 @@ class TestMissingPackages(unittest.TestCase):
     def test_find_spec_errors_count_as_missing(self):
         for error in (ValueError("no spec"), ImportError("broken"), OSError("io")):
             with self.subTest(error=type(error).__name__):
-                with patch.object(
-                    autotag.importlib.util, "find_spec", side_effect=error
+                # Hide an onnxruntime an earlier test imported, so the check
+                # reaches find_spec (#472).
+                with (
+                    _patch_modules({"onnxruntime": None}),
+                    patch.object(
+                        autotag.importlib.util, "find_spec", side_effect=error
+                    ),
                 ):
                     self.assertEqual(missing_packages("wd14-vit"), ["onnxruntime"])
 
diff --git a/tests/test_bulk_library_delete.py b/tests/test_bulk_library_delete.py
index 35b9abe6..0e4bfc08 100644
--- a/tests/test_bulk_library_delete.py
+++ b/tests/test_bulk_library_delete.py
@@ -64,14 +64,15 @@ def table_dump(db):
 class LibraryDeleteCase(unittest.TestCase):
     def setUp(self):
         self.temp = tempfile.TemporaryDirectory()
+        # Cleanups run last-in first-out and after every cleanup a test adds,
+        # so a database a test opens (and closes via addCleanup) is closed
+        # before the folder goes; Windows cannot delete an open file (#510).
+        self.addCleanup(self.temp.cleanup)
         self.root = Path(self.temp.name)
         self.db = PromptDatabase(str(self.root / "delete.db"))
+        self.addCleanup(self.db.close_all)
         self.epoch = self.db.get_library_options()["library_epoch"]
 
-    def tearDown(self):
-        self.db.close_all()
-        self.temp.cleanup()
-
     def paired(self, positive, negative="blur"):
         pair = self.db.save_prompt_pair(positive, negative)
         found = self.db.get_prompt_pair(pair)
diff --git a/tests/test_image_delete_routes.py b/tests/test_image_delete_routes.py
index 1fb364df..53f34471 100644
--- a/tests/test_image_delete_routes.py
+++ b/tests/test_image_delete_routes.py
@@ -306,6 +306,13 @@ class TestFileCheck(ImageDeleteTestCase):
             file_check(str(self.output_dir / "ok.png"), root, "ok.png")[0]
         )
 
+    def test_windows_reason_says_why_plainly(self):
+        self.assertEqual(
+            image_delete.REASONS["windows-recycle"],
+            "Moving files to the Recycle Bin isn't available on Windows: files "
+            "larger than the bin's capacity would be deleted permanently.",
+        )
+
     def test_reasons_cover_every_code(self):
         from utils import trash
 
@@ -324,7 +331,8 @@ class TestNativeBackend(ImageDeleteTestCase):
         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})
+            code = "windows-recycle" if sys.platform == "win32" else "no-trash"
+            self.assertEqual(plan["trash_refused"]["counts"], {code: 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)
diff --git a/tests/test_image_removal_gaps.py b/tests/test_image_removal_gaps.py
new file mode 100644
index 00000000..cb961720
--- /dev/null
+++ b/tests/test_image_removal_gaps.py
@@ -0,0 +1,251 @@
+"""Removal-mark gaps after #480, and trash folders inside output roots (#501).
+
+- Thumbnail jobs and the legacy auto-tag listing skip a library-removed file.
+- Missing-file links also forgets the marks of files that are gone (only under
+  a source that can be read now).
+- Every walker skips a trash folder that sits inside an output root.
+"""
+
+import asyncio
+import os
+import shutil
+import sqlite3
+from pathlib import Path
+from unittest.mock import patch
+
+try:
+    from tests.disk_waits import DISK_BOUND_SECONDS
+    from tests.test_image_delete_routes import ImageDeleteTestCase
+except ImportError:  # discovered with tests/ as the top-level directory
+    from disk_waits import DISK_BOUND_SECONDS
+    from test_image_delete_routes import ImageDeleteTestCase
+
+from py.api import images as images_module
+from py.api.admin import _collect_media_files
+from py.api.autotag_routes import AutotagRoutesMixin
+from utils.image_monitor import ImageGenerationHandler
+
+MAINTENANCE_PREVIEW = "/prompt_manager/maintenance/preview"
+MAINTENANCE_APPLY = "/prompt_manager/maintenance/apply"
+
+
+class _GapCase(ImageDeleteTestCase):
+    def marks(self):
+        conn = sqlite3.connect(self.api.db.model.db_path)
+        try:
+            return [
+                row[0] for row in conn.execute("SELECT locator FROM image_removals")
+            ]
+        finally:
+            conn.close()
+
+    def revision(self):
+        conn = sqlite3.connect(self.api.db.model.db_path)
+        try:
+            return conn.execute(
+                "SELECT revision FROM image_removals_meta WHERE singleton_id=1"
+            ).fetchone()[0]
+        finally:
+            conn.close()
+
+
+class TestThumbnailJobsSkipHiddenFiles(_GapCase):
+    async def test_missing_and_rebuild_plans_leave_a_removed_file_out(self):
+        self.write("hidden.png", b"hidden bytes", mtime=1_000_000)
+        self.write("shown.png", b"shown bytes", mtime=2_000_000)
+        await self.delete(["hidden.png"], "library")
+        thumbs = self.output_dir / "thumbnails"
+        for mode in ("missing", "rebuild"):
+            with self.subTest(mode=mode):
+                plan = self.api._plan_thumbnails(self.output_dir, thumbs, mode=mode)
+                sources = [src.name for src, _dst, _video in plan["targets"]]
+                self.assertEqual(sources, ["shown.png"])
+                self.assertEqual(plan["missing"], 1)
+
+    async def test_a_changed_file_gets_its_preview_again(self):
+        self.write("hidden.png", b"hidden bytes", mtime=1_000_000)
+        await self.delete(["hidden.png"], "library")
+        self.write("hidden.png", b"new and longer bytes", mtime=3_000_000)
+        plan = self.api._plan_thumbnails(
+            self.output_dir, self.output_dir / "thumbnails"
+        )
+        self.assertEqual([t[0].name for t in plan["targets"]], ["hidden.png"])
+
+
+class TestLegacyAutotagListing(_GapCase):
+    async def test_scan_output_dir_leaves_a_removed_file_out(self):
+        self.write("hidden.png", b"hidden bytes", mtime=1_000_000)
+        self.write("shown.png", b"shown bytes", mtime=2_000_000)
+        await self.delete(["hidden.png"], "library")
+        with patch.object(
+            self.api, "_find_comfyui_output_dir", return_value=str(self.output_dir)
+        ):
+            response = await self.client.get("/prompt_manager/scan_output_dir")
+            body = await response.json()
+        self.assertEqual([image["filename"] for image in body["images"]], ["shown.png"])
+
+
+class TestMissingFileLinksPrunesMarks(_GapCase):
+    async def links_review(self, status=200):
+        return await self.post(
+            MAINTENANCE_PREVIEW, {"operation": "missing_links", "options": {}}, status
+        )
+
+    async def apply(self, token, status=202):
+        body = await self.post(MAINTENANCE_APPLY, {"token": token}, status)
+        if status != 202:
+            return body
+        job = self.api._jobs.get(body["job"]["id"])
+        await asyncio.wait_for(asyncio.shield(job.task), DISK_BOUND_SECONDS)
+        return job
+
+    async def removed(self, name):
+        self.write("other.png", b"another output")  # the source lists something
+        self.write(name, f"bytes of {name}".encode(), mtime=1_000_000)
+        await self.delete([name], "library")
+        self.assertEqual(len(self.marks()), 1)
+
+    async def test_the_mark_of_a_gone_file_is_counted_and_forgotten(self):
+        await self.removed("gone.png")
+        # Not an unlink by PromptManager: the user deleted it elsewhere.
+        os.remove(self.output_dir / "gone.png")
+        plan = await self.links_review()
+        self.assertEqual(plan["scope"]["count"], 1)
+        self.assertEqual(
+            plan["scope"]["label"], "1 removed image whose file is gone or changed"
+        )
+        before = self.revision()
+        job = await self.apply(plan["token"])
+        self.assertEqual(job.state, "finished", job.summary())
+        self.assertEqual(self.marks(), [])
+        self.assertGreater(self.revision(), before)
+
+    async def test_the_mark_of_a_present_file_stays(self):
+        await self.removed("kept.png")
+        plan = await self.links_review()
+        self.assertEqual(plan["scope"]["count"], 0)
+        self.assertEqual(plan["scope"]["label"], "No links to missing files")
+        self.assertEqual(len(self.marks()), 1)
+
+    async def test_a_mark_under_an_unreadable_source_stays(self):
+        await self.removed("away.png")
+        os.remove(self.output_dir / "away.png")
+        with patch(
+            "database.maintenance.readable_roots",
+            side_effect=lambda r: [False] * len(r),
+        ):
+            plan = await self.links_review()
+            self.assertEqual(plan["scope"]["count"], 0)
+        self.assertEqual(len(self.marks()), 1)
+
+    async def test_the_mark_of_a_changed_file_is_forgotten(self):
+        await self.removed("changed.png")
+        self.write("changed.png", b"other and longer bytes", mtime=4_000_000)
+        plan = await self.links_review()
+        self.assertEqual(plan["scope"]["count"], 1)
+        job = await self.apply(plan["token"])
+        self.assertEqual(job.state, "finished", job.summary())
+        self.assertEqual(self.marks(), [])
+
+    async def test_marks_under_an_empty_source_stay(self):
+        # An unplugged drive leaves its mount point: an empty, readable folder.
+        await self.removed("mounted.png")
+        for child in list(self.output_dir.iterdir()):
+            if child.is_dir():
+                shutil.rmtree(child)
+            else:
+                os.remove(child)
+        plan = await self.links_review()
+        self.assertEqual(plan["scope"]["count"], 0)
+        self.assertEqual(len(self.marks()), 1)
+
+    async def test_a_name_the_folder_still_lists_is_not_gone(self):
+        # A case-sensitive NTFS folder: lstat of the normcased locator fails.
+        await self.removed("Listed.png")
+        real_lstat = os.lstat
+        locator = self.marks()[0]
+
+        def lstat(path, *args, **kwargs):
+            if os.fspath(path) == locator:
+                raise FileNotFoundError(path)
+            return real_lstat(path, *args, **kwargs)
+
+        with (
+            patch("database.maintenance.os.lstat", side_effect=lstat),
+            patch("database.maintenance.os.path.normcase", side_effect=str.lower),
+        ):
+            from database.maintenance import _stale_marks
+
+            stale = _stale_marks([(locator, 1, 1)], [str(self.output_dir)])
+        self.assertEqual(stale, [])
+
+    async def test_a_file_back_before_the_apply_changes_the_plan(self):
+        await self.removed("back.png")
+        path = self.output_dir / "back.png"
+        saved = Path(self.tmpdir) / "saved.png"
+        shutil.copy2(path, saved)
+        os.remove(path)
+        plan = await self.links_review()
+        shutil.copy2(saved, path)
+        job = await self.apply(plan["token"])
+        self.assertEqual(job.state, "failed", job.summary())
+        self.assertEqual(len(self.marks()), 1)
+
+
+class TestWalkersSkipTrashFolders(_GapCase):
+    def seed(self):
+        self.write("kept.png", b"kept")
+        self.write(
+            "x.png", b"trashed x", folder=self.output_dir / ".Trash-1000" / "files"
+        )
+        self.write(
+            "y.png", b"trashed y", folder=self.output_dir / "sub" / "$RECYCLE.BIN"
+        )
+        self.write("z.png", b"trashed z", folder=self.output_dir / ".Trashes")
+
+    def test_walk_media_and_the_settled_entry(self):
+        self.seed()
+        walked = sorted(
+            Path(entry.path).name
+            for entry in images_module._walk_media(self.output_dir, 100)
+        )
+        self.assertEqual(walked, ["kept.png"])
+        for relative in (
+            os.path.join(".Trash-1000", "files", "x.png"),
+            os.path.join("sub", "$RECYCLE.BIN", "y.png"),
+        ):
+            with self.subTest(relative=relative):
+                self.assertIsNone(
+                    images_module._settled_entry(self.output_dir, relative)
+                )
+
+    def test_scan_duplicate_and_autotag_walkers(self):
+        self.seed()
+        collected = [
+            Path(p).name for p in _collect_media_files([self.output_dir], [".png"])
+        ]
+        self.assertEqual(collected, ["kept.png"])
+        found, _truncated = AutotagRoutesMixin._collect_scan_files(self.output_dir)
+        self.assertEqual([p.name for p in found], ["kept.png"])
+
+    def test_the_image_monitor(self):
+        monitor = ImageGenerationHandler.__new__(ImageGenerationHandler)
+        monitor.supported_extensions = (".png",)
+        base = str(self.output_dir)
+        self.assertTrue(monitor.is_image_file(os.path.join(base, "a.png")))
+        for folder in (".Trash-1000", "$Recycle.Bin", ".Trashes", ".trash"):
+            with self.subTest(folder=folder):
+                self.assertFalse(
+                    monitor.is_image_file(os.path.join(base, folder, "files", "a.png"))
+                )
+
+    async def test_the_workspace_listing(self):
+        self.seed()
+        self.assertEqual(await self.listed(), ["kept.png"])
+
+    def test_a_root_named_like_a_trash_folder_still_lists(self):
+        root = Path(self.tmpdir) / ".Trash-1000"
+        (root / "inner").mkdir(parents=True)
+        (root / "inner" / "a.png").write_bytes(b"a")
+        walked = [Path(e.path).name for e in images_module._walk_media(root, 10)]
+        self.assertEqual(walked, ["a.png"])
diff --git a/tests/test_media_folders.py b/tests/test_media_folders.py
new file mode 100644
index 00000000..d866aa93
--- /dev/null
+++ b/tests/test_media_folders.py
@@ -0,0 +1,41 @@
+"""The shared folder skip every media walker uses (#501)."""
+
+import unittest
+
+from utils.media_folders import in_skipped_folder, is_skipped_folder, is_trash_folder
+
+
+class TrashFolderNameTests(unittest.TestCase):
+    def test_trash_folder_names_in_any_case(self):
+        for name in (
+            ".Trash",
+            ".trash",
+            ".Trash-1000",
+            ".trash-0",
+            ".Trashes",
+            "$RECYCLE.BIN",
+            "$Recycle.Bin",
+        ):
+            with self.subTest(name=name):
+                self.assertTrue(is_trash_folder(name))
+                self.assertTrue(is_skipped_folder(name))
+
+    def test_other_names_are_not_trash(self):
+        for name in (".Trash-abc", ".Trash-", "trash", "Recycle", "RECYCLE.BIN", "x"):
+            with self.subTest(name=name):
+                self.assertFalse(is_trash_folder(name))
+
+    def test_thumbnails_is_skipped_exactly(self):
+        self.assertTrue(is_skipped_folder("thumbnails"))
+        self.assertFalse(is_skipped_folder("Thumbnails"))
+        self.assertFalse(is_trash_folder("thumbnails"))
+
+    def test_folder_parts(self):
+        self.assertTrue(in_skipped_folder(("a", ".Trash-1000", "files")))
+        self.assertTrue(in_skipped_folder(("thumbnails",)))
+        self.assertFalse(in_skipped_folder(("a", "b")))
+        self.assertFalse(in_skipped_folder(()))
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests/test_trash_freedesktop.py b/tests/test_trash_freedesktop.py
index a0f3f3ec..dd8ab6d5 100644
--- a/tests/test_trash_freedesktop.py
+++ b/tests/test_trash_freedesktop.py
@@ -285,6 +285,20 @@ class UnavailableBackendTests(unittest.TestCase):
                 )
         self.assertIsInstance(trash.default_backend("linux"), trash.FreedesktopBackend)
 
+    def test_windows_says_why_the_recycle_bin_is_not_used(self):
+        # #500 ruling (a): the Recycle Bin can delete permanently, so it is never used.
+        backend = trash.default_backend("win32")
+        self.assertEqual(
+            backend.available("C:/x/a.png", ["C:/x"]), trash.WINDOWS_RECYCLE
+        )
+        with self.assertRaises(trash.TrashRefused) as caught:
+            backend.move("C:/x/a.png", ["C:/x"])
+        self.assertEqual(caught.exception.code, trash.WINDOWS_RECYCLE)
+        self.assertIn(trash.WINDOWS_RECYCLE, trash.CODES)
+        self.assertEqual(
+            trash.default_backend("darwin").available("/x", ["/"]), trash.NO_TRASH
+        )
+
     def test_module_functions_use_the_replaceable_backend(self):
         fake = mock.Mock()
         fake.available.return_value = "network"
diff --git a/utils/image_monitor.py b/utils/image_monitor.py
index d2fc1c19..18c62c1f 100644
--- a/utils/image_monitor.py
+++ b/utils/image_monitor.py
@@ -45,6 +45,7 @@ from .foreign_runs import (
     store_foreign,
 )
 from .hashing import generate_prompt_hash
+from .media_folders import in_skipped_folder
 from .metadata_extractor import ComfyUIMetadataExtractor
 from .parallel import note_refusal, refused_at_shutdown
 from .logging_config import get_logger
@@ -428,11 +429,12 @@ class ImageGenerationHandler(FileSystemEventHandler):
 
         Returns:
             True if the file has a supported image extension and is not
-            inside a thumbnails directory, False otherwise
+            inside a thumbnails or trash folder, False otherwise
         """
-        # Skip files in thumbnails directory - those are derivatives, not
-        # generated images
-        if "/thumbnails/" in filepath or "\\thumbnails\\" in filepath:
+        # Previews are derivatives, and a trash folder inside an output
+        # folder holds thrown-away files (#501): neither is a new output.
+        folders = filepath.replace("\\", "/").split("/")[:-1]
+        if in_skipped_folder(folders):
             return False
         return filepath.lower().endswith(self.supported_extensions)
 
diff --git a/utils/media_folders.py b/utils/media_folders.py
new file mode 100644
index 00000000..6052a001
--- /dev/null
+++ b/utils/media_folders.py
@@ -0,0 +1,31 @@
+"""The folders no media walker lists below an output root (#501).
+
+``thumbnails`` holds PromptManager's own previews. A trash folder that sits
+inside an output root (``D:\\$RECYCLE.BIN`` when ``D:\\`` is a root,
+``.Trash-1000`` at a mount point, ``.Trashes`` on a macOS volume) holds files
+the user threw away: they are not outputs, so they are never listed, scanned,
+linked, tagged or previewed. Every walker asks these predicates instead of
+testing names itself; only folders *below* a root are tested.
+"""
+
+TRASH_FOLDER_NAMES = frozenset({".trash", ".trashes", "$recycle.bin"})
+PREVIEW_FOLDER = "thumbnails"
+
+
+def is_trash_folder(name: str) -> bool:
+    """A trash folder name, in any case: ``.Trash``, ``.Trash-<uid>``,
+    ``.Trashes`` or ``$RECYCLE.BIN``."""
+    folded = name.casefold()
+    if folded in TRASH_FOLDER_NAMES:
+        return True
+    return folded.startswith(".trash-") and folded[7:].isdigit()
+
+
+def is_skipped_folder(name: str) -> bool:
+    """A folder no walker descends: ``thumbnails`` (exact) or a trash folder."""
+    return name == PREVIEW_FOLDER or is_trash_folder(name)
+
+
+def in_skipped_folder(parts) -> bool:
+    """Whether any of the folder names *parts* is skipped."""
+    return any(is_skipped_folder(part) for part in parts)
diff --git a/utils/trash.py b/utils/trash.py
index 242236b5..729f6d66 100644
--- a/utils/trash.py
+++ b/utils/trash.py
@@ -14,8 +14,9 @@ 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.
+Windows never uses the Recycle Bin (#500: files over its capacity are deleted
+permanently, undetectably), so it answers ``windows-recycle``; macOS has no
+backend yet (#499). Both use :class:`UnavailableBackend`.
 
 ``BACKEND`` is the one replaceable seam; tests install a fake.
 """
@@ -36,7 +37,11 @@ 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})
+# Windows never uses the Recycle Bin (#500): it can delete permanently.
+WINDOWS_RECYCLE = "windows-recycle"
+CODES = frozenset(
+    {NO_TRASH, NETWORK, TRASH_IN_ROOT, PATH_TOO_LONG, LOCKED, FAILED, WINDOWS_RECYCLE}
+)
 
 HOME_TRASH_LABEL = "home trash"
 RENAME_NOREPLACE = 1
@@ -177,13 +182,16 @@ def _owned_folder(path: str, uid: int, *, create: bool) -> bool:
 
 
 class UnavailableBackend:
-    """No trash on this system: every move is refused, nothing is touched."""
+    """No trash on this system: every move is refused with *code*, untouched."""
+
+    def __init__(self, code: str = NO_TRASH):
+        self.code = code
 
     def available(self, path: str, roots) -> str | None:
-        return NO_TRASH
+        return self.code
 
     def move(self, path: str, roots) -> str:
-        raise TrashRefused(NO_TRASH)
+        raise TrashRefused(self.code)
 
 
 class FreedesktopBackend:
@@ -346,7 +354,12 @@ class FreedesktopBackend:
 
 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"):
+    if platform == "win32" or platform.startswith("cygwin"):
+        # #500, ruling (a): Windows deletes files larger than the Recycle Bin's
+        # capacity permanently, and IFileOperation's PreDeleteItem guard cannot
+        # tell (native probe runs 40951037575, 40953347904).
+        return UnavailableBackend(WINDOWS_RECYCLE)
+    if platform == "darwin":
         return UnavailableBackend()
     return FreedesktopBackend()
 
diff --git a/web/js/workspace/image-tagging-state.mjs b/web/js/workspace/image-tagging-state.mjs
index 4cf050c2..20d0c6d8 100644
--- a/web/js/workspace/image-tagging-state.mjs
+++ b/web/js/workspace/image-tagging-state.mjs
@@ -317,6 +317,9 @@ export function createImageTaggingController({api, target, source = 'images', on
     // A save in flight: closing hands the capture release to its completion.
     let saving = false;
     let observedOnce = false;
+    // #515: the user typed since the last observation started (or before the first one answered), so the answer
+    // keeps that text.
+    let typed = false;
 
     function set(patch) {
         if (closed) return;
@@ -345,7 +348,8 @@ export function createImageTaggingController({api, target, source = 'images', on
     async function open({keepDraft = false} = {}) {
         if (closed) return;
         const serial = ++openSerial;
-        const draft = keepDraft ? state.manual : null;
+        // A reopen after an observation starts from its answer; text typed before any observation is kept (#515).
+        if (observedOnce) typed = false;
         const seenRevision = state.tagRevision;
         stopGenerating();
         releaseCapture();
@@ -385,15 +389,19 @@ export function createImageTaggingController({api, target, source = 'images', on
         const savedModel = !chosen && !modelPicked && read.model ? {model: read.model} : {};
         observedOnce = true;
         const {width, height, media_type: mediaType, preview_url: previewUrl, generate} = value.capture;
-        // A kept draft stays as typed unless another writer changed the set meanwhile: then the set leads it (I2).
-        const changed = draft !== null && value.tag_revision !== seenRevision;
+        // The draft is read now, so text typed while observing is kept (#515). A kept draft stays as typed unless
+        // another writer changed the set meanwhile: then the set leads it (I2). Text typed before the user saw any
+        // saved tags follows them, so a Save never drops tags it never showed.
+        const draft = keepDraft || typed ? state.manual : null;
+        const changed = keepDraft && value.tag_revision !== seenRevision;
+        const unseen = draft !== null && !keepDraft;
         const reason = generate.supported ? '' : (generate.reason || '');
         set({
             phase: 'ready',
             image: {filename: value.filename, title: value.title, width, height, mediaType, previewUrl,
                 dimensions: width && height ? `${width} × ${height}` : null},
             savedTags: value.tags.slice(), tagRevision: value.tag_revision,
-            manual: draft === null ? value.tags.join(', ') : changed ? mergeCurrent(value.tags, draft) : draft,
+            manual: draft === null ? value.tags.join(', ') : changed || unseen ? mergeCurrent(value.tags, draft) : draft,
             draftRevision: state.draftRevision + 1,
             ...savedModel, options: {...defaults, ...chosen?.options}, optionsRevision: state.optionsRevision + 1,
             generateSupported: generate.supported, generateReason: generate.reason,
@@ -402,6 +410,7 @@ export function createImageTaggingController({api, target, source = 'images', on
     }
 
     function setManual(text) {
+        typed = true;
         set({manual: typeof text === 'string' ? text : ''});
     }
 
diff --git a/web/js/workspace/images-delete.mjs b/web/js/workspace/images-delete.mjs
index 7f53015e..28b92219 100644
--- a/web/js/workspace/images-delete.mjs
+++ b/web/js/workspace/images-delete.mjs
@@ -32,7 +32,7 @@ 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.';
+export const NO_TRASH_HERE = 'Moving to the trash is not available for these files; each one lists why.';
 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.',
diff --git a/web/js/workspace/index.mjs b/web/js/workspace/index.mjs
index 389470ab..f75483fb 100644
--- a/web/js/workspace/index.mjs
+++ b/web/js/workspace/index.mjs
@@ -1,7 +1,8 @@
 import '../api-client.js';
 import { createButton, createScope } from '../shared/toolkit/core.js';
 import { routes, mountRouter, summarizeStats, routeTitle, loadNavCounts, navigateTo, COLLECTION_ROUTE,
-    LIBRARY_WRITE_COUNTS, NAV_COUNT_ROUTES, FAVORITE_WRITE_COUNTS, FAVORITE_COUNT_DELAY_MS, REVIEW_WRITE_COUNTS } from './state.mjs';
+    LIBRARY_WRITE_COUNTS, NAV_COUNT_ROUTES, FAVORITE_WRITE_COUNTS, FAVORITE_COUNT_DELAY_MS, REVIEW_WRITE_COUNTS,
+    JOB_END_COUNTS } from './state.mjs';
 import { mountWorkspaceTheme } from './theme.mjs';
 import { forgetImagesVisit, mountImages, onFavoriteChange } from './images.mjs';
 import { forgetLibraryVisit, mountLibrary, mountNewPromptAction, onLibraryWrite } from './library.mjs';
@@ -196,16 +197,20 @@ function mountWorkspace() {
         scope.cleanup(() => window.removeEventListener(type, revive, true));
     }
     scope.listen(document, 'visibilitychange', () => { if (document.visibilityState === 'visible') revive(); });
-    // A metadata scan that ends may have added items to Needs review: re-read its count once per ended scan.
-    const runningScans = new Set();
+    // A job that ends re-reads the counts it can change, once per ended job: a metadata scan may have added items to
+    // Needs review; deleting images (#480) hides or trashes them, with their favorites, review items and collection
+    // memberships (#501).
+    const runningJobs = new Map();
     scope.cleanup(activity.subscribe(() => {
-        let ended = false;
+        const ended = new Set();
         for (const job of activity.jobs()) {
-            if (job.kind !== 'scan') continue;
-            if (job.state === 'running') runningScans.add(job.id);
-            else if (runningScans.delete(job.id)) ended = true;
+            if (!JOB_END_COUNTS[job.kind]) continue;
+            if (job.state === 'running') runningJobs.set(job.id, job.kind);
+            else if (runningJobs.delete(job.id)) ended.add(job.kind);
         }
-        if (ended) refreshCounts(REVIEW_WRITE_COUNTS);
+        if (!ended.size) return;
+        refreshCounts([...new Set([...ended].flatMap(kind => JOB_END_COUNTS[kind]))]);
+        if (ended.has('image_delete') && !leaving) collections.load();
     }));
     refreshCounts(NAV_COUNT_ROUTES);
     // S9: a finished restore replaced the library. The cached visits are dropped, the counts and saved views read
diff --git a/web/js/workspace/state.mjs b/web/js/workspace/state.mjs
index 9af7739c..a2b056c9 100644
--- a/web/js/workspace/state.mjs
+++ b/web/js/workspace/state.mjs
@@ -129,6 +129,11 @@ export const LIBRARY_WRITE_COUNTS = Object.freeze(['library', 'tags']);
 export const FAVORITE_WRITE_COUNTS = Object.freeze(['favorites']);
 /** The count a review decision or a finished metadata scan can change. */
 export const REVIEW_WRITE_COUNTS = Object.freeze(['review']);
+/** Job kind -> the counts re-read once when such a job ends (a scan feeds Needs review; #501: a delete hides images). */
+export const JOB_END_COUNTS = Object.freeze({
+    scan: REVIEW_WRITE_COUNTS,
+    image_delete: Object.freeze(['images', 'favorites', 'review']),
+});
 // Trailing delay that coalesces quick star toggles into one Favorites count read (#249 M1).
 export const FAVORITE_COUNT_DELAY_MS = 300;