test(trash): a trailing-space folder is hidden by name, not a real spelling (#543)

The Windows job showed that this spelling does not reach the file, unlike a
trailing dot, so the test's assumption was wrong. The gate already answered.

.github/workflows/portability.yml

@@ -29,6 +29,9 @@ env:
 #   native-trash (#499) has no green main runs yet: its probe, run 40949370943 (2026-10-29), took
 #     about 1 minute per macOS leg, so ceil(1.5 x 2) + 3 gives it 6 minutes (step 4). Recalibrate
 #     after three green runs.
+#   native-trash-linux (#543) has no green main runs yet: it installs the dev requirements and runs
+#     one test module, about 1 minute by the same estimate, so ceil(1.5 x 2) + 3 gives it 6 minutes.
+#     Recalibrate after three green runs.
 jobs:
   test:
     name: py${{ matrix.python-version }} on ${{ matrix.os }}
@@ -104,3 +107,39 @@ jobs:
         env:
           PM_NATIVE_TRASH: macos
         run: python -m unittest tests.test_trash_macos_native tests.test_trash_macos -v
+
+  # The same tests as the fake mount table, on a real ext4 mount used as the output root (#543). A
+  # runner can mount with sudo; the tests run as the normal user, so owner and mode checks are real.
+  native-trash-linux:
+    name: native trash on a real mount
+    runs-on: ubuntu-latest
+    timeout-minutes: 6
+
+    steps:
+      - uses: actions/checkout@v7
+
+      - name: Set up Python
+        uses: actions/setup-python@v7
+        with:
+          python-version: "3.12"
+          cache: pip
+          cache-dependency-path: requirements-dev.txt
+
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip
+          pip install -r requirements-dev.txt
+
+      - name: Mount an ext4 image
+        run: |
+          truncate -s 64M "$RUNNER_TEMP/drive.img"
+          mkfs.ext4 -q "$RUNNER_TEMP/drive.img"
+          mkdir "$RUNNER_TEMP/drive"
+          sudo mount -o loop "$RUNNER_TEMP/drive.img" "$RUNNER_TEMP/drive"
+          sudo rmdir "$RUNNER_TEMP/drive/lost+found"
+          sudo chown "$(id -u):$(id -g)" "$RUNNER_TEMP/drive"
+          findmnt "$RUNNER_TEMP/drive"
+
+      - name: Native trash proof
+        timeout-minutes: 4
+        run: PM_TRASH_MOUNT="$RUNNER_TEMP/drive" python -m unittest tests.test_trash_mount_root -v

TESTING.md

@@ -2489,7 +2489,25 @@ files to the system trash. **No code path unlinks or overwrites a media file.**
   (`_collect_media_files`), the legacy auto-tag listing (`_collect_scan_files`) and the
   delete review (`trash-folder`), and the image monitor, which tests the folders below the
   watched folder holding the new file (a source named like a trash folder is still watched). Tests: `tests/test_media_folders.py`,
-  `tests/test_image_removal_gaps.py`.
+  `tests/test_image_removal_gaps.py`. Media routes answer not found for a path in such a
+  folder, resolved first so a link or another spelling cannot pass (#543):
+  `_media_access_error` is the one gate (by-id, inspect, reveal, link, rescope as
+  `file-missing`, a prompt's media listing); `/images/serve` and the autotag target ask
+  `in_trash_below_roots` too, `media_target` refuses by name (annotations, collections,
+  whose capture store denies links and aliases), and previews are neither made nor served, the fallback past the preview gate's cap
+  included (`_quick_source`). A History output in one answers `missing`, checked at the
+  candidate and again at the open (a preview can be a link). The legacy duplicate delete
+  removes exactly the folder's real path plus the name (a link itself, never its target)
+  and refuses if either spelling is in a trash folder; `file_problem` refuses too. The delete
+  and Return flows never use this gate. Windows spellings (trailing dots and spaces, case,
+  8.3) run for real on the Windows job in `tests/test_trash_gate_routes.py`.
+- **An output folder that is a mount point** (#543): the mount's own `.Trash-$uid` and
+  `.Trash/$uid` are allowed only for a root that is exactly that mount, with the spec's
+  checks intact. A root above the mount, a mount nested under the root, a source that is or
+  sits in the trash folder, and a link in place of the trash folder stay refused. Tests:
+  `tests/test_trash_mount_root.py`, on a fake mount table in the suite and, with
+  `PM_TRASH_MOUNT=<real mount>`, in portability.yml's `native-trash-linux` job (loop-mounted
+  ext4). Locally, without root: `unshare -Urm` over a tmpfs.
 - **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

py/api/admin.py

@@ -36,10 +36,18 @@ except ImportError:
 
 try:
     from ...utils import exit_signal
-    from ...utils.media_folders import is_skipped_folder
+    from ...utils.media_folders import (
+        in_trash_below_roots,
+        in_trash_folder,
+        is_skipped_folder,
+    )
 except ImportError:
     from utils import exit_signal
-    from utils.media_folders import is_skipped_folder
+    from utils.media_folders import (
+        in_trash_below_roots,
+        in_trash_folder,
+        is_skipped_folder,
+    )
 
 try:
     from ...database.image_removal import mark_hides
@@ -1089,13 +1097,21 @@ class AdminRoutesMixin:
         return result
 
     def _delete_one_duplicate(self, image_path, output_path):
-        """Delete one file and its thumbnail; return a failure reason or None."""
+        """Delete one file and its thumbnail; return a failure reason or None.
+
+        What is removed is exactly the folder's real path plus the name: a
+        link is removed itself, never its target, and the preview removed is
+        that name's. Both the link's place and what it leads to are checked.
+        """
         file_path = Path(image_path)
         if not file_path.is_absolute():
             file_path = output_path / file_path
 
         try:
-            rel_path = file_path.resolve().relative_to(output_path.resolve())
+            real_root = output_path.resolve()
+            rel_path = file_path.resolve().relative_to(real_root)
+            exact = Path(os.path.realpath(file_path.parent)) / file_path.name
+            exact_rel = exact.relative_to(real_root)
         except ValueError:
             self.logger.warning(
                 f"Attempted to delete file outside output directory: {image_path}"
@@ -1104,15 +1120,19 @@ class AdminRoutesMixin:
 
         if file_path.suffix.lower() not in DELETABLE_MEDIA_EXTENSIONS:
             return "not a media file"
-        if "thumbnails" in rel_path.parts:
+        if "thumbnails" in rel_path.parts or "thumbnails" in exact_rel.parts:
             return "thumbnails are managed by the gallery"
+        if in_trash_below_roots(file_path, [output_path]) or in_trash_folder(
+            str(exact_rel)
+        ):
+            return "it is in a trash folder"  # never unlinked from a trash (#543)
 
         if not file_path.is_file():
             return "file not found"
 
-        os.remove(file_path)
+        os.remove(exact)
         self.logger.info(f"Deleted duplicate image: {image_path}")
-        _remove_thumbnail(output_path.resolve(), rel_path, self.logger)
+        _remove_thumbnail(real_root, exact_rel, self.logger)
         return None
 
     async def get_statistics(self, request):

py/api/autotag_routes.py

@@ -24,9 +24,9 @@ from . import thumbnail_settings
 from .. import model_gate
 
 try:
-    from ...utils.media_folders import is_skipped_folder
+    from ...utils.media_folders import in_trash_below_roots, is_skipped_folder
 except ImportError:
-    from utils.media_folders import is_skipped_folder
+    from utils.media_folders import in_trash_below_roots, is_skipped_folder
 
 from aiohttp import web
 
@@ -849,9 +849,12 @@ class AutotagRoutesMixin:
         """Return the real path when it is a media file inside an output directory."""
         if Path(str(raw_path)).suffix.lower() not in _MEDIA_EXTENSIONS:
             return None
-        for output_dir in self._get_all_output_dirs():
+        roots = list(self._get_all_output_dirs())
+        for output_dir in roots:
             if path_is_within(raw_path, output_dir):
                 real = os.path.realpath(raw_path)
+                if in_trash_below_roots(real, roots):
+                    return None  # thrown away, not an output (#543)
                 return real if os.path.isfile(real) else None
         return None
 

py/api/duplicate_review.py

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

py/api/history_current.py

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

py/api/images.py

@@ -66,7 +66,12 @@ try:
     from ...database.models import normalize_image_path
     from ...database.run_attribution import recorded_output_lookup
     from ...utils.card_seed import card_seed, header_params, parsed_positive
-    from ...utils.media_folders import in_skipped_folder, is_skipped_folder
+    from ...utils.media_folders import (
+        in_skipped_folder,
+        in_trash_below_roots,
+        in_trash_folder,
+        is_skipped_folder,
+    )
     from ...utils.parallel import map_parallel
     from ...utils.json_strict import dumps_strict
     from ...utils.inspection_response import (
@@ -92,7 +97,12 @@ except ImportError:
     from database.models import normalize_image_path
     from database.run_attribution import recorded_output_lookup
     from utils.card_seed import card_seed, header_params, parsed_positive
-    from utils.media_folders import in_skipped_folder, is_skipped_folder
+    from utils.media_folders import (
+        in_skipped_folder,
+        in_trash_below_roots,
+        in_trash_folder,
+        is_skipped_folder,
+    )
     from utils.parallel import map_parallel
     from utils.json_strict import dumps_strict
     from utils.inspection_response import (
@@ -277,10 +287,20 @@ def _forbidden(message="Access denied"):
     return web.json_response({"success": False, "error": message}, status=403)
 
 
-def _media_access_error(path, allowed_dirs):
-    """Return a 403 response when *path* must not be served, else None.
+def _file_not_found():
+    return web.json_response(
+        {"success": False, "error": "Image file not found"},
+        status=404,
+        headers={"Cache-Control": "no-store"},
+    )
 
-    Fails closed: an empty *allowed_dirs* denies everything.
+
+def _media_access_error(path, allowed_dirs):
+    """The response refusing *path*, or None when it may be read.
+
+    Fails closed: an empty *allowed_dirs* denies everything (403). A file in a
+    trash folder below its root is thrown away, not an output: a 404, as if it
+    were not there (#543), also through a link or another spelling.
     """
     if Path(path).suffix.lower() not in SERVABLE_EXTENSIONS:
         return _forbidden("Only media files can be served")
@@ -288,6 +308,8 @@ def _media_access_error(path, allowed_dirs):
         return _forbidden("No allowed image directories configured")
     if not any(_is_within(path, root) for root in allowed_dirs):
         return _forbidden()
+    if in_trash_below_roots(path, allowed_dirs):
+        return _file_not_found()
     return None
 
 
@@ -979,6 +1001,8 @@ def _settled_entry(root, relative):
         return None
     if in_skipped_folder(folders) or len(folders) > MAX_SCAN_DEPTH:
         return None
+    if in_trash_below_roots(path, [root]):
+        return None  # a linked folder on the way leads into a trash folder
     folder = os.fspath(root)
     for part in ("", *folders):
         folder = os.path.join(folder, part) if part else folder
@@ -2761,6 +2785,9 @@ class ImageRoutesMixin:
             if _has_traversal(filepath):
                 return _forbidden()
             suffix = Path(filepath).suffix.lower()
+            if in_trash_folder(filepath):
+                # Before any file access, so no preview is made for it (#543).
+                return _file_not_found()
             if suffix not in SERVABLE_EXTENSIONS and not (
                 # A JPEG XL preview is served from thumbnails/ only (#317).
                 suffix in thumbnail_settings.THUMBNAIL_EXTENSIONS
@@ -2784,6 +2811,8 @@ class ImageRoutesMixin:
                 # a fall-through to the next root.
                 if not _is_within(image_path, output_path):
                     return _forbidden()
+                if in_trash_below_roots(image_path, output_dirs):
+                    return _file_not_found()
                 if image_path.is_file():
                     return self._file_response(image_path)
 
@@ -2835,8 +2864,9 @@ class ImageRoutesMixin:
         if _is_thumbnail_cache_path(filepath):
             return _forbidden()
         image_path = root / filepath
-        if not _is_within(image_path, root):
-            return _forbidden()
+        denied = _media_access_error(image_path, [root])
+        if denied is not None:
+            return denied
         if not image_path.is_file():
             return web.json_response(
                 {"success": False, "error": "Image file not found"}, status=404

py/api/media_target.py

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

py/api/source_rescope.py

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

py/api/thumbnail_auto.py

@@ -67,6 +67,7 @@ from aiohttp import web
 
 from . import thumbnail_settings
 from .images import (
+    in_trash_below_roots,
     MEDIA_EXTENSIONS,
     THUMBNAIL_CLEAR_TITLE,
     VIDEO_EXTENSIONS,
@@ -224,6 +225,8 @@ def _quick_source(root, parts):
         for name in (base + extension, base + extension.upper()):
             path = folder / name
             if _regular_file(path) and _is_within(path, root):
+                if in_trash_below_roots(path, [root]):
+                    return None  # also the fallback past the gate's cap (#543)
                 return path, extension in VIDEO_EXTENSIONS
     return None
 
@@ -248,6 +251,8 @@ def _thumbnail_source(root, parts):
         found = _scan_folder(root.joinpath(*folder_parts), base, set(wanted))
         if found is None or not _is_within(found, root):
             return None
+    if in_trash_below_roots(found, [root]):
+        return None
     rel = Path(*folder_parts, found.name)
     dst = root / media_thumbnail_rel(rel)
     if dst != _current_preview(root, parts):
@@ -835,7 +840,7 @@ class AutoThumbnailMixin:
             if not folded.startswith(base) or not _is_within(path, root):
                 continue
             rel = Path(os.path.relpath(path, root))
-            if "thumbnails" in rel.parts[:-1]:
+            if "thumbnails" in rel.parts[:-1] or in_trash_below_roots(path, [root]):
                 return None
             dst = root / media_thumbnail_rel(rel)
             if not _is_within(root / "thumbnails", root) or not _is_within(

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

@@ -169,6 +169,59 @@ test('without a chosen mode nothing can start a delete', async ({ page }) => {
     expect(applies).toEqual([]);
 });
 
+// #543 (D190): a review where nothing can move shows a primary that clearly cannot act: no danger colors, ignored presses.
+test('a trash review with nothing to move has a clearly disabled primary', async ({ page }) => {
+    const before = await openImages(page);
+    await check(page, 0);
+    await page.route('**/images/delete/preview', async route => {
+        const response = await route.fetch();
+        const plan = await response.json();
+        const reason = "This drive's trash is inside an output folder.";
+        await route.fulfill({ response, json: {
+            ...plan,
+            trash: { count: 0 },
+            trash_refused: { count: 1, counts: { 'trash-in-root': 1 }, rows: [{ name: 'a.png', code: 'trash-in-root', detail: reason }] },
+        } });
+    });
+    const applies = [];
+    page.on('request', request => { if (request.url().includes('/maintenance/apply')) applies.push(request.url()); });
+    await bar(page).getByRole('button', { name: 'Delete…', exact: true }).click();
+    const dialog = review(page);
+    await expect(dialog).toBeVisible();
+    await dialog.getByRole('radio', { name: /^Also move the files to the trash/ }).check();
+    const primary = dialog.locator('.dialog-actions button.danger');
+    await expect(primary).toHaveText('Move 0 images to the trash');
+    await expect(primary).toHaveAttribute('aria-disabled', 'true');
+    await expect(primary).toHaveCSS('cursor', 'not-allowed');
+    const look = element => { const style = getComputedStyle(element); return [style.color, style.borderTopColor]; };
+    const cancel = dialog.getByRole('button', { name: 'Cancel', exact: true });
+    const [color, border] = await primary.evaluate(look);
+    const [cancelColor] = await cancel.evaluate(look);
+    const red = await page.evaluate(() => {
+        const probe = document.createElement('i');
+        probe.style.color = 'var(--red)';
+        document.querySelector('.ca-tools').append(probe);
+        const value = getComputedStyle(probe).color;
+        probe.remove();
+        return value;
+    });
+    expect(color).not.toBe(red);
+    expect(border).not.toBe(red);
+    expect(color).not.toBe(cancelColor);
+    await primary.hover({ force: true });
+    await expect(primary).toHaveCSS('background-color', 'rgba(0, 0, 0, 0)');
+    await primary.click({ force: true });
+    await primary.focus();
+    await page.keyboard.press('Enter');
+    await expect(dialog).toBeVisible();
+    expect(applies).toEqual([]);
+    await dialog.getByRole('radio', { name: /^Remove from the library only/ }).check();
+    await expect(primary).not.toHaveAttribute('aria-disabled', 'true');
+    await expect(primary).toHaveCSS('color', red);
+    await cancel.click();
+    await expect(page.locator(CARDS)).toHaveCount(before);
+});
+
 test('a chosen mode starts exactly one delete, with that mode, however often it is pressed', async ({ page }) => {
     const before = await openImages(page);
     await check(page, 0);

tests/test_api_history_outputs.py

@@ -26,6 +26,7 @@ from tests.test_recorded_saver_anchor import LLM_TEXT, text_generate_run
 from utils.execution_contexts import ContextBundle
 from utils.execution_history_events import make_finish, make_start
 from utils.execution_outputs import OutputBundle, OutputReference, snapshot_output_roots
+from py.api.images import media_thumbnail_rel
 from py.config import GalleryConfig
 
 
@@ -212,6 +213,57 @@ class HistoryOutputAPITests(ImageAPITestCase):
                 self.assertNotIn("root=", candidate["media"][field])
                 self.assertNotIn(str(self.comfy_dir), candidate["media"][field])
 
+    async def test_an_output_in_a_trash_folder_is_read_by_no_kind(self):
+        # #543: inspection, media, thumbnail and the listing share one check.
+        path = make_png(self.output_dir / "image.png")
+        start = self.capture([self.ref()])
+        media = await self.media(start)
+        trashed = self.output_dir / ".Trash-1000" / "files"
+        trashed.mkdir(parents=True)
+        os.rename(path, trashed / "image.png")
+        try:
+            os.symlink(trashed / "image.png", path)
+        except OSError:
+            if os.environ.get("GITHUB_ACTIONS"):
+                raise
+            self.skipTest("this machine cannot make symlinks")
+        for field in ("inspection_url", "url", "thumbnail_url"):
+            with self.subTest(field=field):
+                response = await self.current(media[field])
+                self.assertIn(response.status, (404, 409, 410))
+                await response.read()
+        response, page = await self.outputs(start)
+        self.assertEqual(response.status, 200, page)
+        candidate = page["items"][0]["current_candidate"]
+        self.assertEqual(candidate["status"], "unavailable")
+        self.assertEqual(candidate["reason"], "missing")
+
+    async def test_a_preview_linked_into_a_trash_folder_does_not_open(self):
+        # #543: the cover picks its preview after the candidate check.
+        make_png(self.output_dir / "image.png")
+        trashed = self.output_dir / ".Trash-1000" / "files"
+        trashed.mkdir(parents=True)
+        make_png(trashed / "secret.png", color=(0, 0, 255))
+        preview = self.output_dir / media_thumbnail_rel(Path("image.png"))
+        preview.parent.mkdir(parents=True, exist_ok=True)
+        try:
+            os.symlink(trashed / "secret.png", preview)
+        except OSError:
+            if os.environ.get("GITHUB_ACTIONS"):
+                raise
+            self.skipTest("this machine cannot make symlinks")
+        media = await self.media(self.capture([self.ref()]))
+        response = await self.current(media["thumbnail_url"])
+        body = await response.read()
+        # Never the trashed bytes: the link is ignored or refused at the open.
+        self.assertNotEqual(body, (trashed / "secret.png").read_bytes())
+        mod = importlib.import_module("py.api.history_current")
+        scope = mod.snapshot_current_output_scope(self.api)
+        with self.assertRaises(mod.CurrentLocationUnavailable):
+            mod._open_current_target(
+                mod._canonical(preview), self.output_dir, scope, True
+            )
+
     async def test_complete_zero_incomplete_unavailable_and_not_captured_are_distinct(
         self,
     ):

tests/test_trash_freedesktop.py

@@ -114,11 +114,12 @@ class FreedesktopTrashTests(unittest.TestCase):
         self.assertEqual(os.listdir(elsewhere), [])
 
     def test_trash_inside_an_output_root_is_refused(self):
-        # The drive's top folder is itself an output root: its .Trash-uid
-        # would be listed as images.
+        # A root above the drive's top folder would hold the drive's .Trash-uid.
+        # (A root that is exactly the top folder is allowed: #543,
+        # tests/test_trash_mount_root.py.)
         path = self._file(self.root)
         with self.assertRaises(trash.TrashRefused) as caught:
-            self.backend.move(path, [self.drive])
+            self.backend.move(path, [self.tmp])
         self.assertEqual(caught.exception.code, trash.TRASH_IN_ROOT)
         self.assertTrue(os.path.exists(path))
         self.assertFalse(
@@ -147,9 +148,8 @@ class FreedesktopTrashTests(unittest.TestCase):
     def test_available_reports_without_writing(self):
         path = self._file(self.root)
         self.assertIsNone(self.backend.available(path, [self.root]))
-        self.assertEqual(
-            self.backend.available(path, [self.drive]), trash.TRASH_IN_ROOT
-        )
+        self.assertEqual(self.backend.available(path, [self.tmp]), trash.TRASH_IN_ROOT)
+        self.assertIsNone(self.backend.available(path, [self.drive]))
         self.assertEqual(os.listdir(self.drive), ["output"])
 
     # ── names and the no-replace rename ───────────────────────────────

tests/test_trash_gate_routes.py

@@ -0,0 +1,372 @@
+"""No route reads a file in a trash folder below its root (#543).
+
+Each route is tried with the trash folder named outright, through a linked file
+and through a linked folder, and with a file it must still read as the control.
+The Windows spellings run only on Windows, and they do run there: outside a
+developer's machine a missing short name or symlink right is a failure.
+"""
+
+import ctypes
+import os
+import unittest
+import urllib.parse
+from pathlib import Path
+from unittest.mock import AsyncMock, patch
+
+try:
+    from tests.trash_route_audit import (
+        FILE_READER_CALLS,
+        KNOWN_FILE_READERS,
+        ROUTE_AUDIT,
+    )
+    from tests.test_api_admin import _write_png
+    from tests.test_image_delete_routes import ImageDeleteTestCase
+except ImportError:  # discovered with tests/ as the top-level directory
+    from trash_route_audit import FILE_READER_CALLS, KNOWN_FILE_READERS, ROUTE_AUDIT
+    from test_api_admin import _write_png
+    from test_image_delete_routes import ImageDeleteTestCase
+
+from py.api.duplicate_review import OUTSIDE_SOURCE, file_problem
+from py.api.images import media_thumbnail_rel
+from py.api.thumbnail_auto import OVER, _quick_source, _thumbnail_source
+from utils.media_folders import in_trash_below_roots, in_trash_folder, is_trash_folder
+
+ON_CI = bool(os.environ.get("GITHUB_ACTIONS"))
+NAMED = (".Trash-1000/files/x.png", "$RECYCLE.BIN/x.png", "sub/.Trashes/501/x.png")
+LINKED = ("link.png", "linkdir/x.png")
+
+
+def _short_name(path):
+    """The 8.3 spelling of *path* (Windows only)."""
+    buffer = ctypes.create_unicode_buffer(1024)
+    ctypes.windll.kernel32.GetShortPathNameW(str(path), buffer, 1024)
+    return buffer.value
+
+
+class _GateCase(ImageDeleteTestCase):
+    async def get_application(self):
+        app = await super().get_application()
+        self.root = self.output_dir
+        _write_png(self.root / "kept.png", prompt_text="kept")
+        for relative in NAMED:
+            (self.root / relative).parent.mkdir(parents=True, exist_ok=True)
+            _write_png(self.root / relative, prompt_text="thrown away")
+        return app
+
+    def links(self):
+        """Links into the trash folder, or a skip where they cannot be made."""
+        inside = self.root / ".Trash-1000" / "files"
+        try:
+            os.symlink(inside / "x.png", self.root / "link.png")
+            os.symlink(inside, self.root / "linkdir", target_is_directory=True)
+        except OSError:
+            if ON_CI:
+                raise
+            self.skipTest("this machine cannot make symlinks")
+
+    def query(self):
+        scope = self.api._gallery_root_scope(self.api._get_all_output_dirs()[0], 0)
+        return {"root": "0", "root_scope": scope}
+
+    def body(self, relative):
+        return self.target(relative)
+
+    async def status(self, method, url, **kwargs):
+        response = await self.client.request(method, url, **kwargs)
+        await response.read()
+        return response.status
+
+    async def serve(self, relative):
+        url = "/prompt_manager/images/serve/" + urllib.parse.quote(relative)
+        return await self.status("GET", url, params=self.query())
+
+    async def inspect(self, relative):
+        url = "/prompt_manager/images/inspect/" + urllib.parse.quote(relative)
+        return await self.status("GET", url, params=self.query())
+
+    async def rescope(self, relative):
+        response = await self.client.post(
+            "/prompt_manager/images/rescope", json={"target": self.body(relative)}
+        )
+        return response.status, await response.json()
+
+    async def autotag(self, relative):
+        """The target the legacy single/apply bodies pick, by path."""
+        target, error = await self.api._select_autotag_target(
+            {"root_index": 0, "path": relative}
+        )
+        return target, error
+
+    async def assert_hidden_everywhere(self, relative):
+        self.assertEqual(await self.serve(relative), 404)
+        self.assertEqual(await self.inspect(relative), 404)
+        self.assertIsNone(self.api._reveal_path(relative, self.root))
+        status, body = await self.rescope(relative)
+        self.assertEqual((status, body.get("code")), (404, "file-missing"))
+        target, error = await self.autotag(relative)
+        self.assertIsNone(target)
+        self.assertEqual(error.status, 403)
+        answer = await self.client.post(
+            "/prompt_manager/images/link",
+            json={"prompt_id": self.save("p"), "image_path": str(self.root / relative)},
+        )
+        self.assertEqual(answer.status, 404)
+
+
+class TestTheGate(_GateCase):
+    async def test_the_control_file_still_reads_on_every_route(self):
+        self.assertEqual(await self.serve("kept.png"), 200)
+        self.assertEqual(await self.inspect("kept.png"), 200)
+        self.assertIsNotNone(self.api._reveal_path("kept.png", self.root))
+        self.assertEqual((await self.rescope("kept.png"))[0], 200)
+        target, _error = await self.autotag("kept.png")
+        self.assertIsNotNone(target)
+        answer = await self.client.post(
+            "/prompt_manager/images/link",
+            json={
+                "prompt_id": self.save("p"),
+                "image_path": str(self.root / "kept.png"),
+            },
+        )
+        self.assertEqual(answer.status, 200)
+        observed = await self.client.post(
+            "/prompt_manager/image-annotations/observe",
+            json={"target": self.body("kept.png")},
+        )
+        self.assertEqual(observed.status, 200)
+        added = await self.client.post(
+            "/prompt_manager/collections/items",
+            json={"name": "c", "items": [self.body("kept.png")]},
+        )
+        self.assertLess(added.status, 300)
+
+    async def test_a_named_trash_folder_is_hidden_on_every_route(self):
+        for relative in NAMED:
+            with self.subTest(relative=relative):
+                await self.assert_hidden_everywhere(relative)
+
+    async def test_a_linked_file_or_folder_is_hidden_on_every_route(self):
+        self.links()
+        for relative in LINKED:
+            with self.subTest(relative=relative):
+                await self.assert_hidden_everywhere(relative)
+
+    async def test_serving_by_id_autotag_by_id_and_a_prompts_media(self):
+        self.links()
+        roots = self.api._get_all_output_dirs()
+        for relative in (*NAMED, *LINKED):
+            with self.subTest(relative=relative):
+                image_id = self.api.db.link_image_to_prompt(
+                    str(self.save(f"linked {relative}")), str(self.root / relative)
+                )
+                url = f"/prompt_manager/images/{image_id}/file"
+                self.assertEqual(await self.status("GET", url), 404)
+                target, error = await self.api._select_autotag_target(
+                    {"image_id": image_id}
+                )
+                self.assertIsNone(target)
+                self.assertEqual(error.status, 403)
+                records = [{"image_path": str(self.root / relative)}]
+                self.assertEqual(self.api._library_media(records, roots), [])
+        kept = [{"image_path": str(self.root / "kept.png")}]
+        self.assertEqual(len(self.api._library_media(kept, roots)), 1)
+
+    async def test_annotations_refuse_a_trash_path_and_a_link(self):
+        self.links()
+        url = "/prompt_manager/image-annotations/observe"
+        for relative in NAMED:
+            with self.subTest(relative=relative):
+                body = {"target": self.body(relative)}
+                self.assertEqual(await self.status("POST", url, json=body), 404)
+        for relative in LINKED:
+            with self.subTest(relative=relative):
+                body = {"target": self.body(relative)}
+                status = await self.status("POST", url, json=body)
+                self.assertGreaterEqual(status, 400)
+                self.assertLess(status, 500)
+
+    async def test_collections_refuse_a_trash_path(self):
+        for relative in NAMED:
+            with self.subTest(relative=relative):
+                answer = await self.client.post(
+                    "/prompt_manager/collections/items",
+                    json={"name": "c", "items": [self.body(relative)]},
+                )
+                self.assertEqual(answer.status, 404)
+
+    async def test_the_exact_path_listing_does_not_name_it(self):
+        self.links()
+        for relative in (*NAMED, *LINKED):
+            with self.subTest(relative=relative):
+                response = await self.client.get(
+                    "/prompt_manager/images/output",
+                    params={"mode": "workspace", "path": relative},
+                )
+                page = await response.json()
+                self.assertEqual(page.get("images", []), [])
+
+    async def test_no_preview_is_made_through_a_linked_folder(self):
+        self.links()
+        preview = Path(media_thumbnail_rel(Path("linkdir/x.png"))).parts
+        self.assertIsNone(_thumbnail_source(self.root, preview))
+        kept = Path(media_thumbnail_rel(Path("kept.png"))).parts
+        self.assertIsNotNone(_thumbnail_source(self.root, kept))
+
+    async def test_the_preview_fallback_past_the_cap_is_gated_too(self):
+        # The gate's OVER branch looks the source up by name and serves it.
+        self.links()
+        gate = self.api._preview_gate()
+        with patch.object(gate, "run", AsyncMock(return_value=OVER)):
+            linked = media_thumbnail_rel(Path("linkdir/x.png"))
+            self.assertEqual(await self.serve(linked), 404)
+            kept = media_thumbnail_rel(Path("kept.png"))
+            self.assertEqual(await self.serve(kept), 200)
+        self.assertIsNone(
+            _quick_source(
+                self.root, Path(media_thumbnail_rel(Path("linkdir/x.png"))).parts
+            )
+        )
+
+    async def test_the_legacy_duplicate_delete_never_unlinks_from_a_trash(self):
+        self.links()
+        for relative in (*NAMED, *LINKED):
+            with self.subTest(relative=relative):
+                reason = self.api._delete_one_duplicate(relative, self.root)
+                self.assertEqual(reason, "it is in a trash folder")
+                self.assertTrue((self.root / relative).exists())
+
+    async def test_a_link_inside_a_trash_folder_is_never_removed(self):
+        self.links()
+        link = self.root / ".Trash-1000" / "files" / "back.png"
+        os.symlink(self.root / "kept.png", link)
+        reason = self.api._delete_one_duplicate(".Trash-1000/files/back.png", self.root)
+        self.assertEqual(reason, "it is in a trash folder")
+        self.assertTrue(link.is_symlink())
+        self.assertTrue((self.root / "kept.png").exists())
+
+    async def test_the_legacy_delete_removes_the_link_itself_and_its_own_preview(self):
+        self.links()
+        os.symlink(self.root / "kept.png", self.root / "alias.png")
+        thumbs = {}
+        for name in ("kept.png", "alias.png"):
+            thumbs[name] = self.root / media_thumbnail_rel(Path(name))
+            thumbs[name].parent.mkdir(parents=True, exist_ok=True)
+            thumbs[name].write_bytes(b"preview")
+        self.assertIsNone(self.api._delete_one_duplicate("alias.png", self.root))
+        self.assertFalse((self.root / "alias.png").is_symlink())
+        self.assertTrue((self.root / "kept.png").is_file())
+        self.assertFalse(thumbs["alias.png"].exists())
+        self.assertTrue(thumbs["kept.png"].exists())
+
+    async def test_a_linked_folder_out_of_the_root_removes_nothing_there(self):
+        self.links()
+        outside = self.outside_dir / "elsewhere"
+        outside.mkdir(parents=True)
+        os.symlink(self.root / "kept.png", outside / "back.png")
+        os.symlink(outside, self.root / "out", target_is_directory=True)
+        reason = self.api._delete_one_duplicate("out/back.png", self.root)
+        self.assertEqual(reason, "outside output directory")
+        self.assertTrue((outside / "back.png").is_symlink())
+
+    async def test_a_duplicate_removal_review_refuses_a_trash_folder(self):
+        path = self.root / ".Trash-1000" / "files" / "x.png"
+        problem = file_problem(
+            path, self.root, None, path.stat().st_size, hash_check=False
+        )
+        self.assertEqual(problem, OUTSIDE_SOURCE)
+
+    async def test_a_source_named_like_a_trash_folder_still_reads(self):
+        source = self.root / ".Trash-77"
+        source.mkdir()
+        _write_png(source / "mine.png", prompt_text="mine")
+        self.api._get_all_output_dirs = lambda: [source]
+        self.assertEqual(await self.serve("mine.png"), 200)
+        self.assertEqual(await self.inspect("mine.png"), 200)
+        self.assertEqual((await self.rescope("mine.png"))[0], 200)
+        self.assertIsNotNone((await self.autotag("mine.png"))[0])
+
+    async def test_delete_still_moves_a_listed_file(self):
+        await self.review(["kept.png"])
+        _plan, job = await self.delete(["kept.png"], "trash")
+        self.assertEqual(job.detail()["counters"]["trashed"], 1)
+        self.assertFalse((self.root / "kept.png").exists())
+
+
+class TestPredicates(unittest.TestCase):
+    def test_trailing_dots_and_spaces_are_ignored(self):
+        for name in ("$RECYCLE.BIN.", "$recycle.bin ", ".Trash-1000. ", ".Trashes..."):
+            self.assertTrue(is_trash_folder(name), name)
+        for name in ("trash.", "...", " ", ".Trash-x."):
+            self.assertFalse(is_trash_folder(name), name)
+
+    def test_either_separator(self):
+        self.assertTrue(in_trash_folder("a\\.Trash\\b.png"))
+        self.assertFalse(in_trash_folder(".Trash"))
+
+    def test_a_path_no_root_visibly_holds_fails_closed(self):
+        self.assertTrue(in_trash_below_roots("/x/.Trash-1/files/a.png", ["/other"]))
+        self.assertFalse(in_trash_below_roots("/x/a.png", ["/other"]))
+
+
+@unittest.skipUnless(os.name == "nt", "Windows spellings of one folder")
+class TestWindowsSpellings(_GateCase):
+    def spellings(self):
+        spelled = [
+            "$RECYCLE.BIN./x.png",
+            "$recycle.bin/x.png",
+            ".TRASH-1000/files/x.png",
+            ".Trash-1000./files/x.png",
+        ]
+        short = _short_name(self.root / ".Trash-1000")
+        if os.path.basename(short).casefold() == ".trash-1000":
+            if ON_CI:
+                self.fail("the job's work disk gave no 8.3 short name")
+            self.skipTest("this disk makes no 8.3 short names")
+        spelled.append(os.path.basename(short) + "/files/x.png")
+        return spelled
+
+    async def test_every_spelling_reaches_the_file_and_is_hidden(self):
+        for relative in self.spellings():
+            with self.subTest(relative=relative):
+                self.assertTrue((self.root / relative).is_file(), "spelling is real")
+                await self.assert_hidden_everywhere(relative)
+
+    async def test_a_folder_spelled_with_a_trailing_space_is_hidden_by_name(self):
+        # Measured on the Windows job (run 41123443497): unlike a trailing dot,
+        # this spelling does not reach the file through a longer path. The name
+        # check still answers for it, whatever a volume does.
+        await self.assert_hidden_everywhere("$RECYCLE.BIN /x.png")
+
+    async def test_a_root_spelled_in_another_case_does_not_fail_open(self):
+        inside = self.root / ".Trash-1000" / "files" / "x.png"
+        self.assertTrue(in_trash_below_roots(inside, [str(self.root).upper()]))
+        kept = self.root / "kept.png"
+        self.assertFalse(in_trash_below_roots(kept, [str(self.root).upper()]))
+
+
+class TestEveryRouteIsClassified(_GateCase):
+    async def test_a_new_route_must_say_how_it_treats_media_files(self):
+        registered = {
+            f"{route.method} {route.resource.canonical}"
+            for route in self.app.router.routes()
+            if route.method != "HEAD"
+            and route.resource is not None
+            and route.resource.canonical.startswith("/prompt_manager")
+        }
+        audited = {route for routes in ROUTE_AUDIT.values() for route in routes}
+        self.assertEqual(
+            registered ^ audited,
+            set(),
+            "add the route to tests/trash_route_audit.py with its media check",
+        )
+
+    def test_a_file_reader_in_the_api_layer_is_a_known_one(self):
+        # A new way to open a media file must be looked at for the same gate.
+        api = Path(__file__).resolve().parent.parent / "py" / "api"
+        found = set()
+        for module in api.glob("*.py"):
+            text = module.read_text(encoding="utf-8")
+            found.update(
+                f"{module.name}:{call}" for call in FILE_READER_CALLS if call in text
+            )
+        self.assertEqual(found ^ KNOWN_FILE_READERS, set())

tests/test_trash_mount_root.py

@@ -0,0 +1,516 @@
+"""An output folder that is itself a mount point (#543).
+
+The freedesktop per-drive trash is then inside the output root. Only the mount's
+own ``.Trash-$uid`` and ``.Trash/$uid`` are allowed, only for a root that is
+exactly the mount root; every walker and media route ignores them.
+
+Every class runs on a disposable tree with a fake mount table. With
+``PM_TRASH_MOUNT=<empty folder that is a real mount>`` the same tests run on that
+real mount (``.github/workflows/portability.yml``, ``native-trash-linux``, or
+locally ``unshare -Urm`` over a tmpfs); that folder is the only real place they
+touch, and the fake-table-only cases skip.
+"""
+
+import asyncio
+import errno
+import os
+import shutil
+import stat
+import sys
+import tempfile
+import unittest
+import urllib.parse
+from pathlib import Path
+from unittest.mock import patch
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+try:
+    from tests.disk_waits import DISK_BOUND_SECONDS
+    from tests.test_api_admin import _write_png
+    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_api_admin import _write_png
+    from test_image_delete_routes import ImageDeleteTestCase
+
+from py.api import images as images_module  # noqa: E402
+from py.api.admin import _collect_media_files  # noqa: E402
+from py.api.autotag_routes import AutotagRoutesMixin  # noqa: E402
+from utils import trash  # noqa: E402
+from utils.image_monitor import ImageGenerationHandler  # noqa: E402
+
+POSIX = os.name == "posix" and sys.platform != "darwin"
+REAL_MOUNT = os.environ.get("PM_TRASH_MOUNT")
+UID = os.getuid() if hasattr(os, "getuid") else -1
+
+
+def _empty(folder):
+    """Remove what a test left inside *folder*, never the folder itself."""
+    for name in os.listdir(folder):
+        path = os.path.join(folder, name)
+        if os.path.isdir(path) and not os.path.islink(path):
+            os.chmod(path, 0o700)
+            shutil.rmtree(path)
+        else:
+            os.unlink(path)
+
+
+def restore_as_a_file_manager_does(top, trash_dir, name):
+    """Put back one trashed file from its ``.trashinfo``, as the spec's readers do."""
+    info_path = os.path.join(trash_dir, "info", name + ".trashinfo")
+    with open(info_path, encoding="utf-8") as handle:
+        lines = handle.read().splitlines()
+    recorded = next(line[5:] for line in lines if line.startswith("Path="))
+    relative = os.fsdecode(urllib.parse.unquote_to_bytes(recorded))
+    # A top-directory trash records a path relative to the mount (spec 1.0).
+    assert not os.path.isabs(relative), recorded
+    destination = os.path.join(top, relative)
+    os.makedirs(os.path.dirname(destination), exist_ok=True)
+    os.rename(os.path.join(trash_dir, "files", name), destination)
+    os.unlink(info_path)
+    return destination
+
+
+@unittest.skipUnless(POSIX, "freedesktop trash is the Linux/BSD backend")
+class MountRootBackendTests(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.mkdtemp(prefix="pm-trash-root-")
+        self.addCleanup(shutil.rmtree, self.tmp, True)
+        self.data_home = os.path.join(self.tmp, "xdg")
+        os.makedirs(self.data_home)
+        if REAL_MOUNT:
+            self.assertTrue(os.path.ismount(REAL_MOUNT), "PM_TRASH_MOUNT is no mount")
+            self.drive = os.path.realpath(REAL_MOUNT)
+            self.addCleanup(_empty, self.drive)
+            self.backend = trash.FreedesktopBackend(data_home=self.data_home)
+        else:
+            self.drive = os.path.join(self.tmp, "drive")
+            os.makedirs(self.drive)
+            self.backend = self._fake_table()
+        self.dash = os.path.join(self.drive, f".Trash-{UID}")
+
+    def _fake_table(self, *extra, uid=None):
+        mounts = ["/", os.path.realpath(self.drive), *extra]
+        return trash.FreedesktopBackend(
+            data_home=self.data_home, mounts=lambda: list(mounts), uid=uid
+        )
+
+    def _file(self, relative="a.png", content=b"png-bytes"):
+        path = os.path.join(self.drive, relative)
+        os.makedirs(os.path.dirname(path), exist_ok=True)
+        with open(path, "wb") as handle:
+            handle.write(content)
+        return path
+
+    def _assert_refused(self, path, roots, code):
+        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.assertTrue(os.path.lexists(path))
+
+    def _sticky_shared(self):
+        shared = os.path.join(self.drive, ".Trash")
+        os.mkdir(shared)
+        os.chmod(shared, 0o1777)
+        return shared
+
+    # ── the two allowed folders ───────────────────────────────────────
+
+    def test_a_mount_root_output_uses_its_dash_uid_trash(self):
+        path = self._file()
+        identity = os.lstat(path)
+        self.assertIsNone(self.backend.available(path, [self.drive]))
+        self.assertEqual(os.listdir(self.drive), ["a.png"], "available wrote nothing")
+        label = self.backend.move(path, [self.drive])
+        self.assertEqual(label, f".Trash-{UID} on its drive")
+        self.assertFalse(os.path.lexists(path))
+        moved = os.lstat(os.path.join(self.dash, "files", "a.png"))
+        self.assertEqual(
+            (moved.st_dev, moved.st_ino), (identity.st_dev, identity.st_ino)
+        )
+        self.assertEqual(stat.S_IMODE(os.lstat(self.dash).st_mode), 0o700)
+
+    def test_a_mount_root_output_uses_a_sticky_shared_trash_first(self):
+        shared = self._sticky_shared()
+        path = self._file()
+        label = self.backend.move(path, [self.drive])
+        self.assertEqual(label, f"{os.path.join('.Trash', str(UID))} on its drive")
+        self.assertTrue(
+            os.path.exists(os.path.join(shared, str(UID), "files", "a.png"))
+        )
+        self.assertFalse(os.path.lexists(self.dash))
+
+    def test_a_nested_output_root_beside_the_mount_root_is_fine(self):
+        path = self._file("sub/a.png")
+        roots = [self.drive, os.path.join(self.drive, "sub")]
+        self.assertIsNone(self.backend.available(path, roots))
+        self.backend.move(path, roots)
+        self.assertTrue(os.path.exists(os.path.join(self.dash, "files", "a.png")))
+
+    # ── the spec's checks stay ────────────────────────────────────────
+
+    def test_a_shared_trash_without_the_sticky_bit_is_passed_over(self):
+        shared = os.path.join(self.drive, ".Trash")
+        os.mkdir(shared, 0o777)
+        os.chmod(shared, 0o777)
+        path = self._file()
+        self.backend.move(path, [self.drive])
+        self.assertEqual(os.listdir(shared), [])
+        self.assertTrue(os.path.exists(os.path.join(self.dash, "files", "a.png")))
+
+    def test_a_shared_trash_that_is_a_link_is_passed_over(self):
+        real = os.path.join(self.tmp, "sticky")
+        os.mkdir(real)
+        os.chmod(real, 0o1777)
+        os.symlink(real, os.path.join(self.drive, ".Trash"))
+        path = self._file()
+        self.backend.move(path, [self.drive])
+        self.assertEqual(os.listdir(real), [])
+        self.assertTrue(os.path.exists(os.path.join(self.dash, "files", "a.png")))
+
+    def test_a_dash_uid_link_into_the_root_is_refused(self):
+        os.mkdir(os.path.join(self.drive, "images"))
+        os.symlink(os.path.join(self.drive, "images"), self.dash)
+        path = self._file("images/a.png")
+        self._assert_refused(path, [self.drive], trash.TRASH_IN_ROOT)
+        self.assertEqual(os.listdir(os.path.join(self.drive, "images")), ["a.png"])
+
+    def test_a_dash_uid_link_out_of_the_root_is_refused(self):
+        elsewhere = os.path.join(self.tmp, "elsewhere")
+        os.mkdir(elsewhere)
+        os.symlink(elsewhere, self.dash)
+        path = self._file()
+        self._assert_refused(path, [self.drive], trash.NO_TRASH)
+        self.assertEqual(os.listdir(elsewhere), [])
+
+    def test_a_uid_link_in_the_shared_trash_is_never_followed(self):
+        shared = self._sticky_shared()
+        os.mkdir(os.path.join(self.drive, "images"))
+        os.symlink(os.path.join(self.drive, "images"), os.path.join(shared, str(UID)))
+        path = self._file("images/a.png")
+        self._assert_refused(path, [self.drive], trash.TRASH_IN_ROOT)
+
+    def test_a_files_folder_that_is_a_link_is_refused(self):
+        os.mkdir(self.dash, 0o700)
+        elsewhere = os.path.join(self.tmp, "elsewhere")
+        os.mkdir(elsewhere)
+        os.symlink(elsewhere, os.path.join(self.dash, "files"))
+        path = self._file()
+        with self.assertRaises(trash.TrashRefused) as caught:
+            self.backend.move(path, [self.drive])
+        self.assertEqual(caught.exception.code, trash.NO_TRASH)
+        self.assertTrue(os.path.exists(path))
+        self.assertEqual(os.listdir(elsewhere), [])
+
+    @unittest.skipIf(REAL_MOUNT, "needs the fake mount table")
+    def test_a_trash_folder_owned_by_someone_else_is_refused(self):
+        other = UID + 1
+        os.mkdir(os.path.join(self.drive, f".Trash-{other}"), 0o700)
+        backend = self._fake_table(uid=other)
+        path = self._file()
+        self.assertEqual(backend.available(path, [self.drive]), trash.NO_TRASH)
+        self.assertTrue(os.path.exists(path))
+
+    def test_a_rename_across_drives_is_refused_and_nothing_is_copied(self):
+        def cross(*_args):
+            raise OSError(errno.EXDEV, "cross-device")
+
+        backend = trash.FreedesktopBackend(
+            data_home=self.data_home,
+            mounts=lambda: ["/", os.path.realpath(self.drive)],
+            rename=cross,
+        )
+        path = self._file()
+        with self.assertRaises(trash.TrashRefused) as caught:
+            backend.move(path, [self.drive])
+        self.assertEqual(caught.exception.code, trash.NO_TRASH)
+        self.assertTrue(os.path.exists(path))
+        self.assertEqual(os.listdir(os.path.join(self.dash, "files")), [])
+        self.assertEqual(os.listdir(os.path.join(self.dash, "info")), [])
+
+    # ── any other trash inside a root stays refused ───────────────────
+
+    def test_a_root_above_the_mount_is_refused(self):
+        path = self._file()
+        above = os.path.dirname(self.drive)
+        self._assert_refused(path, [above], trash.TRASH_IN_ROOT)
+        self._assert_refused(path, [self.drive, above], trash.TRASH_IN_ROOT)
+        self.assertFalse(os.path.lexists(self.dash))
+
+    def test_a_source_named_like_the_trash_folder_is_refused(self):
+        source = self.dash
+        path = self._file(f".Trash-{UID}/a.png")
+        self._assert_refused(path, [source], trash.TRASH_IN_ROOT)
+        self._assert_refused(path, [self.drive, source], trash.TRASH_IN_ROOT)
+
+    def test_a_source_inside_the_trash_folder_is_refused(self):
+        source = os.path.join(self.dash, "files")
+        os.makedirs(source)
+        path = self._file()
+        self._assert_refused(path, [self.drive, source], trash.TRASH_IN_ROOT)
+        self.assertEqual(os.listdir(source), [])
+
+    def test_a_source_that_is_the_shared_trash_is_refused(self):
+        shared = self._sticky_shared()
+        path = self._file()
+        self._assert_refused(path, [self.drive, shared], trash.TRASH_IN_ROOT)
+
+    @unittest.skipIf(REAL_MOUNT, "needs the fake mount table")
+    def test_a_mount_nested_under_the_root_is_refused(self):
+        # The file's mount is not the root: its trash is "any other location".
+        inner = os.path.join(self.drive, "inner")
+        os.mkdir(inner)
+        backend = self._fake_table(os.path.realpath(inner))
+        path = self._file("inner/a.png")
+        self.assertEqual(backend.available(path, [self.drive]), trash.TRASH_IN_ROOT)
+        # ...and the same inner mount as its own root is fine.
+        self.assertIsNone(backend.available(path, [inner]))
+
+    @unittest.skipIf(REAL_MOUNT, "needs the fake mount table")
+    def test_the_home_trash_inside_a_mount_root_output_stays_refused(self):
+        backend = trash.FreedesktopBackend(
+            data_home=os.path.join(self.drive, "xdg"),
+            mounts=lambda: ["/", os.path.realpath(self.drive)],
+        )
+        path = self._file()
+        self.assertEqual(backend.available(path, [self.drive]), trash.TRASH_IN_ROOT)
+
+    # ── restoring ─────────────────────────────────────────────────────
+
+    def test_the_recorded_path_is_relative_to_the_mount(self):
+        path = self._file("sub dir/b c%é.png")
+        self.backend.move(path, [self.drive])
+        with open(
+            os.path.join(self.dash, "info", "b c%é.png.trashinfo"), encoding="utf-8"
+        ) as handle:
+            info = handle.read()
+        self.assertIn("Path=sub%20dir/b%20c%25%C3%A9.png\n", info)
+
+    def test_a_restored_file_returns_to_its_place_with_its_bytes(self):
+        path = self._file("sub/a.png", b"original")
+        identity = os.lstat(path).st_ino
+        self.backend.move(path, [self.drive])
+        back = restore_as_a_file_manager_does(self.drive, self.dash, "a.png")
+        self.assertEqual(back, path)
+        self.assertEqual(os.lstat(path).st_ino, identity)
+        with open(path, "rb") as handle:
+            self.assertEqual(handle.read(), b"original")
+        self.assertEqual(os.listdir(os.path.join(self.dash, "files")), [])
+
+
+@unittest.skipUnless(POSIX, "freedesktop trash is the Linux/BSD backend")
+class MountRootLayoutTests(ImageDeleteTestCase):
+    """The real backend over the routes: nothing lists the trash, restore relinks."""
+
+    async def get_application(self):
+        app = await super().get_application()
+        self.data_home = Path(self.tmpdir) / "xdg"
+        self.data_home.mkdir()
+        if REAL_MOUNT:
+            self.assertTrue(os.path.ismount(REAL_MOUNT), "PM_TRASH_MOUNT is no mount")
+            self.output_dir = Path(os.path.realpath(REAL_MOUNT))
+            self.addCleanup(_empty, str(self.output_dir))
+            self.api._get_all_output_dirs = lambda: [self.output_dir]
+            mounts = None
+        else:
+            drive = os.path.realpath(self.output_dir)
+            mounts = lambda: ["/", drive]  # noqa: E731
+        trash.BACKEND = trash.FreedesktopBackend(
+            data_home=str(self.data_home), mounts=mounts
+        )
+        self.trash_dir = self.output_dir / f".Trash-{UID}"
+        return app
+
+    async def _trashed_one(self):
+        """``kept.png`` stays; ``gone.png`` was moved to the drive's trash."""
+        _write_png(self.output_dir / "kept.png", prompt_text="a kept prompt")
+        _write_png(self.output_dir / "gone.png", prompt_text="a thrown prompt")
+        _plan, job = await self.delete(["gone.png"], "trash")
+        self.assertEqual(job.detail()["counters"]["trashed"], 1, job.detail())
+        self.assertTrue((self.trash_dir / "files" / "gone.png").is_file())
+        self.assertFalse((self.output_dir / "gone.png").exists())
+
+    async def _json(self, method, url, **kwargs):
+        response = await self.client.request(method, url, **kwargs)
+        self.assertLess(response.status, 500, await response.text())
+        return await response.json()
+
+    async def _scan(self, scope="all"):
+        answer = await self._json(
+            "POST", "/prompt_manager/jobs/scan", json={"scope": scope}
+        )
+        task = self.api._scan_job.task
+        await asyncio.wait_for(asyncio.shield(task), DISK_BOUND_SECONDS)
+        return answer
+
+    def _linked_paths(self):
+        return {
+            os.path.basename(row[0])
+            for row in self._rows("SELECT image_path FROM generated_images")
+        }
+
+    def _rows(self, sql):
+        import sqlite3
+
+        conn = sqlite3.connect(self.api.db.model.db_path)
+        try:
+            return conn.execute(sql).fetchall()
+        finally:
+            conn.close()
+
+    async def test_the_review_can_move_files_on_a_mount_root_output(self):
+        _write_png(self.output_dir / "a.png", prompt_text="p")
+        plan = await self.review(["a.png"])
+        self.assertEqual(plan["trash"], {"count": 1})
+        self.assertEqual(plan["trash_refused"]["count"], 0)
+        self.assertFalse(self.trash_dir.exists(), "the review wrote nothing")
+
+    async def test_refresh_listing_and_legacy_gallery_ignore_the_trash(self):
+        await self._trashed_one()
+        self.assertEqual(await self.listed(), ["kept.png"])
+        rescan = await self._json("POST", "/prompt_manager/gallery/rescan")
+        self.assertNotIn(".Trash", str(rescan))
+        legacy = await self._json("GET", "/prompt_manager/images/output")
+        self.assertNotIn(".Trash", str(legacy))
+        folders = await self._json("GET", "/prompt_manager/gallery/subfolders")
+        self.assertNotIn(".Trash", str(folders))
+
+    async def test_scan_folder_and_its_walkers_ignore_the_trash(self):
+        await self._trashed_one()
+        await self._scan()
+        self.assertEqual(self._linked_paths(), {"kept.png"})
+        collected = _collect_media_files([self.output_dir], [".png"])
+        self.assertEqual([Path(p).name for p in collected], ["kept.png"])
+        found, _truncated = AutotagRoutesMixin._collect_scan_files(self.output_dir)
+        self.assertEqual([p.name for p in found], ["kept.png"])
+
+    async def test_thumbnail_plans_ignore_the_trash(self):
+        await self._trashed_one()
+        thumbs = self.output_dir / "thumbnails"
+        for mode in ("missing", "rebuild"):
+            plan = self.api._plan_thumbnails(self.output_dir, thumbs, mode=mode)
+            sources = [src.name for src, _dst, _video in plan["targets"]]
+            self.assertEqual(sources, ["kept.png"])
+
+    async def test_the_autotag_listing_ignores_the_trash(self):
+        await self._trashed_one()
+        with patch.object(
+            self.api, "_find_comfyui_output_dir", return_value=str(self.output_dir)
+        ):
+            body = await self._json("GET", "/prompt_manager/scan_output_dir")
+        self.assertEqual([i["filename"] for i in body["images"]], ["kept.png"])
+
+    async def test_the_monitor_ignores_the_trash_with_nested_roots(self):
+        await self._trashed_one()
+        monitor = ImageGenerationHandler.__new__(ImageGenerationHandler)
+        monitor.supported_extensions = (".png",)
+        root = str(self.output_dir)
+        monitor.watched_roots = (root, os.path.join(root, "sub"))
+        inside = os.path.join(str(self.trash_dir), "files", "gone.png")
+        self.assertFalse(monitor.is_image_file(inside))
+        self.assertTrue(monitor.is_image_file(os.path.join(root, "new.png")))
+        self.assertTrue(monitor.is_image_file(os.path.join(root, "sub", "n.png")))
+
+    async def test_a_source_below_a_trash_named_folder_lists_on_its_own(self):
+        # The Codex P2 from #502: a source named like a trash folder stays a
+        # source, while the outer root does not list it.
+        source = self.output_dir / ".Trash-77"
+        source.mkdir()
+        _write_png(source / "mine.png", prompt_text="mine")
+        _write_png(self.output_dir / "kept.png", prompt_text="kept")
+        outer = [
+            Path(e.path).name for e in images_module._walk_media(self.output_dir, 100)
+        ]
+        self.assertEqual(outer, ["kept.png"])
+        inner = [Path(e.path).name for e in images_module._walk_media(source, 100)]
+        self.assertEqual(inner, ["mine.png"])
+        monitor = ImageGenerationHandler.__new__(ImageGenerationHandler)
+        monitor.supported_extensions = (".png",)
+        monitor.watched_roots = (str(self.output_dir), str(source))
+        self.assertTrue(monitor.is_image_file(str(source / "new.png")))
+        self.assertFalse(
+            monitor.is_image_file(str(self.output_dir / ".Trash-1000" / "x.png"))
+        )
+
+    async def test_the_media_routes_do_not_serve_a_trashed_file(self):
+        await self._trashed_one()
+        relative = f".Trash-{UID}/files/gone.png"
+        roots = self.api._get_all_output_dirs()
+        scope = self.api._gallery_root_scope(roots[0], 0)
+        query = {"root": "0", "root_scope": scope}
+        served = await self.client.get(
+            f"/prompt_manager/images/serve/{relative}", params=query
+        )
+        self.assertEqual(served.status, 404)
+        preview = await self.client.get(
+            f"/prompt_manager/images/serve/thumbnails/{relative}", params=query
+        )
+        self.assertEqual(preview.status, 404)
+        self.assertFalse((self.output_dir / "thumbnails" / f".Trash-{UID}").exists())
+        inspected = await self.client.get(
+            f"/prompt_manager/images/inspect/{relative}", params=query
+        )
+        self.assertEqual(inspected.status, 404)
+        self.assertIsNone(self.api._reveal_path(relative, roots[0]))
+        self.assertIsNotNone(self.api._reveal_path("kept.png", roots[0]))
+
+    async def test_a_record_pointing_into_the_trash_is_not_served(self):
+        await self._trashed_one()
+        inside = self.trash_dir / "files" / "gone.png"
+        image_id = self.api.db.link_image_to_prompt(
+            str(self.save("linked by hand")), str(inside)
+        )
+        response = await self.client.get(f"/prompt_manager/images/{image_id}/file")
+        self.assertEqual(response.status, 404)
+
+    async def test_a_trashed_file_cannot_be_linked_or_listed_under_a_prompt(self):
+        await self._trashed_one()
+        inside = self.trash_dir / "files" / "gone.png"
+        prompt_id = self.save("linked by hand")
+        refused = await self.client.post(
+            "/prompt_manager/images/link",
+            json={"prompt_id": prompt_id, "image_path": str(inside)},
+        )
+        self.assertEqual(refused.status, 404)
+        self.assertEqual(self._linked_paths(), set())
+        # A record from before this check still lists nothing.
+        self.api.db.link_image_to_prompt(str(prompt_id), str(inside))
+        page = self.api._library_media(
+            self._rows_as_records(), self.api._get_all_output_dirs()
+        )
+        self.assertEqual(page, [])
+
+    def _rows_as_records(self):
+        rows = self._rows("SELECT image_path FROM generated_images")
+        return [{"image_path": row[0]} for row in rows]
+
+    async def test_a_source_named_like_a_trash_folder_is_still_served(self):
+        source = self.output_dir / ".Trash-77"
+        source.mkdir()
+        _write_png(source / "mine.png", prompt_text="mine")
+        self.api._get_all_output_dirs = lambda: [source]
+        scope = self.api._gallery_root_scope(source, 0)
+        served = await self.client.get(
+            "/prompt_manager/images/serve/mine.png",
+            params={"root": "0", "root_scope": scope},
+        )
+        self.assertEqual(served.status, 200)
+
+    async def test_a_restored_file_lists_and_links_as_before(self):
+        await self._trashed_one()
+        await self._scan()
+        self.assertEqual(self._linked_paths(), {"kept.png"})
+        back = restore_as_a_file_manager_does(
+            str(self.output_dir), str(self.trash_dir), "gone.png"
+        )
+        self.assertEqual(back, str(self.output_dir / "gone.png"))
+        self.assertEqual(sorted(await self.listed()), ["gone.png", "kept.png"])
+        await self._scan("new")
+        self.assertEqual(self._linked_paths(), {"gone.png", "kept.png"})
+        prompts = self._rows(
+            "SELECT p.text FROM generated_images g JOIN prompts p ON p.id = g.prompt_id"
+            " WHERE g.image_path LIKE '%gone.png'"
+        )
+        self.assertEqual(prompts, [("a thrown prompt",)])

tests/trash_route_audit.py

@@ -0,0 +1,211 @@
+"""Every /prompt_manager route and how it treats a file in a trash folder (#543).
+
+``tests/test_trash_gate_routes.py`` fails when a route is missing here, so a new
+route has to choose a group. ``FILE_READER_CALLS`` is the same idea for code that
+opens files in ``py/api``.
+"""
+
+FILE_READER_CALLS = ("FileResponse(", "Image.open(", "os.open(", "read_bytes(")
+
+KNOWN_FILE_READERS = frozenset(
+    {
+        "history_current.py:os.open(",
+        "images.py:FileResponse(",
+        "images.py:Image.open(",
+        "images.py:os.open(",
+        "__init__.py:Image.open(",
+        "logging_routes.py:os.open(",
+        "thumbnail_auto.py:FileResponse(",
+        "thumbnail_encode.py:Image.open(",
+    }
+)
+
+ROUTE_AUDIT = {
+    "gate: _media_access_error, resolved": (
+        "GET /prompt_manager/images/{image_id}/file",
+        "GET /prompt_manager/images/inspect/{image_path}",
+        "POST /prompt_manager/images/reveal",
+        "POST /prompt_manager/images/link",
+        "POST /prompt_manager/images/rescope",
+        "GET /prompt_manager/library/{positive_id}/images",
+        "GET /prompt_manager/library/{positive_id}",
+        "GET /prompt_manager/library",
+    ),
+    "in_trash_below_roots, resolved": (
+        "GET /prompt_manager/images/serve/{filepath}",
+        "POST /prompt_manager/autotag/single",
+        "POST /prompt_manager/autotag/apply",
+        "GET /prompt_manager/history/runs/{run_id}/outputs",
+        "GET /prompt_manager/history/runs/{run_id}/output/current/media",
+        "GET /prompt_manager/history/runs/{run_id}/output/current/thumbnail",
+        "GET /prompt_manager/history/runs/{run_id}/output/current/inspection",
+        "POST /prompt_manager/delete_duplicate_images",
+    ),
+    "media_target by name, then the capture alias check or listed files": (
+        "POST /prompt_manager/image-annotations/observe",
+        "POST /prompt_manager/image-annotations/suggest",
+        "PUT /prompt_manager/image-annotations/tags",
+        "GET /prompt_manager/image-annotations/capture/{token}",
+        "DELETE /prompt_manager/image-annotations/capture/{token}",
+        "POST /prompt_manager/jobs/image-tags",
+        "POST /prompt_manager/collections/items",
+        "DELETE /prompt_manager/collections/items",
+        "GET /prompt_manager/collections/membership",
+        "PUT /prompt_manager/images/category",
+        "DELETE /prompt_manager/images/category",
+    ),
+    "walker: utils/media_folders (listed files only)": (
+        "GET /prompt_manager/images/output",
+        "POST /prompt_manager/gallery/rescan",
+        "GET /prompt_manager/gallery/subfolders",
+        "PUT /prompt_manager/images/favorite",
+        "DELETE /prompt_manager/images/favorite",
+        "GET /prompt_manager/images/categories",
+        "POST /prompt_manager/images/delete/preview",
+        "POST /prompt_manager/images/removed/return",
+        "POST /prompt_manager/scan",
+        "POST /prompt_manager/jobs/scan",
+        "GET /prompt_manager/scan_duplicates",
+        "POST /prompt_manager/scan_duplicates/stream",
+        "POST /prompt_manager/jobs/duplicates",
+        "GET /prompt_manager/maintenance/duplicates/{job_id}",
+        "GET /prompt_manager/scan_output_dir",
+        "POST /prompt_manager/autotag/start",
+        "POST /prompt_manager/autotag/batch/preview",
+        "POST /prompt_manager/jobs/autotag",
+        "POST /prompt_manager/jobs/thumbnails",
+        "GET /prompt_manager/images/thumbnails/plan",
+        "GET /prompt_manager/images/thumbnails/status",
+        "POST /prompt_manager/images/generate-thumbnails",
+        "POST /prompt_manager/images/generate-thumbnails/progress",
+        "GET /prompt_manager/diagnostics",
+    ),
+    "record text only, reads no file; its media is served through the gate": (
+        "GET /prompt_manager/images/all",
+        "GET /prompt_manager/images/recent",
+        "GET /prompt_manager/images/search",
+        "GET /prompt_manager/images/prompt/{image_path}",
+        "GET /prompt_manager/prompts/{prompt_id}/images",
+        "DELETE /prompt_manager/images/{image_id}",
+        "GET /prompt_manager/recent",
+        "GET /prompt_manager/search",
+        "GET /prompt_manager/export",
+        "GET /prompt_manager/library/export",
+        "GET /prompt_manager/history/runs/{run_id}",
+        "GET /prompt_manager/review/items",
+        "GET /prompt_manager/review/items/{item_id}",
+        "POST /prompt_manager/review/items/{item_id}/decision",
+        "POST /prompt_manager/maintenance/preview",
+        "POST /prompt_manager/maintenance/apply",
+    ),
+    "no media file": (
+        "DELETE /prompt_manager/collections/{collection_id}",
+        "DELETE /prompt_manager/delete/{prompt_id}",
+        "DELETE /prompt_manager/pairs/{pair_id}",
+        "DELETE /prompt_manager/prompts/{prompt_id}/tags",
+        "DELETE /prompt_manager/restore/preview/{token}",
+        "DELETE /prompt_manager/saved-views/{view_id}",
+        "DELETE /prompt_manager/tags/{tag_name}",
+        "GET /prompt_manager/admin",
+        "GET /prompt_manager/autotag/models",
+        "GET /prompt_manager/backup",
+        "GET /prompt_manager/backup/status",
+        "GET /prompt_manager/categories",
+        "GET /prompt_manager/collections",
+        "GET /prompt_manager/gallery",
+        "GET /prompt_manager/gallery.html",
+        "GET /prompt_manager/jobs",
+        "GET /prompt_manager/jobs/{job_id}",
+        "GET /prompt_manager/jobs/{job_id}/events",
+        "GET /prompt_manager/jobs/{job_id}/report",
+        "GET /prompt_manager/js/{filepath}",
+        "GET /prompt_manager/lib/{filepath}",
+        "GET /prompt_manager/library/options",
+        "GET /prompt_manager/library/{positive_id}/history",
+        "GET /prompt_manager/library/{positive_id}/history/pairs/{pair_id}/newest",
+        "GET /prompt_manager/logs",
+        "GET /prompt_manager/logs/config",
+        "GET /prompt_manager/logs/download/{filename}",
+        "GET /prompt_manager/logs/files",
+        "GET /prompt_manager/logs/stats",
+        "GET /prompt_manager/lora/detect",
+        "GET /prompt_manager/lora/status",
+        "GET /prompt_manager/lora/trigger-words",
+        "GET /prompt_manager/pairs",
+        "GET /prompt_manager/pairs/export",
+        "GET /prompt_manager/pairs/{pair_id}",
+        "GET /prompt_manager/review/events",
+        "GET /prompt_manager/saved-views",
+        "GET /prompt_manager/scan/status",
+        "GET /prompt_manager/scan_duplicates/status",
+        "GET /prompt_manager/settings",
+        "GET /prompt_manager/stats",
+        "GET /prompt_manager/subfolders",
+        "GET /prompt_manager/system/diagnostics",
+        "GET /prompt_manager/system/health",
+        "GET /prompt_manager/system/library",
+        "GET /prompt_manager/system/overview",
+        "GET /prompt_manager/system/statistics",
+        "GET /prompt_manager/tags",
+        "GET /prompt_manager/tags/filter",
+        "GET /prompt_manager/tags/stats",
+        "GET /prompt_manager/tags/suggest",
+        "GET /prompt_manager/tags/{tag_name}/prompts",
+        "GET /prompt_manager/test",
+        "GET /prompt_manager/vocabulary/categories",
+        "GET /prompt_manager/vocabulary/tags",
+        "GET /prompt_manager/vocabulary/tags/{tag_id}/impact",
+        "GET /prompt_manager/web",
+        "GET /prompt_manager/workspace",
+        "PATCH /prompt_manager/collections/{collection_id}",
+        "PATCH /prompt_manager/library/{positive_id}",
+        "PATCH /prompt_manager/pairs/{pair_id}",
+        "PATCH /prompt_manager/saved-views/{view_id}",
+        "PATCH /prompt_manager/vocabulary/tags/{tag_id}",
+        "POST /prompt_manager/autotag/download/{model_type}",
+        "POST /prompt_manager/autotag/unload",
+        "POST /prompt_manager/bulk/category",
+        "POST /prompt_manager/bulk/delete",
+        "POST /prompt_manager/bulk/tags",
+        "POST /prompt_manager/cleanup",
+        "POST /prompt_manager/diagnostics/test-link",
+        "POST /prompt_manager/images/clear-thumbnails",
+        "POST /prompt_manager/jobs/model-download",
+        "POST /prompt_manager/jobs/{job_id}/cancel",
+        "POST /prompt_manager/library",
+        "POST /prompt_manager/library/delete",
+        "POST /prompt_manager/library/delete/preview",
+        "POST /prompt_manager/library/restore",
+        "POST /prompt_manager/library/{positive_id}/variants",
+        "POST /prompt_manager/logs/config",
+        "POST /prompt_manager/logs/truncate",
+        "POST /prompt_manager/lora/enable",
+        "POST /prompt_manager/lora/import/apply",
+        "POST /prompt_manager/lora/import/preview",
+        "POST /prompt_manager/lora/refresh-cache",
+        "POST /prompt_manager/lora/scan",
+        "POST /prompt_manager/maintenance",
+        "POST /prompt_manager/pairs",
+        "POST /prompt_manager/prompts/tags",
+        "POST /prompt_manager/prompts/{prompt_id}/tags",
+        "POST /prompt_manager/restore",
+        "POST /prompt_manager/restore/apply",
+        "POST /prompt_manager/restore/preview",
+        "POST /prompt_manager/save",
+        "POST /prompt_manager/saved-views",
+        "POST /prompt_manager/scan/cancel",
+        "POST /prompt_manager/settings",
+        "POST /prompt_manager/settings/validate-source",
+        "POST /prompt_manager/system/health",
+        "POST /prompt_manager/system/library/keep",
+        "POST /prompt_manager/system/library/move",
+        "POST /prompt_manager/tags/merge",
+        "POST /prompt_manager/vocabulary/categories/apply",
+        "POST /prompt_manager/vocabulary/categories/preview",
+        "POST /prompt_manager/vocabulary/tags",
+        "POST /prompt_manager/vocabulary/tags/{tag_id}/delete",
+        "PUT /prompt_manager/prompts/{prompt_id}",
+        "PUT /prompt_manager/prompts/{prompt_id}/rating",
+        "PUT /prompt_manager/tags/{tag_name}",
+    ),
+}

utils/diagnostics.py

@@ -37,6 +37,7 @@ from pathlib import Path
 from typing import Dict, Any
 
 from .logging_config import get_logger
+from .media_folders import in_skipped_folder
 
 try:
     from ..database.workflow_documents import DocumentReader, resolve_image_row
@@ -298,7 +299,13 @@ class GalleryDiagnostics:
                 try:
                     image_files = []
                     for ext in [".png", ".jpg", ".jpeg", ".webp"]:
-                        image_files.extend(Path(abs_path).rglob(f"*{ext}"))
+                        image_files.extend(
+                            found
+                            for found in Path(abs_path).rglob(f"*{ext}")
+                            if not in_skipped_folder(
+                                found.relative_to(abs_path).parts[:-1]
+                            )
+                        )
                     self.logger.info(f"      [IMG]  Images found: {len(image_files)}")
                 except Exception as e:
                     self.logger.error(f"      [FAIL] Error scanning: {e}")

utils/media_folders.py

@@ -6,16 +6,22 @@ inside an output root (``D:\\$RECYCLE.BIN`` when ``D:\\`` is a root,
 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.
+
+``in_trash_below_roots`` is the check for a path someone names (#543): it
+resolves the path first, so a link or another spelling cannot get past it.
 """
 
+import os
+
 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()
+    ``.Trashes`` or ``$RECYCLE.BIN``. Windows ignores a name's trailing dots
+    and spaces, so ``$RECYCLE.BIN.`` is that folder; they are ignored here too."""
+    folded = name.rstrip(" .").casefold()
     if folded in TRASH_FOLDER_NAMES:
         return True
     return folded.startswith(".trash-") and folded[7:].isdigit()
@@ -26,6 +32,34 @@ def is_skipped_folder(name: str) -> bool:
     return name == PREVIEW_FOLDER or is_trash_folder(name)
 
 
+def in_trash_folder(relative: str) -> bool:
+    """Whether a root-relative file path lies in a trash folder (#543).
+
+    Takes either separator. Previews of it (``thumbnails/.Trash-1000/...``) count.
+    """
+    folders = relative.replace("\\", "/").split("/")[:-1]
+    return any(is_trash_folder(part) for part in folders)
+
+
+def in_trash_below_roots(path, roots) -> bool:
+    """Whether *path*, resolved, is in a trash folder below every root holding it.
+
+    A source named like a trash folder is still a source, so a file it holds
+    with no trash folder below it is not in one. Paths compare resolved and
+    normcased. A path no root visibly holds tests its whole path: a spelling
+    difference must not fail open.
+    """
+    real = os.path.normcase(os.path.realpath(path))
+    below = []
+    for root in roots:
+        real_root = os.path.normcase(os.path.realpath(root))
+        if real.startswith(os.path.join(real_root, "")):
+            below.append(os.path.relpath(real, real_root))
+    if not below:
+        return in_trash_folder(real)
+    return all(in_trash_folder(relative) for relative in below)
+
+
 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

@@ -6,7 +6,10 @@ Two rules hold for every backend:
   filesystem. Nothing is ever copied, unlinked or overwritten: a file that
   cannot be moved recoverably stays where it is, and ``TrashRefused`` says why.
 - **Never into a root.** A trash folder inside an output root would be listed
-  as images, so it is refused (``trash-in-root``).
+  as images, so it is refused (``trash-in-root``). The one exception (#543): for
+  an output root that is exactly a mount root, that mount's own ``.Trash-$uid``
+  and ``.Trash/$uid``, which every walker and media route ignores
+  (``utils.media_folders``).
 
 Linux and BSD use :class:`FreedesktopBackend`, the freedesktop.org Trash
 specification 1.0: the home trash (``$XDG_DATA_HOME/Trash``) for a file on the
@@ -184,10 +187,21 @@ def _canonical(path: str) -> str:
     return os.path.normcase(os.path.realpath(path))
 
 
-def _check_roots(trash_dir: str, roots) -> None:
+def _check_roots(trash_dir: str, roots, top: str | None = None) -> None:
+    """Refuse a trash that a root would list, or that holds a root.
+
+    *top* is the mount a per-drive *trash_dir* was built from. A root that is
+    exactly that mount may hold it, unless a link makes the trash really live
+    somewhere else (#543). Any other root holding it refuses, also beside one.
+    """
     trash_key = _canonical(trash_dir)
+    top_key = None if top is None else _canonical(top)
+    unlinked = trash_key == os.path.normcase(trash_dir)
     for root in roots:
-        if _inside(trash_key, _canonical(str(root))):
+        root_key = _canonical(str(root))
+        if _inside(root_key, trash_key):
+            raise TrashRefused(TRASH_IN_ROOT)
+        if _inside(trash_key, root_key) and not (unlinked and root_key == top_key):
             raise TrashRefused(TRASH_IN_ROOT)
 
 
@@ -267,7 +281,7 @@ class FreedesktopBackend:
             candidates.append(os.path.join(shared, str(self._uid)))
         candidates.append(os.path.join(top, f".Trash-{self._uid}"))
         for trash_dir in candidates:
-            _check_roots(trash_dir, roots)
+            _check_roots(trash_dir, roots, top)
             if self._usable(trash_dir, create=create):
                 label = f"{os.path.relpath(trash_dir, top)} on its drive"
                 return trash_dir, top, label

web/js/workspace/workspace.css

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