Merge commit '704db4be' into feat/481-delete-prompt-page

.github/workflows/portability.yml

@@ -26,6 +26,9 @@ env:
 # Measured 2026-10-29 on main runs 40935337099 (dd247e3), 40938407252 (dd247e3), 40941240624 (dd247e3):
 #   slowest leg: 7m24s (py3.12 windows) / 7m31s (py3.13 windows) / 7m32s (py3.14 windows) -> 15 min
 #     (windows 4m56s-7m32s, macos 4m39s-6m14s)
+#   native-trash (#499) has no green main runs yet: its probe, run 40949370943 (2026-10-29), took
+#     about 1 minute per macOS leg, so ceil(1.5 x 2) + 3 gives it 6 minutes (step 4). Recalibrate
+#     after three green runs.
 jobs:
   test:
     name: py${{ matrix.python-version }} on ${{ matrix.os }}
@@ -76,3 +79,28 @@ jobs:
         env:
           PM_TEST_SLOWEST: "40"
         run: python -m tests.run_suite -v
+
+  # The real Trash on both macOS CPUs (#499 design §10.2). It runs apart from the suite, so a red
+  # suite never skips the native proof.
+  native-trash:
+    name: native trash on ${{ matrix.os }}
+    runs-on: ${{ matrix.os }}
+    timeout-minutes: 6
+    strategy:
+      fail-fast: false
+      matrix:
+        os: [macos-latest, macos-15-intel]
+
+    steps:
+      - uses: actions/checkout@v7
+
+      - name: Set up Python
+        uses: actions/setup-python@v7
+        with:
+          python-version: "3.12"
+
+      - name: Native trash proof
+        timeout-minutes: 4
+        env:
+          PM_NATIVE_TRASH: macos
+        run: python -m unittest tests.test_trash_macos_native tests.test_trash_macos -v

TESTING.md

@@ -53,6 +53,14 @@ stdin open and the server (`--exit-on-stdin-eof`) shuts down cleanly, removing i
 temp root, when that pipe closes, so a worker killed without teardown leaves no
 server behind. A test that fails gets the stderr its server wrote during the test
 attached to the report as `server-stderr` (an aiohttp traceback behind a 500).
+Each worker also records its browser's own log (#513, `tools/vendor/browser-logs.mjs`):
+Playwright's `pw:browser` channel, kept as a 300-line tail without printing it
+unless `DEBUG` asks for it. A browser that exits abnormally writes one
+`pm-browser-exit| browser pid=<n> exited: exitCode=<c>, signal=<s>` line to the job
+log, and a test that fails on a closed or crashed target ("Target page, context or
+browser has been closed") gets the tail attached as `browser-stderr.txt`, which the
+uploaded test-results carry. `tests/browser/browser-logs.spec.mjs` kills a browser of
+its own on purpose, so each browser job logs that line once for it.
 A `/__test__/.../reset` still running after 2 s writes a report to the job log,
 passing test or not: every thread's stack taken while it waits, the default
 executor's threads and queue, the open connections and WAL size, then the total
@@ -2460,14 +2468,57 @@ 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
-  test touches a real trash.
+  a trash inside an output root the move is refused (`no-trash`, `trash-in-root`). macOS
+  (#499, `utils/trash_macos.py`) calls `-[NSFileManager trashItemAtURL:resultingItemURL:error:]`
+  through `ctypes` (ten fixed `objc_msgSend` prototypes, nil checks before every send, one
+  autorelease pool per call, loaded on first use). It answers in the review dialog, without
+  writing: `network` (not a local volume), `no-trash` (read-only, or no trash folder
+  macOS would use), `cloud-sync` (iCloud Drive, `~/Library/CloudStorage`, or ubiquitous),
+  `trash-in-root` (the trash folder is inside an output root, compared by `(st_dev,
+  st_ino)` of every ancestor) and `no-permission` (folder not writable, an immutable or
+  append flag, a foreign file in a sticky folder). A fresh volume's `.Trashes/<uid>` is
+  predicted (`URLForDirectory` answers 3328/`ENOENT` until macOS makes it). A refused move
+  keeps the file and logs the `NSError`; a move macOS reports done but whose place cannot be
+  shown (no result URL, or one inside an output root) goes under `errors` as
+  `trash-unconfirmed`, keeps its records and stops the job. Windows
+  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."). Library-only removal works everywhere. Tests use `tests/fake_trash.py`; the
+  browser test server installs it, so no test touches a real trash. The native proof,
+  `tests/test_trash_macos_native.py`, runs only with `PM_NATIVE_TRASH=macos` in
+  portability.yml's `native-trash` job (`macos-latest` and `macos-15-intel`): the startup
+  volume, a name clash, fresh APFS and FAT32 disk images, an image used as an output root,
+  a read-only image and an unwritable folder (design and probe:
+  `docs/egm/499-macos-trash/`, local).
 
 ### Prompt library cards and recorded deviations (#249 PR E)
 

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

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]],
         }

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:

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 = []

py/api/image_delete.py

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

py/api/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)

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)}")
 

tests/browser/browser-logs.spec.mjs

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

tests/browser/fixtures.mjs

@@ -4,13 +4,29 @@
 // points at that worker's server. A server that dies fails the running test
 // with its stderr tail, and the next test in the worker gets a fresh server.
 // A test that fails gets the stderr its server wrote during it attached as
-// `server-stderr`.
+// `server-stderr`. The worker also records its browser's own log (#513): a
+// browser that exits abnormally prints one `pm-browser-exit|` line, and a test
+// that fails on a closed or crashed target gets that log attached as
+// `browser-stderr`.
 import { test as base, expect } from '../../tools/vendor/node_modules/@playwright/test/index.mjs';
 import { attachServerStderr, createServerSupervisor, startBrowserServer } from '../../tools/vendor/start-browser-server.mjs';
+import { attachBrowserLogs, createBrowserLogRecorder } from '../../tools/vendor/browser-logs.mjs';
 
 const SERVER_FIXTURE_TIMEOUT_MS = 90_000;
 
 export const test = base.extend({
+    browserLogs: [async ({}, use) => {
+        const recorder = createBrowserLogRecorder();
+        try {
+            await use(recorder);
+        } finally {
+            recorder.uninstall();
+        }
+    }, { scope: 'worker', auto: true }],
+    browserLogsOnFailure: [async ({ browserLogs }, use, testInfo) => {
+        await use();
+        await attachBrowserLogs(testInfo, browserLogs);
+    }, { auto: true }],
     testServer: [async ({}, use) => {
         const supervisor = createServerSupervisor({ start: () => startBrowserServer() });
         try {

tests/browser/malloc_tuning.py

@@ -0,0 +1,31 @@
+"""Keep the browser test server's large frees going back to the OS (#516).
+
+glibc serves an allocation of at least the mmap threshold (128 KiB at start) with its
+own mmap and unmaps it on free. Freeing such a block also raises the threshold, up to
+32 MiB, so the server's later image-sized buffers (PIL pixel data, PNG encoding,
+whole-file reads in the executor threads) come from per-thread malloc arenas instead.
+Those arenas fragment and keep their pages: a server grew from 13 MB to 0.6-1 GB RSS
+over one page-workflow shard while its Python heap stayed near 12 MB. Setting
+M_MMAP_THRESHOLD fixes the threshold at its start value and turns the raising off.
+"""
+
+import ctypes
+import sys
+
+M_MMAP_THRESHOLD = -3
+MMAP_THRESHOLD_BYTES = 128 * 1024
+
+
+def pin_mmap_threshold(libc=None) -> bool:
+    """Fix glibc's mmap threshold; False where there is no glibc ``mallopt``."""
+    if not sys.platform.startswith("linux"):
+        return False
+    if libc is None:
+        try:
+            libc = ctypes.CDLL(None)
+        except OSError:
+            return False
+    mallopt = getattr(libc, "mallopt", None)
+    if mallopt is None:
+        return False
+    return bool(mallopt(M_MMAP_THRESHOLD, MMAP_THRESHOLD_BYTES))

tests/browser/server.py

@@ -20,6 +20,7 @@ from PIL import Image
 from PIL.PngImagePlugin import PngInfo
 
 from tests.browser.folder_trash import close_trash, replace_folder
+from tests.browser.malloc_tuning import pin_mmap_threshold
 from tests.browser.request_hold import register_request_holds
 from tests.browser import slow_reset_watch
 from tests.browser.fast_reset import SEED_ON_LIVE_ENV, ScratchSeeder
@@ -755,6 +756,8 @@ def main(argv: list[str] | None = None) -> None:
         help="shut down when stdin closes (the launcher's parent-death guard)",
     )
     args = parser.parse_args(argv)
+    # Before any thread allocates (#516): large frees then return to the OS.
+    pin_mmap_threshold()
     listener = bind_listener(args.port)
     # One machine-readable line tells the launcher which port this process holds.
     print(f"{PORT_ANNOUNCEMENT}{listener.getsockname()[1]}", flush=True)

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

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

@@ -643,7 +643,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 variant…','Delete prompt…','Cancel','Save']);
     for(const name of ['Create text revision','Archive variant…','Delete prompt…'])await expect(button(dialog,name)).toBeDisabled();

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

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');

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, '');
+});

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]);
 });
 

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

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"])
 

tests/test_browser_malloc_tuning.py

