fix(trash): trash on an output folder that is a mount point (#543) (#544)

Closes #543

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

@@ -184,6 +184,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}",
+    ),
+}