fix(trash): allow the mount's own trash when the output folder is a mount point (#543)

An output root that is exactly a mount root may hold that mount's .Trash-$uid
and .Trash/$uid, with the spec's checks intact. Any other trash inside a root
stays refused, and so does a root inside the trash. Media routes no longer
serve files in trash folders, and a review with nothing to move shows a
neutral disabled primary.

.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,16 @@ 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 (serve, inspect, reveal, serving by image id), `/images/link` refuses it and a
+  prompt's media listing leaves it out (#543).
+- **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/images.py

@@ -66,7 +66,11 @@ 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_folder,
+        is_skipped_folder,
+    )
     from ...utils.parallel import map_parallel
     from ...utils.json_strict import dumps_strict
     from ...utils.inspection_response import (
@@ -92,7 +96,11 @@ 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_folder,
+        is_skipped_folder,
+    )
     from utils.parallel import map_parallel
     from utils.json_strict import dumps_strict
     from utils.inspection_response import (
@@ -291,6 +299,21 @@ def _media_access_error(path, allowed_dirs):
     return None
 
 
+def _record_in_trash_folder(path, roots) -> bool:
+    """Whether a recorded *path* lies in a trash folder below every root holding it.
+
+    A source named like a trash folder is still a source, so a path it holds
+    without a trash folder below it stays servable (#543).
+    """
+    real = os.path.realpath(path)
+    below = []
+    for root in roots:
+        real_root = os.path.realpath(root)
+        if real.startswith(os.path.join(real_root, "")):
+            below.append(os.path.relpath(real, real_root))
+    return bool(below) and all(in_trash_folder(relative) for relative in below)
+
+
 # Media types the gallery scans and thumbnails (a superset is never served).
 GALLERY_IMAGE_EXTENSIONS = frozenset({".png", ".jpg", ".jpeg", ".webp", ".gif"})
 MEDIA_EXTENSIONS = GALLERY_IMAGE_EXTENSIONS | VIDEO_EXTENSIONS
@@ -2640,7 +2663,10 @@ class ImageRoutesMixin:
             if denied is not None:
                 return denied
 
-            if not image_path.is_file():
+            if (
+                _record_in_trash_folder(image_path, allowed_dirs)
+                or not image_path.is_file()
+            ):
                 return web.json_response(
                     {"success": False, "error": "Image file not found"}, status=404
                 )
@@ -2761,6 +2787,13 @@ class ImageRoutesMixin:
             if _has_traversal(filepath):
                 return _forbidden()
             suffix = Path(filepath).suffix.lower()
+            if in_trash_folder(filepath):
+                # Thrown away, not an output (#543): as if it were not there.
+                return web.json_response(
+                    {"success": False, "error": "Image file not found"},
+                    status=404,
+                    headers={"Cache-Control": "no-store"},
+                )
             if suffix not in SERVABLE_EXTENSIONS and not (
                 # A JPEG XL preview is served from thumbnails/ only (#317).
                 suffix in thumbnail_settings.THUMBNAIL_EXTENSIONS
@@ -3593,7 +3626,9 @@ class ImageRoutesMixin:
             if denied is not None:
                 return denied
 
-            if not os.path.isfile(image_path):
+            if _record_in_trash_folder(image_path, allowed_dirs) or not os.path.isfile(
+                image_path
+            ):
                 return web.json_response(
                     {"success": False, "error": "Image file not found"}, status=404
                 )
@@ -3673,6 +3708,8 @@ class ImageRoutesMixin:
     @staticmethod
     def _reveal_path(relative: str, root: Path) -> Path | None:
         """The real path of a regular media file under *root*, else None."""
+        if in_trash_folder(relative):
+            return None
         resolved = ImageRoutesMixin._resolve_client_media_path(relative, [root])
         if resolved is None or _media_access_error(resolved, [root]) is not None:
             return None
@@ -3907,6 +3944,10 @@ class ImageRoutesMixin:
         if _is_thumbnail_cache_path(relative):
             # A preview is a placeholder: it carries no metadata (#317).
             return _forbidden("Previews are not inspected; inspect the original")
+        if in_trash_folder(relative):
+            return web.json_response(
+                {"success": False, "error": "Image file not found"}, status=404
+            )
         path = Path(roots[index]) / relative
         denied = _media_access_error(path, [roots[index]])
         if denied is None:

py/api/library.py

@@ -15,7 +15,7 @@ from pathlib import Path
 
 from aiohttp import web
 
-from .images import _canonical, _media_access_error
+from .images import _canonical, _media_access_error, _record_in_trash_folder
 from .pairs import pair_response
 
 try:
@@ -268,6 +268,8 @@ class LibraryRoutesMixin:
                 denied = _media_access_error(path, roots)
                 if denied is not None or not path.is_file():
                     continue
+                if _record_in_trash_folder(path, roots):
+                    continue
                 for index, root in enumerate(roots):
                     canonical_root = _canonical(root)
                     if path.is_relative_to(canonical_root):

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

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

@@ -26,6 +26,15 @@ 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_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 */ }