@@ -0,0 +1,92 @@
+"""#516: the browser test server pins glibc's mmap threshold.
+
+Large frees then go back to the OS instead of fragmenting per-thread arenas.
+"""
+
+import ctypes
+import sys
+import unittest
+from unittest import mock
+
+from tests.browser import malloc_tuning
+
+
+class FakeLibc:
+    def __init__(self, result=1):
+        self.calls = []
+        self.result = result
+
+    def mallopt(self, option, value):
+        self.calls.append((option, value))
+        return self.result
+
+
+class PinMmapThresholdTests(unittest.TestCase):
+    def test_sets_a_fixed_threshold_through_mallopt(self):
+        libc = FakeLibc()
+        with mock.patch.object(sys, "platform", "linux"):
+            self.assertTrue(malloc_tuning.pin_mmap_threshold(libc))
+        # M_MMAP_THRESHOLD is -3; setting it also turns off glibc's dynamic
+        # raising of the threshold.
+        self.assertEqual(libc.calls, [(-3, 128 * 1024)])
+
+    def test_reports_a_refused_setting(self):
+        with mock.patch.object(sys, "platform", "linux"):
+            self.assertFalse(malloc_tuning.pin_mmap_threshold(FakeLibc(result=0)))
+
+    def test_skips_a_libc_without_mallopt(self):
+        with mock.patch.object(sys, "platform", "linux"):
+            self.assertFalse(malloc_tuning.pin_mmap_threshold(object()))
+
+    def test_skips_other_platforms_without_loading_a_libc(self):
+        for name in ("win32", "darwin"):
+            with (
+                self.subTest(name),
+                mock.patch.object(sys, "platform", name),
+                mock.patch.object(
+                    ctypes, "CDLL", side_effect=AssertionError("no libc load")
+                ),
+            ):
+                self.assertFalse(malloc_tuning.pin_mmap_threshold())
+
+    def test_skips_when_the_process_libc_cannot_be_loaded(self):
+        with (
+            mock.patch.object(sys, "platform", "linux"),
+            mock.patch.object(ctypes, "CDLL", side_effect=OSError("no libc")),
+        ):
+            self.assertFalse(malloc_tuning.pin_mmap_threshold())
+
+    @unittest.skipUnless(
+        sys.platform.startswith("linux")
+        and hasattr(ctypes.CDLL(None), "gnu_get_libc_version"),
+        "glibc only",
+    )
+    def test_glibc_accepts_the_setting(self):
+        # This pins the threshold for the rest of this test process too; the
+        # test server does the same at start, so nothing here depends on it.
+        self.assertTrue(malloc_tuning.pin_mmap_threshold())
+
+
+class ServerMainTests(unittest.TestCase):
+    def test_main_pins_the_threshold_before_it_starts_serving(self):
+        from tests.browser import server
+
+        order = []
+
+        def bind(port):
+            order.append("bind")
+            raise RuntimeError("stop before serving")
+
+        with (
+            mock.patch.object(
+                server, "pin_mmap_threshold", side_effect=lambda: order.append("pin")
+            ),
+            mock.patch.object(server, "bind_listener", side_effect=bind),
+        ):
+            with self.assertRaises(RuntimeError):
+                server.main(["--port", "0"])
+        self.assertEqual(order, ["pin", "bind"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_bulk_library_delete.py

@@ -66,14 +66,15 @@ 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)

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
 
@@ -313,18 +320,86 @@ class TestFileCheck(ImageDeleteTestCase):
             self.assertIn(code, image_delete.REASONS)
 
 
+class TestMacOSOutcomes(ImageDeleteTestCase):
+    """The codes and the unconfirmed outcome the macOS backend adds (#499)."""
+
+    async def test_permission_and_cloud_refusals_show_their_sentences(self):
+        self.write("a.png", b"aaa")
+        self.write("b.png", b"bbb")
+        self.fake.refuse[str(self.output_dir / "a.png")] = "no-permission"
+        self.fake.refuse[str(self.output_dir / "b.png")] = "cloud-sync"
+        plan = await self.review(["a.png", "b.png"])
+        self.assertEqual(plan["trash"], {"count": 0})
+        self.assertEqual(
+            plan["trash_refused"]["counts"], {"no-permission": 1, "cloud-sync": 1}
+        )
+        details = {row["detail"] for row in plan["trash_refused"]["rows"]}
+        self.assertEqual(
+            details,
+            {image_delete.REASONS["no-permission"], image_delete.REASONS["cloud-sync"]},
+        )
+
+    async def test_an_unconfirmed_move_keeps_records_and_stops_the_job(self):
+        from utils import trash
+
+        self.write("a.png", b"aaa", mtime=1_000_000)
+        self.write("b.png", b"bbb", mtime=2_000_000)
+        self.link("a.png")
+        original = self.fake.move
+
+        def move(path, roots):
+            if path.endswith("a.png"):
+                original(path, roots)  # the file did leave its folder
+                raise trash.TrashUnconfirmed("resultingItemURL=None")
+            return original(path, roots)
+
+        self.fake.move = move
+        _plan, job = await self.delete(["a.png", "b.png"], "trash")
+        self.assertEqual(job.state, "failed")
+        self.assertEqual(job.scope.get("failure"), "records-failed")
+        self.assertEqual(
+            self.details(job, "errors"), [image_delete.REASONS["trash-unconfirmed"]]
+        )
+        self.assertEqual(self.count("SELECT COUNT(*) FROM generated_images"), 1)
+        self.assertEqual(
+            self.details(job, "not_changed"), [image_delete.REASONS["not-attempted"]]
+        )
+        self.assertTrue((self.output_dir / "b.png").exists())
+
+    async def test_a_refusal_detail_is_logged(self):
+        from utils import trash
+
+        self.write("a.png", b"aaa")
+
+        def move(path, roots):
+            raise trash.TrashRefused("no-permission", "NSCocoaErrorDomain 513")
+
+        self.fake.move = move
+        with self.assertLogs("prompt_manager.api", level="INFO") as logs:
+            _plan, job = await self.delete(["a.png"], "trash")
+        self.assertEqual(
+            self.details(job, "not_changed"), [image_delete.REASONS["no-permission"]]
+        )
+        self.assertTrue(any("NSCocoaErrorDomain 513" in line for line in logs.output))
+        self.assertTrue((self.output_dir / "a.png").exists())
+
+
 class TestNativeBackend(ImageDeleteTestCase):
-    """The real backend of this OS: Windows and macOS have no trash yet (#480)."""
+    """The real backend of this OS: Windows has no trash yet (#500).
+
+    Only library mode is applied, so even on macOS and Linux no file reaches
+    a real trash.
+    """
 
     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")
+        native_trash = sys.platform != "win32"
         if not native_trash:
             self.assertEqual(plan["trash"], {"count": 0})
-            self.assertEqual(plan["trash_refused"]["counts"], {"no-trash": 1})
+            self.assertEqual(plan["trash_refused"]["counts"], {"windows-recycle": 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)

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"])

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

tests/test_thumbnail_jobs.py

@@ -490,16 +490,25 @@ class TestThumbnailJob(ThumbnailJobCase):
         real_write = images_module._write_image_thumbnail
         job_ids = []
         written = []
+        # #504: the worker can write its first preview before the response
+        # carrying the job id is read. The first write waits until the test
+        # has stored the id, so the cancel always lands after exactly one write.
+        first_written, id_stored = threading.Event(), threading.Event()
+        self.addCleanup(id_stored.set)
 
         def counting(src, dst, *args, **kwargs):
             real_write(src, dst, *args, **kwargs)
             written.append(dst)
             if len(written) == 1:
+                first_written.set()
+                id_stored.wait(10)
                 self.api._jobs.cancel(job_ids[0])
 
         with patch.object(images_module, "_write_image_thumbnail", counting):
             _, body = await self._start()
+            await self._wait(first_written)
             job_ids.append(body["job"]["id"])
+            id_stored.set()
             job = await self._finished(body["job"]["id"])
         self.assertEqual(job.state, "cancelled")
         self.assertEqual(job.status_line, "Cancelled · 1 preview written")

tests/test_trash_freedesktop.py

@@ -277,14 +277,26 @@ class UnavailableBackendTests(unittest.TestCase):
             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
-                )
+    def test_each_platform_gets_its_backend(self):
+        from utils import trash_macos
+
+        self.assertIsInstance(trash.default_backend("win32"), trash.UnavailableBackend)
+        self.assertIsInstance(
+            trash.default_backend("darwin"), trash_macos.MacTrashBackend
+        )
         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)
+
     def test_module_functions_use_the_replaceable_backend(self):
         fake = mock.Mock()
         fake.available.return_value = "network"

tests/test_trash_macos.py

