feat(images): review and apply Delete images as a reviewed plan job (#480)

POST /prompt_manager/images/delete/preview reviews both modes; the
maintenance apply takes {token, mode}. Library mode removes the records and
writes the removal mark; trash mode moves the file first and removes its
records only after a successful move. The workspace listing, the output scan,
the duplicate find and Tag output images skip marked files.

py/api/__init__.py

@@ -74,6 +74,7 @@ from .autotag_routes import AutotagRoutesMixin
 from .lora_integration import LoraIntegrationMixin
 from .lora_import_routes import LoraImportRoutesMixin
 from .image_categories import ImageCategoryRoutesMixin
+from .image_delete import ImageDeleteMixin
 from .source_rescope import SourceRescopeMixin
 
 try:
@@ -343,6 +344,7 @@ class PromptManagerAPI(
     LoraIntegrationMixin,
     LoraImportRoutesMixin,
     ImageCategoryRoutesMixin,
+    ImageDeleteMixin,
     SourceRescopeMixin,
 ):
     """REST API handler for PromptManager operations and web interface.
@@ -710,6 +712,7 @@ class PromptManagerAPI(
         self._register_lora_routes(routes)
         self._register_lora_import_routes(routes)
         self._register_image_category_routes(routes)
+        self._register_image_delete_routes(routes)
         self._register_rescope_routes(routes)
 
         # Register gzip compression middleware (once)

py/api/admin.py

@@ -883,6 +883,8 @@ class AdminRoutesMixin:
         extensions = IMAGE_EXTENSIONS + VIDEO_EXTENSIONS
         media_files = []
         seen = set()
+        # #480: a file removed from the library is never a copy or a keeper.
+        marks = self.db.removal_marks()
         for slot, root in enumerate(roots):
             for path in _collect_media_files([root], extensions):
                 try:
@@ -891,6 +893,12 @@ class AdminRoutesMixin:
                     stat = os.stat(path)
                 except OSError:
                     continue
+                mark = marks.get(os.path.normcase(os.path.realpath(str(path))))
+                if mark is not None and tuple(mark) == (
+                    stat.st_size,
+                    stat.st_mtime_ns,
+                ):
+                    continue
                 key = _physical_file_key(path, stat)
                 if key not in seen:
                     seen.add(key)
@@ -1551,8 +1559,13 @@ class AdminRoutesMixin:
             if job.cancelled:
                 return
             batch = media_files[batch_start : batch_start + SCAN_BATCH_SIZE]
-            skipped = [f for f in batch if str(f) in known]
-            to_read = [f for f in batch if str(f) not in known]
+            # #480: a file removed from the library is not read or linked
+            # again while its bytes are unchanged.
+            removed = await self._run_in_executor(self._removed_paths, batch)
+            skipped = [f for f in batch if str(f) in known or str(f) in removed]
+            to_read = [
+                f for f in batch if str(f) not in known and str(f) not in removed
+            ]
             outcomes = [_scan_outcome(f, roots, "skipped") for f in skipped]
             counts["processed"] += len(skipped)
             counts["unchanged"] += len(skipped)

py/api/image_delete.py

@@ -0,0 +1,465 @@
+"""Delete selected images: remove them from the library, or trash them (#480).
+
+``POST /prompt_manager/images/delete/preview {items | filter}`` reviews the
+selection for both modes at once and answers a reviewed plan (the System ›
+Maintenance contract, ``py/api/maintenance.py``): the token, the counts of
+the records each file loses, the refused files with a coded reason (both
+modes, and the trash-only ones), the prompts that lose their last image and
+the safety backup. It writes nothing. ``items`` are Images listing identities
+(at most 500); ``filter`` is "Select all N matching". Each target is matched
+to its entry in the current listing (``not-listed`` otherwise), so every
+later step uses the walked spelling, never the request's text.
+
+``POST /prompt_manager/maintenance/apply {token, mode}`` runs the plan as the
+``image_delete`` job (admission ``records``, not cancellable): the sources
+are checked, the promised safety backup is taken, then each file is checked
+again (containment and fingerprint) and
+
+- ``mode="library"``: its records are deleted and its removal mark written
+  in one transaction; the file stays on disk, hidden;
+- ``mode="trash"``: it is moved to the trash first (``utils/trash.py``); a
+  refused move leaves the file and its records alone; after a move its
+  records are deleted.
+
+No path here unlinks a media file. A failed records write stops the job; the
+files after it are not attempted.
+"""
+
+import hashlib
+import json
+import os
+import stat as stat_module
+from pathlib import Path
+
+from .admin import DELETABLE_MEDIA_EXTENSIONS, _remove_thumbnail
+from .duplicate_review import (
+    CHANGED_CONTENT,
+    GONE,
+    LINK,
+    OUTSIDE_SOURCE,
+    UNREADABLE,
+    file_problem,
+)
+from .images import _favorite_relative_key, _favorite_root_key
+from .jobs import IMAGE_DELETE_GROUP_NAMES, IMAGE_DELETE_JOB_KIND, JobBusy
+from .job_routes import busy_response
+from .maintenance import MAINTENANCE_PREVIEW_LEASE, _fail_job, _failure, backup_note
+
+try:
+    from ...database.annotation_schema import MAX_LOCATOR_LENGTH
+    from ...utils import trash
+except ImportError:
+    from database.annotation_schema import MAX_LOCATOR_LENGTH
+    from utils import trash
+
+IMAGE_DELETE = "delete_images"
+MODES = ("library", "trash")
+TITLES = {
+    "library": "Delete images (remove from the library)",
+    "trash": "Delete images (move to the trash)",
+}
+MAX_DETAIL_ROWS = 50
+TRASH_FOLDERS = frozenset({".trash", ".trashes", "$recycle.bin"})
+
+# Reason codes (design §4.4) and the sentence each shows.
+REASONS = {
+    "not-listed": "No longer listed in Images.",
+    "outside": "Outside the output folders.",
+    "link": "A link, not a file.",
+    "gone": "The file no longer exists.",
+    "unreadable": "The file can't be read.",
+    "thumbnail": "A generated preview, not an image.",
+    "not-media": "Not a media file.",
+    "trash-folder": "Inside a trash folder.",
+    "path-too-long": "The path is too long.",
+    "changed": "The file changed after the review.",
+    trash.NO_TRASH: "This drive has no trash.",
+    trash.NETWORK: "Network drives can't use the trash.",
+    trash.TRASH_IN_ROOT: "This drive's trash is inside an output folder.",
+    trash.LOCKED: "The file is open in another program. Close it and try again.",
+    trash.FAILED: "The file could not be moved to the trash.",
+    "not-attempted": "Not attempted: the job stopped after an error.",
+}
+_PROBLEM_CODES = {
+    OUTSIDE_SOURCE: "outside",
+    LINK: "link",
+    GONE: "gone",
+    UNREADABLE: "unreadable",
+    CHANGED_CONTENT: "changed",
+}
+SOURCES_CHANGED = "The sources changed since the review; nothing was changed."
+RECORDS_FAILED = "The image records could not be removed; the job stopped."
+
+
+def _is_trash_folder(name: str) -> bool:
+    folded = name.casefold()
+    return folded in TRASH_FOLDERS or (
+        folded.startswith(".trash-") and folded[7:].isdigit()
+    )
+
+
+def lexical_problem(relative: str) -> str | None:
+    """The refusal code a root-relative POSIX path earns before any file access."""
+    parts = relative.split("/")
+    if "thumbnails" in parts[:-1]:
+        return "thumbnail"
+    if any(_is_trash_folder(part) for part in parts[:-1]):
+        return "trash-folder"
+    if os.path.splitext(parts[-1])[1].lower() not in DELETABLE_MEDIA_EXTENSIONS:
+        return "not-media"
+    return None
+
+
+def file_check(path, root, relative, *, expected=None):
+    """``(code, lstat)``: why *path* may not be deleted, or ``(None, lstat)``.
+
+    Text first (inside the root, not a preview or trash folder, a media
+    file), then ``duplicate_review.file_problem``: a regular file, not a
+    link, whose resolved path is still inside the resolved root. With
+    *expected* (``{size, mtime_ns, inode}`` from the review) the file must
+    still be those bytes. Every handle is closed when this returns.
+    """
+    code = lexical_problem(relative)
+    if code:
+        return code, None
+    try:
+        info = os.lstat(path)
+    except FileNotFoundError:
+        return "gone", None
+    except OSError:
+        return "unreadable", None
+    if getattr(info, "st_file_attributes", 0) & getattr(
+        stat_module, "FILE_ATTRIBUTE_REPARSE_POINT", 0
+    ):
+        return "link", info
+    if expected is None:
+        problem = file_problem(path, root, None, info.st_size, hash_check=False)
+    else:
+        problem = file_problem(
+            path,
+            root,
+            None,
+            expected["size"],
+            hash_check=False,
+            mtime_ns=expected["mtime_ns"],
+            inode=expected["inode"],
+        )
+    if problem:
+        return _PROBLEM_CODES.get(problem, "outside"), info
+    if len(os.path.normcase(os.path.realpath(path))) > MAX_LOCATOR_LENGTH:
+        return "path-too-long", info
+    return None, info
+
+
+def _row(target, code):
+    return {
+        "path": target["rel"],
+        "root_label": target["root_label"],
+        "code": code,
+        "detail": REASONS[code],
+    }
+
+
+def _summary(rows):
+    counts = {}
+    for row in rows:
+        counts[row["code"]] = counts.get(row["code"], 0) + 1
+    return {"count": len(rows), "counts": counts, "rows": rows[:MAX_DETAIL_ROWS]}
+
+
+def _fingerprint(targets) -> str:
+    material = [
+        (t["root_index"], t["rel"], t["size"], t["mtime_ns"], t["inode"])
+        for t in targets
+    ]
+    return hashlib.sha256(json.dumps(material).encode("utf-8")).hexdigest()
+
+
+class ImageDeleteMixin:
+    """The Images Delete… review and its apply job (#480)."""
+
+    def _register_image_delete_routes(self, routes):
+        # A fixed path beside the dynamic /images/{image_id} routes: aiohttp
+        # (3.10+) looks fixed paths up first (see image_categories.py).
+        @routes.post("/prompt_manager/images/delete/preview")
+        async def preview_image_delete_route(request):
+            return await self.preview_image_delete(request)
+
+    # ── review ────────────────────────────────────────────────────────
+
+    async def preview_image_delete(self, request):
+        """Review the selection under the maintenance preview lease."""
+        try:
+            async with self._jobs.lease(MAINTENANCE_PREVIEW_LEASE):
+                return await self._preview_image_delete(request)
+        except JobBusy as busy:
+            return busy_response(busy)
+
+    async def _preview_image_delete(self, request):
+        _data, targets, roots, error = await self._category_request(
+            request, frozenset()
+        )
+        if error is not None:
+            return error
+        db = self.db
+        identity = self._database_identity(db)
+        try:
+            files = {}
+            for index in dict.fromkeys(t[0] for t in targets):
+                files[index] = await self._get_gallery_files(roots[index])
+            review = await self._run_in_executor(
+                self._review_image_delete, db, roots, targets, files
+            )
+        except Exception:
+            self.logger.exception("Image delete review failed")
+            return _failure("The review failed. Check server logs for details.", 500)
+        return await self._plan_response(
+            db,
+            identity,
+            IMAGE_DELETE,
+            _fingerprint(review["targets"]),
+            review["public"],
+            {"targets": review["targets"]},
+        )
+
+    def _review_image_delete(self, db, roots, targets, files):
+        """Every target checked now (blocking); ``{targets, public}``."""
+        listings, seen = {}, set()
+        acted, refused, trash_refused = [], [], []
+        spellings = self._source_spellings()
+        root_texts = [str(root) for root in roots]
+        for index, _scope, relative in targets:
+            root = roots[index]
+            if index not in listings:
+                listings[index] = self._favorite_listing(root, files[index])
+            spelled = listings[index].get(_favorite_relative_key(relative))
+            target = {
+                "root_index": index,
+                "root_scope": _scope,
+                "root": str(root),
+                "root_label": self._public_path(root),
+                "rel": spelled or relative,
+            }
+            if spelled is None:
+                refused.append(_row(target, "not-listed"))
+                continue
+            absolute = os.path.join(str(root), *spelled.split("/"))
+            code, info = file_check(absolute, str(root), spelled)
+            if code:
+                refused.append(_row(target, code))
+                continue
+            identity = (info.st_dev, info.st_ino)
+            if identity in seen:
+                continue  # one physical file named twice (nested roots, hardlinks)
+            seen.add(identity)
+            target.update(self._removal_of(absolute, info, spellings))
+            target["trash"] = trash.trash_available(absolute, root_texts)
+            if target["trash"]:
+                trash_refused.append(_row(target, target["trash"]))
+            acted.append(target)
+        counted = db.count_removal_records([t["removal"] for t in acted])
+        library_count = len(acted)
+        public = {
+            "operation": IMAGE_DELETE,
+            "scope": {"count": library_count, "label": "images"},
+            "library": {"count": library_count},
+            "trash": {"count": library_count - len(trash_refused)},
+            "protected": {"count": len(refused), "reasons": []},
+            "refused": _summary(refused),
+            "trash_refused": _summary(trash_refused),
+            "records": counted["records"],
+            "orphaned_prompts": counted["orphaned_prompts"],
+            "media": {"deletes_files": False, "label": "Chosen in the review"},
+            "details": [],
+        }
+        return {"targets": acted, "public": public}
+
+    @staticmethod
+    def _removal_of(absolute, info, spellings):
+        favorites = [
+            (_favorite_root_key(root), _favorite_relative_key(relative))
+            for root, relative in spellings.favorites(absolute).items()
+        ]
+        return {
+            "abs": absolute,
+            "size": info.st_size,
+            "mtime_ns": info.st_mtime_ns,
+            "inode": [info.st_dev, info.st_ino],
+            "removal": {
+                "spellings": list(spellings.of(absolute).values()),
+                "locator": os.path.normcase(os.path.realpath(absolute)),
+                "favorites": favorites,
+                "category": spellings.category_key(absolute),
+            },
+        }
+
+    # ── apply ─────────────────────────────────────────────────────────
+
+    def _start_image_delete(self, db, plan, mode, backup_on):
+        """Register the delete job for a reviewed plan (on the loop)."""
+        return self._jobs.start(
+            IMAGE_DELETE_JOB_KIND,
+            TITLES[mode],
+            admission="records",
+            cancellable=False,
+            scope={"operation": IMAGE_DELETE, "mode": mode},
+            groups=IMAGE_DELETE_GROUP_NAMES,
+            runner=lambda job: self._run_image_delete(job, db, plan, mode, backup_on),
+        )
+
+    async def _run_image_delete(self, job, db, plan, mode, backup_on):
+        targets = plan.payload["targets"]
+        unchanged = await self._run_in_executor(self._sources_unchanged, targets)
+        if not unchanged:
+            return _fail_job(job, "sources-changed", SOURCES_CHANGED)
+        ok, backup = await self._safety_backup_step(job, db, backup_on)
+        if not ok:
+            return
+        try:
+            await self._delete_images(job, db, targets, mode, backup)
+        finally:
+            if backup:
+                await self._prune_backup_pool(job, db)
+
+    async def _delete_images(self, job, db, targets, mode, backup):
+        job.update(total=len(targets), status_line=TITLES[mode])
+        roots = [str(root) for root in self._get_all_output_dirs()]
+        done_group = "trashed" if mode == "trash" else "removed"
+        changed, failed = 0, False
+        for done, target in enumerate(targets, start=1):
+            if failed:
+                self._record_delete(job, "not_changed", target, "not-attempted")
+                continue
+            group, code, label = await self._run_in_executor(
+                self._delete_one_image, db, target, mode, roots
+            )
+            if group == "errors":
+                failed = True
+                job.record(
+                    "errors",
+                    {
+                        "path": target["rel"],
+                        "root": target["root_label"],
+                        "root_index": target["root_index"],
+                        "root_scope": target["root_scope"],
+                        "detail": RECORDS_FAILED,
+                    },
+                    f"! {target['root_label']}/{target['rel']} · {RECORDS_FAILED}",
+                )
+            elif group == done_group:
+                changed += 1
+                detail = (
+                    f"Moved to the {label}" if label else "Removed from the library"
+                )
+                self._record_delete(job, group, target, None, detail)
+            else:
+                self._record_delete(job, group, target, code)
+            job.update(processed=done, progress=int(done / len(targets) * 100))
+        if changed:
+            self.invalidate_gallery_cache()
+        verb = "Moved" if mode == "trash" else "Removed"
+        where = "to the trash" if mode == "trash" else "from the library"
+        line = f"{verb} {changed:,} of {len(targets):,} images {where}"
+        job.log_line(line)
+        job.update(progress=100, status_line=f"{line} · {backup_note(backup)}")
+        if failed:
+            job.scope["failure"] = "records-failed"
+            job.publish({"type": "error", "code": "records-failed", "message": line})
+            job.finish("failed")
+
+    @staticmethod
+    def _record_delete(job, group, target, code, detail=None):
+        text = detail or REASONS[code]
+        row = {
+            "path": target["rel"],
+            "root": target["root_label"],
+            "root_index": target["root_index"],
+            "root_scope": target["root_scope"],
+            "code": code,
+            "detail": text,
+        }
+        mark = {"removed": "-", "trashed": "-"}.get(group, "=")
+        job.record(
+            group, row, f"{mark} {target['root_label']}/{target['rel']} · {text}"
+        )
+
+    def _delete_one_image(self, db, target, mode, roots):
+        """``(group, code, trash label)`` for one reviewed file (blocking)."""
+        code, _info = file_check(
+            target["abs"],
+            target["root"],
+            target["rel"],
+            expected={
+                "size": target["size"],
+                "mtime_ns": target["mtime_ns"],
+                "inode": target["inode"],
+            },
+        )
+        if code:
+            return "not_changed", code, None
+        label = None
+        if mode == "trash":
+            try:
+                label = trash.move_to_trash(target["abs"], roots)
+            except trash.TrashRefused as refused:
+                code = refused.code if refused.code in REASONS else trash.FAILED
+                return "not_changed", code, None
+            mark = None
+        else:
+            mark = (target["size"], target["mtime_ns"])
+        try:
+            db.remove_image_records(target["removal"], mark=mark)
+        except Exception:
+            self.logger.exception("Removing an image's records failed")
+            return "errors", None, label
+        _remove_thumbnail(Path(target["root"]), Path(target["rel"]), self.logger)
+        return ("trashed" if mode == "trash" else "removed"), None, label
+
+    # ── the workspace listing ─────────────────────────────────────────
+
+    def _hidden_entries(self, listings) -> set:
+        """Normalized paths of listed files a removal mark hides (blocking).
+
+        Only marked locators are looked at, and only those files are
+        ``lstat``'ed: a mark hides a file while its size and ``mtime_ns`` are
+        still the ones it was removed with.
+        """
+        marks = self.db.removal_marks()
+        if not marks:
+            return set()
+        located = self._workspace_locators(listings)
+        hidden = set()
+        for locator, (size, mtime_ns) in marks.items():
+            entry = located.get(locator)
+            if entry is None:
+                continue
+            try:
+                info = os.lstat(entry.path)
+            except OSError:
+                continue
+            if (info.st_size, info.st_mtime_ns) == (size, mtime_ns):
+                hidden.add(entry.normalized)
+        return hidden
+
+    def _removed_paths(self, paths) -> set:
+        """The absolute paths among *paths* a removal mark hides (blocking)."""
+        marks = self.db.removal_marks()
+        if not marks:
+            return set()
+        hidden = set()
+        for path in paths:
+            mark = marks.get(os.path.normcase(os.path.realpath(str(path))))
+            if mark is None:
+                continue
+            try:
+                info = os.lstat(path)
+            except OSError:
+                continue
+            if (info.st_size, info.st_mtime_ns) == tuple(mark):
+                hidden.add(str(path))
+        return hidden
+
+
+def parse_mode(value):
+    """The apply's ``mode``, or None when it is not one of MODES."""
+    return value if isinstance(value, str) and value in MODES else None

py/api/image_tag_jobs.py

@@ -488,7 +488,11 @@ class ImageTagJobsMixin:
                     _still_targets, self, index, root, files, seen
                 )
             )
-        return found
+        # #480: a file removed from the library is never tagged by a batch.
+        hidden = await self._run_in_executor(
+            self._removed_paths, [target.path for target in found]
+        )
+        return [target for target in found if target.path not in hidden]
 
     async def _start_autotag(
         self, *, model_type, existing, options, keep_in_memory, title, owner_kind

py/api/images.py

@@ -2450,6 +2450,10 @@ class ImageRoutesMixin:
             def key(entry):
                 return _workspace_key(entry, order)
 
+        # #480: files removed from the library stay on disk, hidden here.
+        hidden = self._hidden_entries(listings)
+        if hidden:
+            entries = [entry for entry in entries if entry.normalized not in hidden]
         return (
             entries,
             key,

py/api/jobs.py

@@ -123,6 +123,8 @@ MODEL_DOWNLOAD_JOB_KIND = "model_download"
 DUPLICATES_JOB_KIND = "duplicates"
 DUPLICATE_GROUP_NAMES = ("groups_found",)
 DUPLICATE_CLEANUP_GROUP_NAMES = ("removed", "kept", "errors")
+IMAGE_DELETE_JOB_KIND = "image_delete"  # #480
+IMAGE_DELETE_GROUP_NAMES = ("trashed", "removed", "not_changed", "errors")
 # The System health check (py/api/system.py): a read-only ``read`` job with
 # no file outcomes, so it may run beside a scan.
 HEALTH_JOB_KIND = "health"

py/api/maintenance.py

@@ -488,7 +488,9 @@ class MaintenanceRoutesMixin:
     # ── apply ─────────────────────────────────────────────────────────
 
     async def apply_maintenance_plan(self, request):
-        data, error = await self._read_maintenance_body(request, ("token",))
+        from .image_delete import IMAGE_DELETE, parse_mode
+
+        data, error = await self._read_maintenance_body(request, ("token", "mode"))
         if error is not None:
             return error
         token = data.get("token")
@@ -497,6 +499,13 @@ class MaintenanceRoutesMixin:
         plan, refused = self._peek_plan(token)
         if refused is not None:
             return refused
+        mode = None
+        if plan.operation == IMAGE_DELETE:
+            mode = parse_mode(data.get("mode"))
+            if mode is None:
+                return _invalid('mode must be "library" or "trash"')
+        elif "mode" in data:
+            return _invalid("Unknown field: mode")
         store = self._maintenance_plans
         if plan.operation == ROTATED_LOGS:
             # Log files only: no database identity, backup or disk check.
@@ -544,6 +553,8 @@ class MaintenanceRoutesMixin:
         try:
             if plan.operation == DUPLICATE_FILES:
                 job = self._start_duplicate_cleanup(db, plan, backup)
+            elif plan.operation == IMAGE_DELETE:
+                job = self._start_image_delete(db, plan, mode, backup)
             elif plan.operation == WORKFLOW_STORAGE:
                 job = self._start_workflow_storage(db, backup)
             else:

tests/fake_trash.py

@@ -0,0 +1,52 @@
+"""A fake system trash for tests (#480): renames into a folder outside every root.
+
+``install(folder)`` replaces ``utils.trash.BACKEND`` and returns the fake;
+``uninstall`` puts the real one back. ``refuse[path] = code`` refuses one
+file with that code. The fake never copies or unlinks; it records each call.
+The disposable browser server installs it when ``PM_TEST_FAKE_TRASH=1``, so
+no test touches a real trash.
+"""
+
+import os
+
+from utils import trash
+
+
+class FakeTrash:
+    def __init__(self, folder):
+        self.folder = str(folder)
+        os.makedirs(self.folder, exist_ok=True)
+        self.refuse = {}
+        self.moved = []  # (source, destination)
+        self.calls = []
+
+    def available(self, path, roots):
+        self.calls.append(("available", str(path)))
+        return self.refuse.get(str(path))
+
+    def move(self, path, roots):
+        self.calls.append(("move", str(path)))
+        code = self.refuse.get(str(path))
+        if code:
+            raise trash.TrashRefused(code)
+        destination = os.path.join(
+            self.folder, f"{len(self.moved)}-{os.path.basename(path)}"
+        )
+        if os.path.lexists(destination):
+            raise trash.TrashRefused(trash.FAILED)
+        os.rename(path, destination)
+        self.moved.append((str(path), destination))
+        return "fake trash"
+
+
+_ORIGINAL = trash.BACKEND
+
+
+def install(folder):
+    fake = FakeTrash(folder)
+    trash.BACKEND = fake
+    return fake
+
+
+def uninstall():
+    trash.BACKEND = _ORIGINAL

tests/test_image_delete_routes.py

@@ -0,0 +1,281 @@
+"""Delete selected images: the review and the apply job (#480).
+
+``POST /prompt_manager/images/delete/preview`` reviews both modes;
+``POST /prompt_manager/maintenance/apply {token, mode}`` removes the files'
+records (library mode, with a removal mark) or moves the files to the trash
+first (trash mode, the fake trash here). No path unlinks a media file.
+"""
+
+import asyncio
+import os
+import sqlite3
+from pathlib import Path
+from unittest.mock import patch
+
+try:
+    from tests import fake_trash
+    from tests.disk_waits import DISK_BOUND_SECONDS
+    from tests.test_api_duplicate_files import DuplicateFilesTestCase
+except ImportError:  # discovered with tests/ as the top-level directory
+    import fake_trash
+    from disk_waits import DISK_BOUND_SECONDS
+    from test_api_duplicate_files import DuplicateFilesTestCase
+
+from py.api import image_delete
+from py.api.image_delete import file_check, lexical_problem
+
+PREVIEW = "/prompt_manager/images/delete/preview"
+APPLY = "/prompt_manager/maintenance/apply"
+LISTING = "/prompt_manager/images/output"
+
+
+class ImageDeleteTestCase(DuplicateFilesTestCase):
+    async def get_application(self):
+        app = await super().get_application()
+        self.fake = fake_trash.install(Path(self.tmpdir) / "fake-trash")
+        self.addCleanup(fake_trash.uninstall)
+        return app
+
+    def target(self, path, index=0):
+        roots = self.api._get_all_output_dirs()
+        return {
+            "root_index": index,
+            "root_scope": self.api._gallery_root_scope(roots[index], index),
+            "path": path,
+        }
+
+    async def review(self, paths, status=200):
+        items = [p if isinstance(p, dict) else self.target(p) for p in paths]
+        return await self.post(PREVIEW, {"items": items}, status)
+
+    async def delete(self, paths, mode):
+        plan = await self.review(paths)
+        body = await self.post(APPLY, {"token": plan["token"], "mode": mode}, 202)
+        job = self.api._jobs.get(body["job"]["id"])
+        await asyncio.wait_for(asyncio.shield(job.task), DISK_BOUND_SECONDS)
+        return plan, job
+
+    async def listed(self):
+        response = await self.client.get(LISTING, params={"mode": "workspace"})
+        page = await response.json()
+        return [image["path"] for image in page["images"]]
+
+    def link(self, name, text="a prompt"):
+        prompt_id = self.save(text)
+        return self.api.db.link_image_to_prompt(
+            str(prompt_id), str(self.output_dir / name)
+        )
+
+    def count(self, sql):
+        conn = sqlite3.connect(self.api.db.model.db_path)
+        try:
+            return conn.execute(sql).fetchone()[0]
+        finally:
+            conn.close()
+
+    @staticmethod
+    def details(job, group):
+        return [row["detail"] for row in job.detail()["outcomes"][group]]
+
+
+class TestReview(ImageDeleteTestCase):
+    async def test_review_counts_records_and_refuses_what_it_cannot_delete(self):
+        self.write("a.png", b"aaa", mtime=1_000_000)
+        self.write("b.png", b"bbb", mtime=2_000_000)
+        self.link("a.png", "only a")
+        self.fake.refuse[str(self.output_dir / "b.png")] = "no-trash"
+        plan = await self.review(["a.png", "b.png", "missing.png"])
+        self.assertEqual(plan["library"], {"count": 2})
+        self.assertEqual(plan["trash"], {"count": 1})
+        self.assertEqual(plan["refused"]["counts"], {"not-listed": 1})
+        self.assertEqual(plan["trash_refused"]["counts"], {"no-trash": 1})
+        self.assertEqual(plan["records"]["prompt_links"], 1)
+        self.assertEqual(plan["orphaned_prompts"]["count"], 1)
+        self.assertIn(plan["backup"]["policy"], ("on", "off"))
+        self.assertNotIn(self.tmpdir, str(plan))
+        # The review wrote nothing.
+        self.assertEqual(self.count("SELECT COUNT(*) FROM generated_images"), 1)
+        self.assertEqual(self.fake.moved, [])
+
+    async def test_body_shapes_and_the_cap(self):
+        await self.post(PREVIEW, {}, 400)
+        await self.post(PREVIEW, {"items": [], "mode": "trash"}, 400)
+        too_many = [self.target(f"{n}.png") for n in range(501)]
+        await self.post(PREVIEW, {"items": too_many}, 400)
+
+    async def test_mode_is_required_and_only_for_this_plan(self):
+        self.write("a.png", b"aaa")
+        plan = await self.review(["a.png"])
+        await self.post(APPLY, {"token": plan["token"]}, 400)
+        await self.post(APPLY, {"token": plan["token"], "mode": "delete"}, 400)
+        self.assertTrue((self.output_dir / "a.png").exists())
+
+
+class TestLibraryMode(ImageDeleteTestCase):
+    async def test_records_go_and_the_file_stays_hidden(self):
+        self.write("a.png", b"aaa", mtime=1_000_000)
+        self.write("b.png", b"bbb", mtime=2_000_000)
+        self.link("a.png")
+        self.assertEqual(sorted(await self.listed()), ["a.png", "b.png"])
+        before = os.stat(self.output_dir / "a.png")
+        _plan, job = await self.delete(["a.png"], "library")
+        self.assertEqual(job.state, "finished", job.summary())
+        self.assertEqual(self.count("SELECT COUNT(*) FROM generated_images"), 0)
+        after = os.stat(self.output_dir / "a.png")
+        self.assertEqual(
+            (before.st_size, before.st_mtime_ns), (after.st_size, after.st_mtime_ns)
+        )
+        self.assertEqual(await self.listed(), ["b.png"])
+        # A later scan or link never brings it back.
+        self.assertEqual(self.link("a.png", "rescanned"), 0)
+        self.assertEqual(
+            self.api._removed_paths([self.output_dir / "a.png"]),
+            {str(self.output_dir / "a.png")},
+        )
+        self.assertEqual(self.fake.moved, [])
+
+    async def test_a_modified_file_shows_again(self):
+        self.write("a.png", b"aaa", mtime=1_000_000)
+        await self.delete(["a.png"], "library")
+        self.assertEqual(await self.listed(), [])
+        self.write("a.png", b"new and longer", mtime=3_000_000)
+        self.api.invalidate_gallery_cache()
+        self.assertEqual(await self.listed(), ["a.png"])
+
+    async def test_a_safety_backup_is_taken_first(self):
+        self.write("a.png", b"aaa")
+        with patch.object(self.api, "_safety_backup_policy", return_value="on"):
+            await self.delete(["a.png"], "library")
+        self.assertEqual(len(self.backups()), 1)
+
+
+class TestTrashMode(ImageDeleteTestCase):
+    async def test_the_file_moves_first_then_its_records_go(self):
+        self.write("a.png", b"aaa")
+        self.link("a.png")
+        seen = {}
+        original = self.fake.move
+
+        def move(path, roots):
+            # The records still exist while the file moves.
+            seen["links"] = self.count("SELECT COUNT(*) FROM generated_images")
+            return original(path, roots)
+
+        self.fake.move = move
+        _plan, job = await self.delete(["a.png"], "trash")
+        self.assertEqual(job.state, "finished", job.summary())
+        self.assertEqual(seen["links"], 1)
+        self.assertFalse((self.output_dir / "a.png").exists())
+        [(source, destination)] = self.fake.moved
+        self.assertTrue(os.path.exists(destination))
+        self.assertEqual(self.count("SELECT COUNT(*) FROM generated_images"), 0)
+        self.assertEqual(self.api.db.removal_marks(), {})
+
+    async def test_a_refused_move_changes_nothing(self):
+        self.write("a.png", b"aaa")
+        self.link("a.png")
+        plan = await self.review(["a.png"])
+        self.fake.refuse[str(self.output_dir / "a.png")] = "locked"
+        body = await self.post(APPLY, {"token": plan["token"], "mode": "trash"}, 202)
+        job = self.api._jobs.get(body["job"]["id"])
+        await asyncio.wait_for(asyncio.shield(job.task), DISK_BOUND_SECONDS)
+        self.assertTrue((self.output_dir / "a.png").exists())
+        self.assertEqual(self.count("SELECT COUNT(*) FROM generated_images"), 1)
+        self.assertEqual(self.api.db.removal_marks(), {})
+        self.assertEqual(job.counters["not_changed"], 1)
+        self.assertEqual(
+            self.details(job, "not_changed"), [image_delete.REASONS["locked"]]
+        )
+
+    async def test_a_file_changed_after_the_review_is_not_changed(self):
+        self.write("a.png", b"aaa", mtime=1_000_000)
+        plan = await self.review(["a.png"])
+        self.write("a.png", b"changed bytes", mtime=2_000_000)
+        body = await self.post(APPLY, {"token": plan["token"], "mode": "trash"}, 202)
+        job = self.api._jobs.get(body["job"]["id"])
+        await asyncio.wait_for(asyncio.shield(job.task), DISK_BOUND_SECONDS)
+        self.assertTrue((self.output_dir / "a.png").exists())
+        self.assertEqual(self.fake.moved, [])
+        self.assertEqual(
+            self.details(job, "not_changed"), [image_delete.REASONS["changed"]]
+        )
+
+    async def test_a_records_failure_stops_the_job(self):
+        self.write("a.png", b"aaa", mtime=1_000_000)
+        self.write("b.png", b"bbb", mtime=2_000_000)
+        with patch.object(
+            type(self.api.db),
+            "remove_image_records",
+            side_effect=sqlite3.OperationalError("locked"),
+        ):
+            _plan, job = await self.delete(["a.png", "b.png"], "library")
+        self.assertEqual(job.state, "failed")
+        self.assertEqual(job.scope.get("failure"), "records-failed")
+        self.assertEqual(job.counters["errors"], 1)
+        self.assertEqual(
+            self.details(job, "not_changed"), [image_delete.REASONS["not-attempted"]]
+        )
+
+    async def test_nothing_is_ever_unlinked(self):
+        self.write("a.png", b"aaa")
+        self.write("b.png", b"bbb")
+        media = str(self.output_dir)
+        real_remove, real_unlink = os.remove, os.unlink
+
+        def guard(real):
+            def call(path, *args, **kwargs):
+                text = os.path.realpath(str(path))
+                if (
+                    text.startswith(os.path.realpath(media))
+                    and "thumbnails" not in text
+                ):
+                    raise AssertionError(f"unlinked a media file: {path}")
+                return real(path, *args, **kwargs)
+
+            return call
+
+        with (
+            patch("os.remove", guard(real_remove)),
+            patch("os.unlink", guard(real_unlink)),
+        ):
+            await self.delete(["a.png"], "trash")
+            await self.delete(["b.png"], "library")
+        self.assertTrue((self.output_dir / "b.png").exists())
+
+
+class TestFileCheck(ImageDeleteTestCase):
+    def test_lexical_codes(self):
+        self.assertEqual(lexical_problem("thumbnails/a.png"), "thumbnail")
+        self.assertEqual(lexical_problem(".Trash-1000/files/a.png"), "trash-folder")
+        self.assertEqual(lexical_problem("x/$RECYCLE.BIN/a.png"), "trash-folder")
+        self.assertEqual(lexical_problem("a.txt"), "not-media")
+        self.assertIsNone(lexical_problem(".Trashy/a.png"))
+        self.assertIsNone(lexical_problem("sub/a.png"))
+
+    def test_links_and_folders_leading_outside_are_refused(self):
+        root = str(self.output_dir)
+        outside = Path(self.tmpdir) / "elsewhere"
+        outside.mkdir()
+        (outside / "x.png").write_bytes(b"x")
+        os.symlink(outside / "x.png", self.output_dir / "link.png")
+        os.symlink(outside, self.output_dir / "linked")
+        self.assertEqual(
+            file_check(str(self.output_dir / "link.png"), root, "link.png")[0], "link"
+        )
+        self.assertEqual(
+            file_check(str(self.output_dir / "linked/x.png"), root, "linked/x.png")[0],
+            "outside",
+        )
+        self.assertEqual(
+            file_check(str(self.output_dir / "gone.png"), root, "gone.png")[0], "gone"
+        )
+        self.write("ok.png", b"ok")
+        self.assertIsNone(
+            file_check(str(self.output_dir / "ok.png"), root, "ok.png")[0]
+        )
+
+    def test_reasons_cover_every_code(self):
+        from utils import trash
+
+        for code in trash.CODES:
+            self.assertIn(code, image_delete.REASONS)