@@ -0,0 +1,360 @@
+"""The macOS trash backend's decisions (#499), with a fake Foundation bridge.
+
+No Objective-C is involved: the bridge is faked, ``uid``/``home``/``lstat``/
+``access`` are injected, and the only real files are temp files used for the
+"source still exists" and output-root identity checks. Runs on every OS.
+"""
+
+import os
+import shutil
+import stat
+import sys
+import tempfile
+import unittest
+from types import SimpleNamespace
+from unittest import mock
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from utils import trash, trash_macos  # noqa: E402
+from utils.trash_macos import Facts, MacTrashBackend, NSErr  # noqa: E402
+
+LOCAL = Facts(local=True, read_only=False, root_fs=True, volume="/", ubiquitous=False)
+COCOA = "NSCocoaErrorDomain"
+NO_TRASH_DIR = NSErr(COCOA, 3328, "NSPOSIXErrorDomain", 2)
+
+
+class FakeBridge:
+    """The §6.3 bridge with canned answers; ``trash`` moves nothing by default."""
+
+    def __init__(self, facts=LOCAL, trash_dir=("/Users/me/.Trash", None), result=None):
+        self.facts_value = facts
+        self.trash_dir_value = trash_dir
+        self.result = result or (True, None, None)
+        self.calls = []
+
+    def facts(self, path):
+        self.calls.append(("facts", path))
+        return self.facts_value
+
+    def trash_dir(self, path):
+        self.calls.append(("trash_dir", path))
+        return self.trash_dir_value
+
+    def trash(self, path):
+        self.calls.append(("trash", path))
+        result = self.result
+        return result(path) if callable(result) else result
+
+
+def _stat(mode=stat.S_IFREG | 0o644, uid=501, flags=0):
+    return SimpleNamespace(st_mode=mode, st_uid=uid, st_flags=flags)
+
+
+class MacBackendTestCase(unittest.TestCase):
+    def setUp(self):
+        # realpath: the backend resolves paths (/var -> /private/var on macOS, 8.3
+        # short names on Windows), so the lstat fakes must use the same spelling.
+        self.tmp = os.path.realpath(tempfile.mkdtemp(prefix="pm-mac-trash-"))
+        self.addCleanup(shutil.rmtree, self.tmp, True)
+        self.home = os.path.join(self.tmp, "home")
+        self.root = os.path.join(self.tmp, "output")
+        os.makedirs(os.path.join(self.home, ".Trash"))
+        os.makedirs(self.root)
+        self.file = os.path.join(self.root, "a.png")
+        with open(self.file, "wb") as handle:
+            handle.write(b"png")
+        self.home_trash = os.path.join(self.home, ".Trash")
+        self.stats = {}
+        self.access_ok = True
+
+    def _lstat(self, path):
+        if path in self.stats:
+            value = self.stats[path]
+            if isinstance(value, Exception):
+                raise value
+            return value
+        return os.lstat(path)
+
+    def backend(self, **bridge_args):
+        bridge_args.setdefault("trash_dir", (self.home_trash, None))
+        self.bridge = FakeBridge(**bridge_args)
+        return MacTrashBackend(
+            bridge=self.bridge,
+            uid=501,
+            home=self.home,
+            lstat=self._lstat,
+            access=lambda path, mode: self.access_ok,
+        )
+
+    def moved_to(self, destination):
+        """A ``trash`` result that really renames the file to *destination*."""
+
+        def run(path):
+            os.makedirs(os.path.dirname(destination), exist_ok=True)
+            os.rename(path, destination)
+            return True, destination, None
+
+        return run
+
+    def assert_refused(self, backend, code):
+        self.assertEqual(backend.available(self.file, [self.root]), code)
+        with self.assertRaises(trash.TrashRefused) as caught:
+            backend.move(self.file, [self.root])
+        self.assertEqual(caught.exception.code, code)
+        self.assertNotIn("trash", [name for name, _ in self.bridge.calls])
+        self.assertTrue(os.path.exists(self.file))
+        return caught.exception
+
+
+class AvailableOrderTests(MacBackendTestCase):
+    def test_a_writable_local_file_is_available(self):
+        self.assertIsNone(self.backend().available(self.file, [self.root]))
+
+    def test_network_volume_is_refused_without_trying(self):
+        facts = Facts(False, False, False, "/Volumes/nas", False)
+        self.assert_refused(self.backend(facts=facts), trash.NETWORK)
+
+    def test_unknown_locality_counts_as_network(self):
+        facts = Facts(None, False, True, "/", False)
+        self.assert_refused(self.backend(facts=facts), trash.NETWORK)
+
+    def test_read_only_volume_or_unknown_is_no_trash(self):
+        for read_only in (True, None):
+            with self.subTest(read_only=read_only):
+                facts = Facts(True, read_only, False, "/Volumes/ro", False)
+                self.assert_refused(self.backend(facts=facts), trash.NO_TRASH)
+
+    def test_ubiquitous_item_is_cloud_sync(self):
+        facts = Facts(True, False, True, "/", True)
+        self.assert_refused(self.backend(facts=facts), trash.CLOUD_SYNC)
+
+    def test_icloud_and_file_provider_folders_are_cloud_sync(self):
+        for parts in (("Library", "Mobile Documents"), ("Library", "CloudStorage")):
+            with self.subTest(parts=parts):
+                folder = os.path.join(self.home, *parts, "Box")
+                os.makedirs(folder)
+                path = os.path.join(folder, "b.png")
+                with open(path, "wb") as handle:
+                    handle.write(b"x")
+                backend = self.backend()
+                self.assertEqual(backend.available(path, [folder]), trash.CLOUD_SYNC)
+                with self.assertRaises(trash.TrashRefused):
+                    backend.move(path, [folder])
+                self.assertTrue(os.path.exists(path))
+
+    def test_found_trash_folder_inside_an_output_root_is_trash_in_root(self):
+        self.assertEqual(
+            self.backend().available(self.file, [self.home]), trash.TRASH_IN_ROOT
+        )
+
+    def test_trash_dir_error_other_than_missing_is_no_trash(self):
+        backend = self.backend(
+            trash_dir=(None, NSErr(COCOA, 3328, "NSPOSIXErrorDomain", 30))
+        )
+        self.assert_refused(backend, trash.NO_TRASH)
+
+    def test_missing_trash_on_the_startup_volume_is_no_trash(self):
+        self.assert_refused(
+            self.backend(trash_dir=(None, NO_TRASH_DIR)), trash.NO_TRASH
+        )
+
+
+class FreshVolumeTests(MacBackendTestCase):
+    def setUp(self):
+        super().setUp()
+        self.volume = os.path.join(self.tmp, "vol")
+        self.root = os.path.join(self.volume, "output")
+        os.makedirs(self.root)
+        self.file = os.path.join(self.root, "a.png")
+        with open(self.file, "wb") as handle:
+            handle.write(b"png")
+        self.facts = Facts(True, False, False, self.volume, False)
+
+    def test_missing_volume_trash_is_predicted(self):
+        backend = self.backend(facts=self.facts, trash_dir=(None, NO_TRASH_DIR))
+        self.assertIsNone(backend.available(self.file, [self.root]))
+
+    def test_prediction_refused_when_trashes_is_a_link(self):
+        os.symlink(self.tmp, os.path.join(self.volume, ".Trashes"))
+        backend = self.backend(facts=self.facts, trash_dir=(None, NO_TRASH_DIR))
+        self.assert_refused(backend, trash.NO_TRASH)
+
+    def test_predicted_trash_inside_the_volume_root_is_trash_in_root(self):
+        backend = self.backend(facts=self.facts, trash_dir=(None, NO_TRASH_DIR))
+        self.assert_refused_root = backend.available(self.file, [self.volume])
+        self.assertEqual(self.assert_refused_root, trash.TRASH_IN_ROOT)
+
+    def test_move_to_a_predicted_trash_returns_the_volume_label(self):
+        destination = os.path.join(self.volume, ".Trashes", "501", "a.png")
+        backend = self.backend(
+            facts=self.facts,
+            trash_dir=(None, NO_TRASH_DIR),
+            result=self.moved_to(destination),
+        )
+        self.assertEqual(
+            backend.move(self.file, [self.root]), ".Trashes/501 on its drive"
+        )
+        self.assertFalse(os.path.exists(self.file))
+
+
+class PermissionTests(MacBackendTestCase):
+    def test_parent_not_writable_is_no_permission(self):
+        self.access_ok = False
+        self.assert_refused(self.backend(), trash.NO_PERMISSION)
+
+    def test_immutable_or_append_flags_on_file_or_parent(self):
+        for flag in ("UF_IMMUTABLE", "SF_IMMUTABLE", "UF_APPEND", "SF_APPEND"):
+            for target in (self.file, self.root):
+                with self.subTest(flag=flag, target=target):
+                    mode = (
+                        stat.S_IFDIR | 0o755
+                        if target == self.root
+                        else stat.S_IFREG | 0o644
+                    )
+                    self.stats = {target: _stat(mode=mode, flags=getattr(stat, flag))}
+                    self.assert_refused(self.backend(), trash.NO_PERMISSION)
+
+    def test_sticky_parent_and_foreign_owner(self):
+        self.stats = {
+            self.root: _stat(mode=stat.S_IFDIR | 0o1777),
+            self.file: _stat(uid=502),
+        }
+        self.assert_refused(self.backend(), trash.NO_PERMISSION)
+
+    def test_sticky_parent_with_own_file_is_allowed(self):
+        self.stats = {self.root: _stat(mode=stat.S_IFDIR | 0o1777), self.file: _stat()}
+        self.assertIsNone(self.backend().available(self.file, [self.root]))
+
+
+class MoveOutcomeTests(MacBackendTestCase):
+    def test_success_into_the_home_trash(self):
+        destination = os.path.join(self.home_trash, "a.png")
+        backend = self.backend(result=self.moved_to(destination))
+        self.assertEqual(backend.move(self.file, [self.root]), trash.HOME_TRASH_LABEL)
+        self.assertTrue(os.path.exists(destination))
+
+    def test_cocoa_errors_map_to_codes_with_detail(self):
+        cases = {
+            3328: trash.NO_TRASH,
+            642: trash.NO_TRASH,
+            513: trash.NO_PERMISSION,
+            257: trash.NO_PERMISSION,
+            4: trash.FAILED,
+            640: trash.FAILED,
+        }
+        for code, expected in cases.items():
+            with self.subTest(code=code):
+                error = NSErr(COCOA, code, "NSOSStatusErrorDomain", -5000)
+                backend = self.backend(result=(False, None, error))
+                with self.assertRaises(trash.TrashRefused) as caught:
+                    backend.move(self.file, [self.root])
+                self.assertEqual(caught.exception.code, expected)
+                self.assertEqual(
+                    caught.exception.detail,
+                    f"NSCocoaErrorDomain {code} (NSOSStatusErrorDomain -5000)",
+                )
+                self.assertTrue(os.path.exists(self.file))
+
+    def test_other_domain_or_missing_error_is_failed(self):
+        for error in (NSErr("NSPOSIXErrorDomain", 13), None):
+            with self.subTest(error=error):
+                backend = self.backend(result=(False, None, error))
+                with self.assertRaises(trash.TrashRefused) as caught:
+                    backend.move(self.file, [self.root])
+                self.assertEqual(caught.exception.code, trash.FAILED)
+
+    def test_yes_with_the_source_still_present_is_failed(self):
+        backend = self.backend(
+            result=(True, os.path.join(self.home_trash, "a.png"), None)
+        )
+        with self.assertRaises(trash.TrashRefused) as caught:
+            backend.move(self.file, [self.root])
+        self.assertEqual(caught.exception.code, trash.FAILED)
+        self.assertTrue(os.path.exists(self.file))
+
+    def test_yes_with_no_result_path_is_unconfirmed(self):
+        def vanished(path):
+            os.rename(path, os.path.join(self.tmp, "elsewhere.png"))
+            return True, None, None
+
+        backend = self.backend(result=vanished)
+        with self.assertRaises(trash.TrashUnconfirmed):
+            backend.move(self.file, [self.root])
+
+    def test_yes_into_an_output_root_is_unconfirmed(self):
+        backend = self.backend(
+            result=self.moved_to(os.path.join(self.root, "x", "a.png"))
+        )
+        with self.assertRaises(trash.TrashUnconfirmed):
+            backend.move(self.file, [self.root])
+
+    def test_yes_into_an_unplanned_folder_is_trashed_and_logged(self):
+        other = os.path.join(self.tmp, "other-trash", "a.png")
+        backend = self.backend(result=self.moved_to(other))
+        with self.assertLogs("prompt_manager.trash", level="WARNING") as logs:
+            label = backend.move(self.file, [self.root])
+        self.assertEqual(label, os.path.realpath(os.path.dirname(other)))
+        self.assertIn(other, "\n".join(logs.output))
+
+
+class IdentityTests(MacBackendTestCase):
+    @unittest.skipUnless(hasattr(os, "symlink") and os.name == "posix", "symlinks")
+    def test_root_given_through_a_symlink_still_contains_its_trash(self):
+        alias = os.path.join(self.tmp, "home-alias")
+        os.symlink(self.home, alias)
+        self.assertEqual(
+            self.backend().available(self.file, [alias]), trash.TRASH_IN_ROOT
+        )
+
+    def test_missing_roots_are_ignored(self):
+        missing = os.path.join(self.tmp, "nope")
+        self.assertIsNone(self.backend().available(self.file, [missing, self.root]))
+
+
+class LoadingTests(unittest.TestCase):
+    def test_default_backend_for_darwin_does_not_load_the_bridge(self):
+        with mock.patch.object(trash_macos, "load_bridge") as loader:
+            backend = trash.default_backend("darwin")
+        self.assertIsInstance(backend, MacTrashBackend)
+        loader.assert_not_called()
+
+    def test_failed_load_answers_no_trash_and_is_cached_and_logged_once(self):
+        backend = MacTrashBackend(uid=501)
+        with (
+            mock.patch.object(
+                trash_macos, "load_bridge", side_effect=OSError("no Foundation")
+            ) as loader,
+            self.assertLogs("prompt_manager.trash", level="WARNING") as logs,
+        ):
+            self.assertEqual(backend.available("/x/a.png", ["/x"]), trash.NO_TRASH)
+            with self.assertRaises(trash.TrashRefused) as caught:
+                backend.move("/x/a.png", ["/x"])
+        self.assertEqual(caught.exception.code, trash.NO_TRASH)
+        self.assertEqual(loader.call_count, 1)
+        self.assertEqual(len(logs.output), 1)
+
+    def test_load_bridge_off_macos_raises(self):
+        if sys.platform == "darwin":
+            self.skipTest("the real bridge loads on macOS (native test covers it)")
+        with self.assertRaises(OSError):
+            trash_macos.load_bridge()
+
+
+class ErrorTextTests(unittest.TestCase):
+    def test_detail_without_underlying(self):
+        self.assertEqual(
+            trash_macos.error_detail(NSErr(COCOA, 4)), "NSCocoaErrorDomain 4"
+        )
+        self.assertEqual(trash_macos.error_detail(None), "no NSError")
+
+    def test_refusal_keeps_detail(self):
+        refused = trash.TrashRefused(trash.FAILED, "why")
+        self.assertEqual(
+            (refused.code, refused.detail, str(refused)), ("failed", "why", "why")
+        )
+        self.assertEqual(trash.TrashRefused(trash.FAILED).detail, "")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_trash_macos_native.py

@@ -0,0 +1,238 @@
+"""Native proof of the macOS trash backend (#499 design §10.2).
+
+Runs only on macOS with ``PM_NATIVE_TRASH=macos`` (the ``native-trash`` job in
+portability.yml), because it puts files into the real Trash of the user
+running it. With the variable set, a bridge that cannot load fails the test
+rather than skipping it. Files moved into a trash folder are left there: the
+runner is disposable and this test never deletes from a trash folder.
+"""
+
+import os
+import subprocess
+import sys
+import tempfile
+import time
+import unittest
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from utils import trash  # noqa: E402
+
+NATIVE = sys.platform == "darwin" and os.environ.get("PM_NATIVE_TRASH") == "macos"
+HDIUTIL_SECONDS = 60
+# Measured: macos-15-intel kept a fresh image busy after a trash (run 40954618922).
+DETACH_WAIT_SECONDS = 30
+
+
+class SpyBridge:
+    """The real bridge, recording every ``trash`` result."""
+
+    def __init__(self, real):
+        self.real = real
+        self.results = []
+
+    def facts(self, path):
+        return self.real.facts(path)
+
+    def trash_dir(self, path):
+        return self.real.trash_dir(path)
+
+    def trash(self, path):
+        result = self.real.trash(path)
+        self.results.append(result)
+        return result
+
+
+def _hdiutil(*args):
+    return subprocess.run(
+        ["hdiutil", *args], capture_output=True, text=True, timeout=HDIUTIL_SECONDS
+    )
+
+
+def _our_open_files(mount):
+    """Files this process holds open under *mount* (``lsof -p``): the handle proof."""
+    listed = subprocess.run(
+        ["lsof", "-Fn", "-p", str(os.getpid())],
+        capture_output=True,
+        text=True,
+        timeout=HDIUTIL_SECONDS,
+    )
+    prefix = os.path.realpath(mount) + os.sep
+    return [
+        line[1:]
+        for line in listed.stdout.splitlines()
+        if line.startswith("n") and os.path.realpath(line[1:]).startswith(prefix)
+    ]
+
+
+def _detach(mount):
+    """Detach *mount*; None on success, else why (after a forced detach).
+
+    This process must hold nothing under it. A volume macOS has just written a
+    trash folder to can stay busy for a moment (Spotlight, fseventsd), so the
+    plain detach is retried until the system releases it, within a bound.
+    """
+    ours = _our_open_files(mount)
+    if ours:
+        _hdiutil("detach", mount, "-force")
+        return f"this test still held files open under {mount}: {ours}"
+    deadline = time.monotonic() + DETACH_WAIT_SECONDS
+    while True:
+        if not os.path.ismount(mount):
+            return None  # a "busy" detach can still finish after it returns
+        done = _hdiutil("detach", mount)
+        if done.returncode == 0:
+            return None
+        if time.monotonic() > deadline:
+            holders = subprocess.run(
+                ["lsof", "+f", "--", mount],
+                capture_output=True,
+                text=True,
+                timeout=HDIUTIL_SECONDS,
+            ).stdout
+            _hdiutil("detach", mount, "-force")
+            return f"detach {mount}: {done.stderr.strip()}; held by: {holders!r}"
+        time.sleep(1)  # poll the volume's release (a state), never a fixed wait
+
+
+@unittest.skipUnless(NATIVE, "macOS native trash proof (PM_NATIVE_TRASH=macos)")
+class MacTrashNativeTests(unittest.TestCase):
+    def setUp(self):
+        from utils import trash_macos
+
+        self.trash_macos = trash_macos
+        self.spy = SpyBridge(trash_macos.load_bridge())  # fails, never skips
+        self.backend = trash_macos.MacTrashBackend(bridge=self.spy)
+        self._tmp = tempfile.TemporaryDirectory(prefix="pm-native-trash-")
+        self.tmp = os.path.realpath(self._tmp.name)
+        self._mounts = []
+        self._modes = []
+
+    def tearDown(self):
+        failures = []
+        try:
+            for path, mode in reversed(self._modes):
+                os.chmod(path, mode)
+            for mount in reversed(self._mounts):
+                problem = _detach(mount)
+                if problem:
+                    failures.append(problem)
+        finally:
+            self._tmp.cleanup()
+        self.assertEqual(failures, [])
+
+    def _file(self, folder, name="a.png", content=None):
+        os.makedirs(folder, exist_ok=True)
+        path = os.path.join(folder, name)
+        with open(path, "wb") as handle:
+            handle.write(content or os.urandom(64))
+        return path
+
+    def _image(self, name, fs="APFS"):
+        image = os.path.join(self.tmp, name + ".dmg")
+        mount = os.path.join(self.tmp, name + "-mnt")
+        os.makedirs(mount)
+        made = _hdiutil("create", "-size", "40m", "-fs", fs, "-volname", name, image)
+        self.assertEqual(made.returncode, 0, made.stderr)
+        attached = _hdiutil("attach", image, "-nobrowse", "-mountpoint", mount)
+        self.assertEqual(attached.returncode, 0, attached.stderr)
+        self._mounts.append(mount)
+        return image, mount
+
+    def _assert_trashed(self, source, content, info, folder):
+        self.assertFalse(os.path.lexists(source))
+        ok, resulting, error = self.spy.results[-1]
+        self.assertTrue(ok, error)
+        self.assertEqual(os.path.realpath(os.path.dirname(resulting)), folder)
+        moved = os.lstat(resulting)
+        self.assertEqual((moved.st_dev, moved.st_ino), (info.st_dev, info.st_ino))
+        with open(resulting, "rb") as handle:
+            self.assertEqual(handle.read(), content)
+        return resulting
+
+    def _assert_kept(self, path, content):
+        with open(path, "rb") as handle:
+            self.assertEqual(handle.read(), content)
+
+    def _assert_refused(self, path, roots, code):
+        with open(path, "rb") as handle:
+            content = handle.read()
+        self.assertEqual(self.backend.available(path, roots), code)
+        with self.assertRaises(trash.TrashRefused) as caught:
+            self.backend.move(path, roots)
+        self.assertEqual(caught.exception.code, code)
+        self._assert_kept(path, content)
+
+    def test_n1_startup_volume_goes_to_the_home_trash(self):
+        root = os.path.join(self.tmp, "output")
+        source = self._file(root)
+        with open(source, "rb") as handle:
+            content = handle.read()
+        info = os.lstat(source)
+        self.assertIsNone(self.backend.available(source, [root]))
+        self.assertEqual(self.backend.move(source, [root]), trash.HOME_TRASH_LABEL)
+        home_trash = os.path.realpath(os.path.expanduser("~/.Trash"))
+        self._assert_trashed(source, content, info, home_trash)
+
+    def test_n2_a_name_clash_never_replaces(self):
+        first_root = os.path.join(self.tmp, "one")
+        second_root = os.path.join(self.tmp, "two")
+        first = self._file(first_root, "same.png", b"first")
+        second = self._file(second_root, "same.png", b"second")
+        self.backend.move(first, [first_root])
+        self.backend.move(second, [second_root])
+        (_, one, _), (_, two, _) = self.spy.results
+        self.assertNotEqual(one, two)
+        self._assert_kept(one, b"first")
+        self._assert_kept(two, b"second")
+
+    def test_n3_fresh_external_volume_uses_its_own_trash(self):
+        for fs in ("APFS", "MS-DOS FAT32"):
+            with self.subTest(fs=fs):
+                _image, mount = self._image("fresh" + fs.split()[0], fs)
+                root = os.path.join(mount, "output")
+                source = self._file(root)
+                with open(source, "rb") as handle:
+                    content = handle.read()
+                info = os.lstat(source)
+                self.assertFalse(os.path.lexists(os.path.join(mount, ".Trashes")))
+                self.assertIsNone(self.backend.available(source, [root]))
+                label = self.backend.move(source, [root])
+                uid = str(os.getuid())
+                self.assertEqual(label, f".Trashes/{uid} on its drive")
+                folder = os.path.realpath(os.path.join(mount, ".Trashes", uid))
+                self._assert_trashed(source, content, info, folder)
+
+    def test_n4_volume_root_as_output_root_is_trash_in_root(self):
+        _image, mount = self._image("rooted")
+        source = self._file(os.path.join(mount, "output"))
+        self._assert_refused(source, [mount], trash.TRASH_IN_ROOT)
+        self.assertEqual(self.spy.results, [])
+
+    def test_n5_read_only_volume_is_no_trash(self):
+        image, mount = self._image("readonly")
+        source = self._file(mount, "ro.png", b"read-only bytes")
+        self.assertIsNone(_detach(mount))
+        self._mounts.remove(mount)
+        attached = _hdiutil(
+            "attach", image, "-nobrowse", "-readonly", "-mountpoint", mount
+        )
+        self.assertEqual(attached.returncode, 0, attached.stderr)
+        self._mounts.append(mount)
+        self._assert_refused(source, [os.path.join(mount, "output")], trash.NO_TRASH)
+
+    def test_n6_unwritable_folder_is_no_permission_and_macos_agrees(self):
+        root = os.path.join(self.tmp, "locked")
+        source = self._file(root, "p.png", b"kept")
+        self._modes.append((root, 0o755))
+        os.chmod(root, 0o555)
+        self._assert_refused(source, [self.tmp], trash.NO_PERMISSION)
+        self.assertEqual(self.spy.results, [])
+        ok, _resulting, error = self.spy.real.trash(source)
+        self.assertFalse(ok)
+        self.assertEqual((error.domain, error.code), ("NSCocoaErrorDomain", 513))
+        self._assert_kept(source, b"kept")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/vendor/browser-logs.test.mjs

@@ -0,0 +1,127 @@
+import test from 'node:test';
+import assert from 'node:assert/strict';
+import { mkdtempSync, readFileSync, rmSync } from 'node:fs';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
+import { attachBrowserLogs, BROWSER_EXIT_PREFIX, createBrowserLogRecorder, isAbnormalExit, isTargetGone, parseExitLine,
+    playwrightDebug } from '../../tools/vendor/browser-logs.mjs';
+
+const debug = playwrightDebug();
+
+/** Runs `body` with DEBUG set to `namespaces` and restores the caller's setting afterwards. */
+function withDebug(namespaces, body) {
+    const saved = debug.disable();
+    const savedLog = debug.log;
+    if (namespaces) debug.enable(namespaces);
+    try {
+        return body();
+    } finally {
+        debug.log = savedLog;
+        debug.disable();
+        if (saved) debug.enable(saved);
+    }
+}
+
+const outputDir = mkdtempSync(join(tmpdir(), 'pm-browser-logs-'));
+test.after(() => rmSync(outputDir, { recursive: true, force: true }));
+
+function testInfo({ status = 'failed', errors = [] } = {}) {
+    const attachments = [];
+    const dir = mkdtempSync(join(outputDir, 'test-'));
+    return { status, expectedStatus: 'passed', errors, attachments, outputPath: name => join(dir, name),
+        attach: async (name, options) => { attachments.push({ name, ...options }); } };
+}
+
+test('parses the exit line Playwright logs for a browser process', () => {
+    assert.deepEqual(parseExitLine('[pid=7124] <process did exit: exitCode=null, signal=SIGKILL> +21ms'),
+        { pid: 7124, exitCode: null, signal: 'SIGKILL' });
+    assert.deepEqual(parseExitLine('[pid=9] <process did exit: exitCode=0, signal=null>'), { pid: 9, exitCode: 0, signal: null });
+    assert.deepEqual(parseExitLine('[pid=9] <process did exit: exitCode=11, signal=null>'), { pid: 9, exitCode: 11, signal: null });
+    assert.equal(parseExitLine('[pid=9][err] JavaScript warning'), null);
+    assert.equal(isAbnormalExit({ pid: 9, exitCode: 0, signal: null }), false);
+    assert.equal(isAbnormalExit({ pid: 9, exitCode: null, signal: 'SIGKILL' }), true);
+    assert.equal(isAbnormalExit({ pid: 9, exitCode: 139, signal: null }), true);
+});
+
+test('recognises the errors of a page, context or browser that went away', () => {
+    for (const message of ['Error: browser.newContext: Target page, context or browser has been closed',
+        'Protocol error (Page.navigate): Target closed', 'page.goto: Page crashed', 'Target crashed']) {
+        assert.equal(isTargetGone(message), true, message);
+    }
+    for (const message of ['expect(locator).toBeVisible() failed', undefined, 'Test timeout of 30000ms exceeded.']) {
+        assert.equal(isTargetGone(message), false, String(message));
+    }
+});
+
+test('records the browser channel into a bounded tail without printing it', () => withDebug('', () => {
+    const printed = [];
+    debug.log = (...args) => printed.push(args.join(' '));
+    const written = [];
+    const recorder = createBrowserLogRecorder({ debug, limit: 3, write: text => written.push(text) });
+    const browser = debug('pw:browser');
+    browser('<launched> pid=5');
+    for (const n of [1, 2, 3, 4]) browser(`[pid=5][err] line ${n}`);
+    assert.deepEqual(printed, [], 'the browser channel stays quiet when DEBUG did not ask for it');
+    assert.match(recorder.tail(), /line 2[\s\S]*line 3[\s\S]*line 4/);
+    assert.doesNotMatch(recorder.tail(), /line 1/);
+    assert.deepEqual(written, []);
+    assert.deepEqual(recorder.launches(), [5], 'launches outlive the bounded tail');
+    browser('[pid=5] <process did exit: exitCode=null, signal=SIGKILL>');
+    assert.deepEqual(recorder.exits(), [{ pid: 5, exitCode: null, signal: 'SIGKILL' }]);
+    assert.deepEqual(written, [`${BROWSER_EXIT_PREFIX}browser pid=5 exited: exitCode=null, signal=SIGKILL\n`]);
+    browser('[pid=6] <process did exit: exitCode=0, signal=null>');
+    assert.equal(written.length, 1, 'a clean exit prints nothing');
+    recorder.uninstall();
+    assert.equal(debug.enabled('pw:browser'), false, 'uninstall restores the caller\'s DEBUG');
+}));
+
+test('keeps printing the channels the user enabled, the browser one included', () => withDebug('pw:api,pw:browser', () => {
+    const printed = [];
+    debug.log = function (...args) { printed.push(`${this.namespace} ${args.join(' ')}`); };
+    const recorder = createBrowserLogRecorder({ debug, write: () => {} });
+    debug('pw:api')('call');
+    debug('pw:browser')('[pid=1][out] hello');
+    debug('pw:protocol')('not enabled');
+    assert.equal(printed.length, 2);
+    assert.match(printed[0], /^pw:api /);
+    assert.match(printed[1], /^pw:browser /);
+    assert.doesNotMatch(recorder.tail(), /call/);
+    recorder.uninstall();
+    assert.equal(debug.enabled('pw:api'), true);
+    assert.equal(debug.enabled('pw:browser'), true);
+}));
+
+test('attaches the tail and the exit only to a test whose target went away', async () => withDebug('', async () => {
+    debug.log = () => {};
+    const recorder = createBrowserLogRecorder({ debug, write: () => {} });
+    const browser = debug('pw:browser');
+    browser('\u001b[30m[pid=8][err] Crash Annotation GraphicsCriticalError\u001b[0m');
+    browser('[pid=8] <process did exit: exitCode=null, signal=SIGKILL>');
+
+    const passed = testInfo({ status: 'passed' });
+    assert.equal(await attachBrowserLogs(passed, recorder), false);
+    const assertion = testInfo({ errors: [{ message: 'expect(locator).toBeVisible() failed' }] });
+    assert.equal(await attachBrowserLogs(assertion, recorder), false);
+    assert.deepEqual(assertion.attachments, []);
+
+    const closed = testInfo({ errors: [{ message: 'Error: browser.newContext: Target page, context or browser has been closed' }] });
+    assert.equal(await attachBrowserLogs(closed, recorder), true);
+    assert.equal(closed.attachments.length, 1);
+    const [{ name, path, contentType }] = closed.attachments;
+    assert.equal(name, 'browser-stderr');
+    assert.equal(contentType, 'text/plain');
+    const body = readFileSync(path, 'utf8');
+    assert.match(body, /^browser pid=8 exited: exitCode=null, signal=SIGKILL\n/);
+    assert.match(body, /\[pid=8\]\[err\] Crash Annotation GraphicsCriticalError/);
+    assert.doesNotMatch(body, /\u001b/, 'the attached log has no terminal colours');
+    recorder.uninstall();
+}));
+
+test('says so when the worker logged no abnormal exit', async () => withDebug('', async () => {
+    debug.log = () => {};
+    const recorder = createBrowserLogRecorder({ debug, write: () => {} });
+    const closed = testInfo({ errors: [{ message: 'x', stack: 'Error: Target closed\n    at y' }] });
+    assert.equal(await attachBrowserLogs(closed, recorder), true);
+    assert.match(readFileSync(closed.attachments[0].path, 'utf8'), /^no abnormal browser exit was logged in this worker\n/);
+    recorder.uninstall();
+}));

tests/vendor/ci-workflow.test.mjs

@@ -353,6 +353,20 @@ test('Windows and macOS run 3.12-3.14 on main, nightly and on demand, never on a
     assert.doesNotMatch(text, /cancel-in-progress: true|cancel-in-progress: \$/, 'no portability run is cancelled');
 });
 
+// #499: the native trash proof runs on both macOS CPUs, apart from the suite so a red suite never skips it.
+test('portability runs the native trash proof on arm64 and Intel macOS', () => {
+    const body = jobs(readWorkflow('portability.yml')).get('native-trash');
+    assert.ok(body, 'portability.yml has a native-trash job');
+    assert.deepEqual(matrixList(body, 'os'), ['macos-latest', 'macos-15-intel']);
+    assert.match(body, /^ {6}fail-fast: false$/m);
+    assert.doesNotMatch(body, /^ {4}needs:/m, 'it does not wait for the suite job');
+    assert.match(body, /^ {10}PM_NATIVE_TRASH: macos$/m);
+    assert.match(body, /^ {8}run: python -m unittest tests\.test_trash_macos_native tests\.test_trash_macos -v$/m);
+    const jobMinutes = Number(/^ {4}timeout-minutes: (\d+)$/m.exec(body)[1]);
+    const stepMinutes = Number(/^ {8}timeout-minutes: (\d+)$/m.exec(body)?.[1]);
+    assert.ok(stepMinutes > 0 && jobMinutes - stepMinutes >= 2, 'the proof step stops before the job limit');
+});
+
 // #493: the portability limit follows the rule in its comment: ceil(1.5 x the slowest of three
 // green main runs) + 3 minutes, never under 5, with those runs and times written above jobs:.
 // The times are whole-job wall times. That the runs were green and on main is the author's word.

tools/vendor/browser-logs.mjs

@@ -0,0 +1,91 @@
+// What a browser printed before it died (#513). Playwright writes a browser's stderr, its launch and its exit
+// (`<process did exit: exitCode=…, signal=…>`) only to its `pw:browser` debug channel, which is off unless DEBUG is
+// set, so a browser that dies between two tests leaves the next test only "Target page, context or browser has been
+// closed". A worker records that channel into a bounded tail, prints one line when a browser exits abnormally and
+// attaches the tail to a test that fails on a closed or crashed target. Nothing is retried or hidden.
+import { createRequire } from 'node:module';
+import { writeFile } from 'node:fs/promises';
+import { format, stripVTControlCharacters } from 'node:util';
+
+const BROWSER_NAMESPACE = 'pw:browser';
+export const BROWSER_LOG_LIMIT = 300;
+export const BROWSER_EXIT_PREFIX = 'pm-browser-exit| ';
+const LAUNCH_LINE = /<launched> pid=(\d+)/;
+const EXIT_LINE = /\[pid=(\d+)\] <process did exit: exitCode=([^,]+), signal=([^>]+)>/;
+// The client's TargetClosedError, a protocol error of a closed or crashed target, and a crashed page.
+const TARGET_GONE = /Target page, context or browser has been closed|Target (?:closed|crashed)|Browser closed|browser has disconnected|Page crashed/i;
+
+/** The `debug` module Playwright logs through (the instance this worker's tests launch browsers with). */
+export function playwrightDebug() {
+    return createRequire(import.meta.url)('playwright-core/lib/utilsBundle').debug;
+}
+
+/** The pid, exit code and signal of a `process did exit` line, or null. */
+export function parseExitLine(line) {
+    const match = EXIT_LINE.exec(line);
+    if (!match) return null;
+    const value = text => (text === 'null' ? null : text);
+    const code = value(match[2]);
+    return { pid: Number(match[1]), exitCode: code === null ? null : Number(code), signal: value(match[3]) };
+}
+
+export const isAbnormalExit = exit => exit !== null && (exit.exitCode !== 0 || exit.signal !== null);
+
+/** True for an error message that says the page, context or browser under the test went away. */
+export const isTargetGone = message => TARGET_GONE.test(String(message ?? ''));
+
+/**
+ * Records `pw:browser` into a tail of at most `limit` lines. Other debug channels keep printing as before; the
+ * browser channel prints only when the user's own DEBUG asked for it. `write` gets one line per abnormal exit.
+ */
+export function createBrowserLogRecorder({ debug = playwrightDebug(), limit = BROWSER_LOG_LIMIT,
+    write = text => process.stderr.write(text) } = {}) {
+    const lines = [];
+    const launches = [];
+    const exits = [];
+    const userWantsBrowser = debug.enabled(BROWSER_NAMESPACE);
+    const previous = debug.disable();
+    const original = debug.log;
+    debug.log = function record(...args) {
+        if (this?.namespace !== BROWSER_NAMESPACE || userWantsBrowser) {
+            (original ?? (() => {})).apply(this, args);
+        }
+        if (this?.namespace !== BROWSER_NAMESPACE) return;
+        const line = stripVTControlCharacters(format(...args));
+        lines.push(line);
+        if (lines.length > limit) lines.splice(0, lines.length - limit);
+        const launched = LAUNCH_LINE.exec(line);
+        if (launched) launches.push(Number(launched[1]));
+        const exit = parseExitLine(line);
+        if (!exit) return;
+        exits.push(exit);
+        if (isAbnormalExit(exit)) {
+            write(`${BROWSER_EXIT_PREFIX}browser pid=${exit.pid} exited: exitCode=${exit.exitCode}, signal=${exit.signal}\n`);
+        }
+    };
+    debug.enable([previous, BROWSER_NAMESPACE].filter(Boolean).join(','));
+    return {
+        tail: () => lines.join('\n'),
+        launches: () => [...launches],
+        exits: () => [...exits],
+        uninstall() {
+            debug.log = original;
+            debug.disable();
+            if (previous) debug.enable(previous);
+        },
+    };
+}
+
+/** Attaches the browser log tail to a test that failed because its page, context or browser went away. */
+export async function attachBrowserLogs(testInfo, recorder) {
+    if (testInfo.status === testInfo.expectedStatus) return false;
+    if (!testInfo.errors.some(error => isTargetGone(error.message) || isTargetGone(error.stack))) return false;
+    const exits = recorder.exits().filter(isAbnormalExit)
+        .map(exit => `browser pid=${exit.pid} exited: exitCode=${exit.exitCode}, signal=${exit.signal}`);
+    const summary = exits.length ? exits.join('\n') : 'no abnormal browser exit was logged in this worker';
+    // A file, so the log travels with the uploaded test-results (the list reporter shortens a long body).
+    const path = testInfo.outputPath('browser-stderr.txt');
+    await writeFile(path, `${summary}\n--- last ${BROWSER_LOG_LIMIT} pw:browser lines ---\n${recorder.tail()}\n`);
+    await testInfo.attach('browser-stderr', { path, contentType: 'text/plain' });
+    return true;
+}

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)
 

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)

utils/trash.py

@@ -14,8 +14,11 @@ 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.
+macOS uses :class:`utils.trash_macos.MacTrashBackend` (#499): NSFileManager's
+``trashItemAtURL`` through ctypes, after refusing what it cannot show to be
+recoverable. Windows never uses the Recycle Bin (#500: files over its capacity
+are deleted permanently, undetectably), so it answers ``windows-recycle``
+through :class:`UnavailableBackend`.
 
 ``BACKEND`` is the one replaceable seam; tests install a fake.
 """
@@ -36,7 +39,23 @@ 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})
+NO_PERMISSION = "no-permission"
+CLOUD_SYNC = "cloud-sync"
+# 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,
+        NO_PERMISSION,
+        CLOUD_SYNC,
+        WINDOWS_RECYCLE,
+    }
+)
 
 HOME_TRASH_LABEL = "home trash"
 RENAME_NOREPLACE = 1
@@ -61,6 +80,19 @@ class TrashRefused(Exception):
     def __init__(self, code: str, detail: str = ""):
         super().__init__(detail or code)
         self.code = code
+        self.detail = detail
+
+
+class TrashUnconfirmed(Exception):
+    """The system reported the file moved, but not to a place shown to be a trash.
+
+    The file is no longer where it was, so this is not a refusal: the caller
+    keeps the file's records and stops (#499 design §7.4).
+    """
+
+    def __init__(self, detail: str):
+        super().__init__(detail)
+        self.detail = detail
 
 
 def parse_mountinfo(text: str) -> list[str]:
@@ -177,13 +209,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:
@@ -345,9 +380,21 @@ 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"):
-        return UnavailableBackend()
+    """The backend for *platform* (#480, #499, #500).
+
+    macOS gets :class:`utils.trash_macos.MacTrashBackend`, which loads its
+    Foundation bridge only on first use (#499). Windows never uses the Recycle
+    Bin (#500).
+    """
+    if platform == "darwin":
+        from . import trash_macos
+
+        return trash_macos.MacTrashBackend()
+    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)
     return FreedesktopBackend()
 
 

utils/trash_macos.py

@@ -0,0 +1,489 @@
+"""The macOS Trash backend (#499): NSFileManager through ctypes, or refuse.
+
+``-[NSFileManager trashItemAtURL:resultingItemURL:error:]`` moves a file into
+its volume's trash folder (``~/.Trash`` on the startup volume,
+``<volume>/.Trashes/<uid>`` elsewhere) or fails with an ``NSError``; it renames
+name clashes itself and never replaces a file. Before calling it,
+:class:`MacTrashBackend` refuses what it cannot show to be recoverable:
+network volumes, read-only volumes, iCloud Drive and File Provider sync
+folders, a trash folder inside an output root, and files macOS would refuse
+for permission. Nothing here unlinks, copies or overwrites a file.
+
+The :class:`Bridge` is the only code that touches the Objective-C runtime. It
+sends ten fixed messages, each through ``objc_msgSend`` cast to its exact
+prototype (required on arm64), checks every object for nil before a send (a
+nil can raise an Objective-C exception, which would abort the process), and
+runs each call inside its own autorelease pool. It is loaded lazily, on the
+first ``available``/``move``; when it cannot load, every file answers
+``no-trash`` as before. The facts behind these rules were measured on macOS
+14, 15 (x86_64) and 26 (docs/egm/499-macos-trash/design.md §4).
+"""
+
+from __future__ import annotations
+
+import ctypes
+import logging
+import os
+import stat
+import sys
+import threading
+from dataclasses import dataclass
+
+from . import trash
+
+logger = logging.getLogger("prompt_manager.trash")
+
+COCOA = "NSCocoaErrorDomain"
+POSIX = "NSPOSIXErrorDomain"
+FEATURE_UNSUPPORTED = 3328
+_ENOENT = 2
+# Top-level Cocoa codes of a refused move (design §7.5); anything else is "failed".
+_MOVE_CODES = {
+    FEATURE_UNSUPPORTED: trash.NO_TRASH,
+    642: trash.NO_TRASH,  # NSFileWriteVolumeReadOnlyError
+    513: trash.NO_PERMISSION,  # NSFileWriteNoPermissionError
+    257: trash.NO_PERMISSION,  # NSFileReadNoPermissionError
+}
+_LOCK_FLAGS = stat.UF_IMMUTABLE | stat.SF_IMMUTABLE | stat.UF_APPEND | stat.SF_APPEND
+_CLOUD_FOLDERS = (("Library", "Mobile Documents"), ("Library", "CloudStorage"))
+
+
+@dataclass(frozen=True)
+class NSErr:
+    domain: str
+    code: int
+    under_domain: str | None = None
+    under_code: int | None = None
+
+
+@dataclass(frozen=True)
+class Facts:
+    """Resource values of a file's URL; ``None`` means the lookup failed or was nil."""
+
+    local: bool | None
+    read_only: bool | None
+    root_fs: bool | None
+    volume: str | None
+    ubiquitous: bool | None
+
+
+def error_detail(error: NSErr | None) -> str:
+    """``"<domain> <code> (<under_domain> <under_code>)"`` for the log."""
+    if error is None:
+        return "no NSError"
+    text = f"{error.domain} {error.code}"
+    if error.under_domain is not None:
+        text += f" ({error.under_domain} {error.under_code})"
+    return text
+
+
+def refusal_code(error: NSErr | None) -> str:
+    if error is None or error.domain != COCOA:
+        return trash.FAILED
+    return _MOVE_CODES.get(error.code, trash.FAILED)
+
+
+def _within(path: str, folder: str) -> bool:
+    path = os.path.normcase(path)
+    folder = os.path.normcase(folder)
+    trimmed = folder.rstrip("\\/") or folder
+    return path == trimmed or path.startswith(trimmed.rstrip(os.sep) + os.sep)
+
+
+def _identity(path: str):
+    info = os.stat(path)
+    return info.st_dev, info.st_ino
+
+
+def inside_any_root(path: str, roots) -> bool:
+    """Whether *path* (existing or not) lies in an output root, by file identity.
+
+    Every existing folder from *path* up to the top is compared with the
+    roots' ``(st_dev, st_ino)``, so case, NFC/NFD and symlinked spellings of a
+    root all match.
+    """
+    identities = set()
+    for root in roots:
+        try:
+            identities.add(_identity(str(root)))
+        except OSError:
+            continue
+    current = os.path.abspath(path)
+    while True:
+        try:
+            if _identity(current) in identities:
+                return True
+        except OSError:
+            pass
+        parent = os.path.dirname(current)
+        if parent == current:
+            return False
+        current = parent
+
+
+def _same_folder(left: str, right: str) -> bool:
+    try:
+        return _identity(left) == _identity(right)
+    except OSError:
+        return os.path.realpath(left) == os.path.realpath(right)
+
+
+class MacTrashBackend:
+    """``available``/``move`` for macOS; the bridge and file checks are seams."""
+
+    def __init__(
+        self, bridge=None, uid=None, home=None, lstat=os.lstat, access=os.access
+    ):
+        self._bridge = bridge
+        self._load_failed = False
+        self._lock = threading.Lock()
+        if uid is None:
+            uid = os.getuid() if hasattr(os, "getuid") else -1
+        self._uid = uid
+        self._home = home
+        self._lstat = lstat
+        self._access = access
+
+    def _get_bridge(self):
+        if self._bridge is None and not self._load_failed:
+            with self._lock:
+                if self._bridge is None and not self._load_failed:
+                    try:
+                        self._bridge = load_bridge()
+                    except (OSError, AttributeError, ValueError) as error:
+                        self._load_failed = True
+                        logger.warning("The macOS trash is unavailable: %s", error)
+        if self._bridge is None:
+            raise trash.TrashRefused(trash.NO_TRASH, "Foundation bridge not loaded")
+        return self._bridge
+
+    def _home_dir(self) -> str:
+        return os.path.realpath(self._home or os.path.expanduser("~"))
+
+    def _plan(self, path: str, roots):
+        """``(trash folder, facts)`` for *path*; raises TrashRefused (design §7.2)."""
+        bridge = self._get_bridge()
+        real = os.path.realpath(path)
+        facts = bridge.facts(real)
+        if facts.local is not True:
+            raise trash.TrashRefused(trash.NETWORK)
+        if facts.read_only is not False:
+            raise trash.TrashRefused(trash.NO_TRASH, "read-only volume")
+        home = self._home_dir()
+        if facts.ubiquitous is True or any(
+            _within(real, os.path.join(home, *parts)) for parts in _CLOUD_FOLDERS
+        ):
+            raise trash.TrashRefused(trash.CLOUD_SYNC)
+        folder, error = bridge.trash_dir(real)
+        if folder is None:
+            folder = self._predict(facts, error)
+        if inside_any_root(folder, roots):
+            raise trash.TrashRefused(trash.TRASH_IN_ROOT)
+        self._check_permission(real)
+        return folder, facts
+
+    def _predict(self, facts: Facts, error: NSErr | None) -> str:
+        """``<volume>/.Trashes/<uid>`` for a volume whose trash is not made yet."""
+        missing = (
+            error is not None
+            and error.domain == COCOA
+            and error.code == FEATURE_UNSUPPORTED
+            and error.under_domain == POSIX
+            and error.under_code == _ENOENT
+        )
+        if not missing or facts.root_fs is not False or not facts.volume:
+            raise trash.TrashRefused(trash.NO_TRASH, error_detail(error))
+        trashes = os.path.join(facts.volume, ".Trashes")
+        try:
+            info = self._lstat(trashes)
+        except FileNotFoundError:
+            info = None
+        except OSError as failure:
+            raise trash.TrashRefused(trash.NO_TRASH, str(failure)) from failure
+        if info is not None and not stat.S_ISDIR(info.st_mode):
+            raise trash.TrashRefused(trash.NO_TRASH, ".Trashes is not a folder")
+        return os.path.join(trashes, str(self._uid))
+
+    def _check_permission(self, real: str) -> None:
+        parent = os.path.dirname(real)
+        if not self._access(parent, os.W_OK | os.X_OK):
+            raise trash.TrashRefused(trash.NO_PERMISSION, "folder not writable")
+        try:
+            file_info = self._lstat(real)
+            parent_info = self._lstat(parent)
+        except OSError as failure:
+            raise trash.TrashRefused(trash.FAILED, str(failure)) from failure
+        for info in (file_info, parent_info):
+            if getattr(info, "st_flags", 0) & _LOCK_FLAGS:
+                raise trash.TrashRefused(trash.NO_PERMISSION, "immutable flag")
+        if parent_info.st_mode & stat.S_ISVTX and file_info.st_uid != self._uid:
+            raise trash.TrashRefused(trash.NO_PERMISSION, "sticky folder, foreign file")
+
+    def _label(self, folder: str, facts: Facts) -> str:
+        if _same_folder(folder, os.path.join(self._home_dir(), ".Trash")):
+            return trash.HOME_TRASH_LABEL
+        real = os.path.realpath(folder)
+        volume = os.path.realpath(facts.volume) if facts.volume else None
+        if volume and volume != os.path.realpath(os.sep) and _within(real, volume):
+            relative = os.path.relpath(real, volume).replace(os.sep, "/")
+            return f"{relative} on its drive"
+        return real
+
+    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)
+        except trash.TrashRefused as refused:
+            return refused.code
+        except OSError:
+            return trash.NO_TRASH
+        return None
+
+    def move(self, path: str, roots) -> str:
+        """Move *path* into its trash folder and return the label (design §7.4)."""
+        try:
+            planned, facts = self._plan(path, roots)
+        except OSError as failure:
+            raise trash.TrashRefused(trash.NO_TRASH, str(failure)) from failure
+        real = os.path.realpath(path)
+        ok, resulting, error = self._get_bridge().trash(real)
+        if not ok:
+            raise trash.TrashRefused(refusal_code(error), error_detail(error))
+        if os.path.lexists(real):
+            raise trash.TrashRefused(trash.FAILED, "reported moved, still in place")
+        if not resulting or inside_any_root(os.path.dirname(resulting), roots):
+            raise trash.TrashUnconfirmed(f"{real} -> {resulting!r}")
+        folder = os.path.dirname(resulting)
+        if not _same_folder(folder, planned):
+            logger.warning(
+                "macOS trashed %s to %s, not the planned %s", real, resulting, planned
+            )
+        return self._label(folder, facts)
+
+
+# ── the Foundation bridge ────────────────────────────────────────────────
+
+_NS_TRASH_DIRECTORY = 102
+_NS_USER_DOMAIN_MASK = 1
+_LIBOBJC = "/usr/lib/libobjc.A.dylib"
+_FOUNDATION = "/System/Library/Frameworks/Foundation.framework/Foundation"
+_KEYS = (
+    "NSURLVolumeIsLocalKey",
+    "NSURLVolumeIsReadOnlyKey",
+    "NSURLVolumeIsRootFileSystemKey",
+    "NSURLVolumeURLKey",
+    "NSURLIsUbiquitousItemKey",
+    "NSUnderlyingErrorKey",
+)
+_REQUIRED_SELECTORS = (
+    ("NSURL", "fileURLWithFileSystemRepresentation:isDirectory:relativeToURL:"),
+    ("NSURL", "getResourceValue:forKey:error:"),
+    ("NSFileManager", "URLForDirectory:inDomain:appropriateForURL:create:error:"),
+    ("NSFileManager", "trashItemAtURL:resultingItemURL:error:"),
+)
+
+
+class _Nil(Exception):
+    """An Objective-C object was nil where a send needs one."""
+
+
+class Bridge:  # pragma: no cover - macOS only; the native-trash job runs it
+    """Foundation through ctypes: ``facts``, ``trash_dir`` and ``trash`` (design §6)."""
+
+    def __init__(self):
+        if sys.platform != "darwin":
+            raise OSError("the Foundation bridge exists only on macOS")
+        objc = ctypes.CDLL(_LIBOBJC)
+        foundation = ctypes.CDLL(_FOUNDATION)
+        void_p, char_p, byte = ctypes.c_void_p, ctypes.c_char_p, ctypes.c_byte
+        objc.objc_getClass.argtypes = [char_p]
+        objc.objc_getClass.restype = void_p
+        objc.sel_registerName.argtypes = [char_p]
+        objc.sel_registerName.restype = void_p
+        objc.objc_autoreleasePoolPush.argtypes = []
+        objc.objc_autoreleasePoolPush.restype = void_p
+        objc.objc_autoreleasePoolPop.argtypes = [void_p]
+        objc.objc_autoreleasePoolPop.restype = None
+        self._objc = objc
+        address = ctypes.cast(objc.objc_msgSend, void_p).value
+
+        def send(restype, *argtypes):
+            return ctypes.CFUNCTYPE(restype, void_p, void_p, *argtypes)(address)
+
+        out = ctypes.POINTER(void_p)
+        self._send_id = send(void_p)
+        self._send_responds = send(byte, void_p)
+        self._send_url = send(void_p, char_p, byte, void_p)
+        self._send_resource = send(byte, out, void_p, out)
+        self._send_bool = send(byte)
+        self._send_dir = send(void_p, ctypes.c_ulong, ctypes.c_ulong, void_p, byte, out)
+        self._send_trash = send(byte, void_p, out, out)
+        self._send_cstr = send(char_p)
+        self._send_long = send(ctypes.c_long)
+        self._send_object = send(void_p, void_p)
+        self._keys = {}
+        for name in _KEYS:
+            value = void_p.in_dll(foundation, name).value
+            if not value:
+                raise ValueError(f"Foundation constant {name} is nil")
+            self._keys[name] = value
+        for class_name, selector in _REQUIRED_SELECTORS:
+            if not self._send_responds(
+                self._class(class_name),
+                self._sel("respondsToSelector:"),
+                self._sel(selector),
+            ) and not self._instances_respond(class_name, selector):
+                raise ValueError(f"{class_name} does not respond to {selector}")
+
+    # helpers (call inside a pool) ------------------------------------------------
+
+    def _class(self, name: str):
+        value = self._objc.objc_getClass(name.encode())
+        if not value:
+            raise ValueError(f"no Objective-C class {name}")
+        return value
+
+    def _sel(self, name: str):
+        return self._objc.sel_registerName(name.encode())
+
+    def _instances_respond(self, class_name: str, selector: str) -> bool:
+        return (
+            self._send_responds(
+                self._class(class_name),
+                self._sel("instancesRespondToSelector:"),
+                self._sel(selector),
+            )
+            != 0
+        )
+
+    @staticmethod
+    def _need(value):
+        if not value:
+            raise _Nil()
+        return value
+
+    def _manager(self):
+        return self._need(
+            self._send_id(self._class("NSFileManager"), self._sel("defaultManager"))
+        )
+
+    def _url(self, path: str):
+        return self._need(
+            self._send_url(
+                self._class("NSURL"),
+                self._sel(
+                    "fileURLWithFileSystemRepresentation:isDirectory:relativeToURL:"
+                ),
+                os.fsencode(path),
+                0,
+                None,
+            )
+        )
+
+    def _path(self, url) -> str | None:
+        if not url:
+            return None
+        raw = self._send_cstr(url, self._sel("fileSystemRepresentation"))
+        return os.fsdecode(raw) if raw else None
+
+    def _text(self, obj) -> str | None:
+        if not obj:
+            return None
+        raw = self._send_cstr(obj, self._sel("UTF8String"))
+        return raw.decode("utf-8", "replace") if raw else None
+
+    def _error(self, error) -> NSErr | None:
+        if not error:
+            return None
+        domain = self._text(self._send_id(error, self._sel("domain"))) or "?"
+        code = self._send_long(error, self._sel("code"))
+        under_domain = under_code = None
+        info = self._send_id(error, self._sel("userInfo"))
+        if info:
+            under = self._send_object(
+                info, self._sel("objectForKey:"), self._keys["NSUnderlyingErrorKey"]
+            )
+            if under:
+                under_domain = self._text(self._send_id(under, self._sel("domain")))
+                under_code = self._send_long(under, self._sel("code"))
+        return NSErr(domain, code, under_domain, under_code)
+
+    def _resource(self, url, key_name: str):
+        value, error = ctypes.c_void_p(), ctypes.c_void_p()
+        ok = self._send_resource(
+            url,
+            self._sel("getResourceValue:forKey:error:"),
+            ctypes.byref(value),
+            self._keys[key_name],
+            ctypes.byref(error),
+        )
+        return value.value if ok != 0 and value.value else None
+
+    def _flag(self, url, key_name: str) -> bool | None:
+        value = self._resource(url, key_name)
+        if value is None:
+            return None
+        return self._send_bool(value, self._sel("boolValue")) != 0
+
+    def _pooled(self, work):
+        pool = self._objc.objc_autoreleasePoolPush()
+        try:
+            return work()
+        except _Nil as nil:
+            raise trash.TrashRefused(trash.NO_TRASH, "nil Objective-C object") from nil
+        finally:
+            self._objc.objc_autoreleasePoolPop(pool)
+
+    # the bridge API (design §6.3) ----------------------------------------------
+
+    def facts(self, path: str) -> Facts:
+        def work():
+            url = self._url(path)
+            return Facts(
+                local=self._flag(url, "NSURLVolumeIsLocalKey"),
+                read_only=self._flag(url, "NSURLVolumeIsReadOnlyKey"),
+                root_fs=self._flag(url, "NSURLVolumeIsRootFileSystemKey"),
+                volume=self._path(self._resource(url, "NSURLVolumeURLKey")),
+                ubiquitous=self._flag(url, "NSURLIsUbiquitousItemKey"),
+            )
+
+        return self._pooled(work)
+
+    def trash_dir(self, path: str):
+        def work():
+            error = ctypes.c_void_p()
+            folder = self._send_dir(
+                self._manager(),
+                self._sel("URLForDirectory:inDomain:appropriateForURL:create:error:"),
+                _NS_TRASH_DIRECTORY,
+                _NS_USER_DOMAIN_MASK,
+                self._url(path),
+                0,
+                ctypes.byref(error),
+            )
+            if folder:
+                return self._path(folder), None
+            return None, self._error(error.value)
+
+        return self._pooled(work)
+
+    def trash(self, path: str):
+        def work():
+            resulting, error = ctypes.c_void_p(), ctypes.c_void_p()
+            ok = self._send_trash(
+                self._manager(),
+                self._sel("trashItemAtURL:resultingItemURL:error:"),
+                self._url(path),
+                ctypes.byref(resulting),
+                ctypes.byref(error),
+            )
+            if ok != 0:
+                return True, self._path(resulting.value), None
+            return False, None, self._error(error.value)
+
+        return self._pooled(work)
+
+
+def load_bridge() -> Bridge:
+    """The real Foundation bridge; raises OSError/ValueError where it cannot load."""
+    return Bridge()

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 : ''});
     }
 

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.',

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

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;