diff --git a/docs/src/content/docs/features/gallery.mdx b/docs/src/content/docs/features/gallery.mdx
index 2fc37776a19..3ea698efc91 100644
--- a/docs/src/content/docs/features/gallery.mdx
+++ b/docs/src/content/docs/features/gallery.mdx
@@ -138,6 +138,10 @@ Additionally, each image has a context menu (right-click or Ctrl+click) with pow
   Selecting **Delete Image** will remove the image entirely from your InvokeAI installation. This action cannot be undone.
 :::
 
+:::note
+A deletion removes the image's gallery record first and its files immediately afterwards. If the record cannot be removed, the image is left completely untouched — it stays in the gallery with its files intact. If Invoke is interrupted, or the storage refuses the delete, after the record is gone, the leftover files are noted in a journal and cleaned up the next time Invoke starts. The same applies to **Clear Intermediates**.
+:::
+
 ---
 
 ## Videos in the Gallery
diff --git a/invokeai/app/api/routers/images.py b/invokeai/app/api/routers/images.py
index cd7e42f1a26..e192dfd905a 100644
--- a/invokeai/app/api/routers/images.py
+++ b/invokeai/app/api/routers/images.py
@@ -232,25 +232,35 @@ def delete_image(
     _assert_image_owner(image_name, current_user)
     assert_image_move_maintenance_inactive()
 
-    deleted_images: set[str] = set()
-    affected_boards: set[str] = set()
-
+    # Let service-level failures surface as errors rather than swallowing them and returning
+    # a success-shaped response. A previous version of this handler caught everything and
+    # returned an empty ``deleted_images`` list with HTTP 200; the frontend treated that as
+    # success and dropped the item from its cache even though the record was still live.
     try:
         image_dto = ApiDependencies.invoker.services.images.get_dto(image_name)
-        board_id = image_dto.board_id or "none"
+    except ImageRecordNotFoundException:
+        raise HTTPException(status_code=404, detail="Image not found")
+    except Exception:
+        # A record/URL/board lookup failure for an image that does exist is a server fault, not a
+        # missing image — reporting it as 404 would tell the frontend to drop a live item.
+        raise HTTPException(status_code=500, detail="Failed to delete image")
+
+    board_id = image_dto.board_id or "none"
+    try:
         ApiDependencies.invoker.services.images.delete(image_name)
-        deleted_images.add(image_name)
-        affected_boards.add(board_id)
+    except ImageRecordNotFoundException:
+        # Another request deleted the image between the lookup above and the service call. The
+        # image is gone, which is what the client asked for — answer as the lookup would have.
+        raise HTTPException(status_code=404, detail="Image not found")
     except Exception:
-        # TODO: Does this need any exception handling at all?
-        pass
+        raise HTTPException(status_code=500, detail="Failed to delete image")
 
     return DeleteImagesResult(
-        deleted_images=list(deleted_images),
-        # Single-image route: the swallowed failure above already leaves deleted_images empty,
-        # which is how this route has always reported it.
+        deleted_images=[image_name],
+        # Every failure path above raises, so a returned result always describes a completed
+        # delete; nothing can land in ``failed_images``.
         failed_images=[],
-        affected_boards=list(affected_boards),
+        affected_boards=[board_id],
     )
 
 
diff --git a/invokeai/app/services/image_files/image_files_base.py b/invokeai/app/services/image_files/image_files_base.py
index 782e661c4b0..a5eb8fdc89e 100644
--- a/invokeai/app/services/image_files/image_files_base.py
+++ b/invokeai/app/services/image_files/image_files_base.py
@@ -1,4 +1,5 @@
 from abc import ABC, abstractmethod
+from collections.abc import Collection, Sequence
 from pathlib import Path
 from typing import Optional
 
@@ -67,8 +68,27 @@ def stage_delete(self, image_name: str, image_subfolder: str = "") -> object:
         pass
 
     @abstractmethod
-    def commit_delete(self, token: object) -> None:
-        """Permanently removes files represented by a staged-delete token."""
+    def begin_delete(self, images: Sequence[tuple[str, str]]) -> object:
+        """Durably records the intent to purge the (image_name, image_subfolder) pairs' files.
+
+        Call this before deleting the records, then ``commit_delete()`` after. If the process dies
+        in between, startup recovery uses the journal to purge the files of every listed image
+        whose record is gone, and leaves the files of every image whose record survives.
+        """
+        pass
+
+    @abstractmethod
+    def commit_delete(self, token: object, image_names: Optional[Collection[str]] = None) -> None:
+        """Permanently removes the files represented by a delete token.
+
+        ``image_names`` narrows a pending-delete token to the records that were actually deleted;
+        it is ignored for a staged-delete token.
+        """
+        pass
+
+    @abstractmethod
+    def abandon_delete(self, token: object) -> None:
+        """Drops a pending-delete journal without purging anything, leaving the files in place."""
         pass
 
     @abstractmethod
diff --git a/invokeai/app/services/image_files/image_files_disk.py b/invokeai/app/services/image_files/image_files_disk.py
index 133c71ab05d..2c6ab8286cc 100644
--- a/invokeai/app/services/image_files/image_files_disk.py
+++ b/invokeai/app/services/image_files/image_files_disk.py
@@ -6,6 +6,7 @@
 import tempfile
 import threading
 import zlib
+from collections.abc import Collection, Sequence
 from dataclasses import dataclass
 from pathlib import Path
 from queue import Queue
@@ -32,8 +33,25 @@
 
 @dataclass
 class _StagedDelete:
+    """Files moved aside by ``stage_delete()``, restorable until the token is committed."""
+
     directory: Path
     files: list[tuple[Path, Path]]
+    image_name: str
+    image_subfolder: str
+
+
+@dataclass
+class _PendingDelete:
+    """A durable record of an intent to purge files, written before the records are deleted.
+
+    Nothing is moved: the journal directory names the images whose files are about to become
+    unreferenced. Startup recovery reconciles any journal that outlives its operation by asking the
+    record store which of its images are really gone.
+    """
+
+    directory: Path
+    images: list[tuple[str, str]]
 
 
 def _get_png_size(image: PILImageType, compress_type: Optional[int] = None) -> int:
@@ -89,7 +107,7 @@ def __init__(self, output_folder: Union[str, Path]):
 
     def start(self, invoker: Invoker) -> None:
         self.__invoker = invoker
-        self.__recover_staged_deletes()
+        self.__recover_pending_deletes()
 
     @property
     def image_root(self) -> Path:
@@ -202,10 +220,7 @@ def delete(self, image_name: str, image_subfolder: str = "") -> None:
         self.commit_delete(token)
 
     def stage_delete(self, image_name: str, image_subfolder: str = "") -> _StagedDelete:
-        candidates = [
-            self.get_path(image_name, image_subfolder=image_subfolder),
-            self.get_path(image_name, thumbnail=True, image_subfolder=image_subfolder),
-        ]
+        candidates = self.__delete_candidates(image_name, image_subfolder)
         staging_dir = Path(tempfile.mkdtemp(prefix=".delete_", dir=self.__output_folder))
         staged: list[tuple[Path, Path]] = []
         try:
@@ -213,6 +228,9 @@ def stage_delete(self, image_name: str, image_subfolder: str = "") -> _StagedDel
                 manifest.write(json.dumps({"image_name": image_name, "image_subfolder": image_subfolder}))
                 manifest.flush()
                 os.fsync(manifest.fileno())
+            # The manifest has to be durable before the files move, or a crash can leave staged
+            # files in a directory that names nothing and recovery cannot put them back.
+            self.__persist_journal_directory(staging_dir)
             for index, source in enumerate(candidates):
                 with self.__cache_lock:
                     self.__cache.pop(source, None)
@@ -220,7 +238,9 @@ def stage_delete(self, image_name: str, image_subfolder: str = "") -> _StagedDel
                     destination = staging_dir / str(index)
                     source.replace(destination)
                     staged.append((source, destination))
-            return _StagedDelete(directory=staging_dir, files=staged)
+            return _StagedDelete(
+                directory=staging_dir, files=staged, image_name=image_name, image_subfolder=image_subfolder
+            )
         except Exception as e:
             for source, destination in reversed(staged):
                 if destination.exists():
@@ -229,14 +249,79 @@ def stage_delete(self, image_name: str, image_subfolder: str = "") -> _StagedDel
             shutil.rmtree(staging_dir, ignore_errors=True)
             raise ImageFileDeleteException from e
 
-    def commit_delete(self, token: object) -> None:
+    def begin_delete(self, images: Sequence[tuple[str, str]]) -> _PendingDelete:
+        """Durably records the intent to purge these images' files, before their records are deleted.
+
+        Callers delete the records first and purge afterwards, so the only inconsistency that can
+        outlive a crash or a storage failure is a file nobody references. The journal written here
+        is what makes that recoverable: ``__recover_pending_deletes()`` asks the record store about
+        every image it names, purges the ones whose record is gone, and leaves the rest untouched.
+        """
+        # Resolve every path up front. A name that cannot be turned into a path must fail here,
+        # while the caller can still abort — not after it has deleted the records.
+        for image_name, image_subfolder in images:
+            self.__delete_candidates(image_name, image_subfolder)
+        journal_dir = Path(tempfile.mkdtemp(prefix=".delete_", dir=self.__output_folder))
+        try:
+            entries = [
+                {"image_name": image_name, "image_subfolder": image_subfolder} for image_name, image_subfolder in images
+            ]
+            manifest_path = journal_dir / "manifest.json"
+            with open(manifest_path, "w", encoding="utf-8") as manifest:
+                manifest.write(json.dumps({"version": 2, "images": entries}))
+                manifest.flush()
+                os.fsync(manifest.fileno())
+            self.__persist_journal_directory(journal_dir)
+            return _PendingDelete(directory=journal_dir, images=[(name, subfolder) for name, subfolder in images])
+        except Exception as e:
+            shutil.rmtree(journal_dir, ignore_errors=True)
+            raise ImageFileDeleteException from e
+
+    def abandon_delete(self, token: object) -> None:
+        """Drops a pending-delete journal without purging anything.
+
+        Used when the record deletion failed: the images are still live, so their files must stay.
+        """
+        if not isinstance(token, _PendingDelete):
+            raise ImageFileDeleteException("Invalid pending-delete token")
+        shutil.rmtree(token.directory, ignore_errors=True)
+
+    def commit_delete(self, token: object, image_names: Optional[Collection[str]] = None) -> None:
+        if isinstance(token, _PendingDelete):
+            self.__commit_pending_delete(token, image_names)
+            return
         if not isinstance(token, _StagedDelete):
             raise ImageFileDeleteException("Invalid staged-delete token")
         try:
+            # Purge the live paths as well as the staged copies. stage_delete() captures whatever
+            # was there at that instant, so a second deleter racing the first gets an empty token —
+            # and if that second one is the one whose record deletion succeeds, dropping its empty
+            # staging directory alone would strand the files the first deleter restores. Committing
+            # has to mean "no file for this image survives", whichever request moved them.
+            self.__purge_files(token.image_name, token.image_subfolder)
             shutil.rmtree(token.directory)
         except Exception as e:
             raise ImageFileDeleteException from e
 
+    def __commit_pending_delete(self, token: _PendingDelete, image_names: Optional[Collection[str]]) -> None:
+        # ``image_names`` narrows the purge to the records that were actually deleted; the journal
+        # still lists every candidate, which is harmless because recovery re-checks each one against
+        # the record store and skips any that survived.
+        selected = image_names if image_names is None else set(image_names)
+        failures: list[str] = []
+        for image_name, image_subfolder in token.images:
+            if selected is not None and image_name not in selected:
+                continue
+            try:
+                self.__purge_files(image_name, image_subfolder)
+            except OSError as e:
+                failures.append(f"{image_name}: {e}")
+        if failures:
+            # Leave the journal in place so the next startup retries every entry whose record is
+            # gone. Removing it here would turn a transient storage error into a permanent orphan.
+            raise ImageFileDeleteException(f"Failed to purge deleted image files: {'; '.join(failures)}")
+        shutil.rmtree(token.directory, ignore_errors=True)
+
     def rollback_delete(self, token: object) -> None:
         if not isinstance(token, _StagedDelete):
             raise ImageFileDeleteException("Invalid staged-delete token")
@@ -245,10 +330,67 @@ def rollback_delete(self, token: object) -> None:
                 if destination.exists():
                     source.parent.mkdir(parents=True, exist_ok=True)
                     destination.replace(source)
+            # While these files sat in the staging directory another request may have deleted the
+            # record; restoring them would leave files nothing references and no journal to find
+            # them by. Re-check now that the files are back: every deleter purges an image's files
+            # only *after* its record is committed as gone, so a record still present here cannot
+            # have been purged before this restore, and a record already absent means the purge
+            # either found nothing or is still to come — either way the files must go.
+            self.__purge_if_record_absent(token.image_name, token.image_subfolder)
             shutil.rmtree(token.directory, ignore_errors=True)
         except Exception as e:
             raise ImageFileDeleteException from e
 
+    def __delete_candidates(self, image_name: str, image_subfolder: str) -> list[Path]:
+        return [
+            self.get_path(image_name, image_subfolder=image_subfolder),
+            self.get_path(image_name, thumbnail=True, image_subfolder=image_subfolder),
+        ]
+
+    def __purge_files(self, image_name: str, image_subfolder: str) -> None:
+        """Removes an image's file and thumbnail. Missing files are not an error."""
+        for path in self.__delete_candidates(image_name, image_subfolder):
+            with self.__cache_lock:
+                self.__cache.pop(path, None)
+            path.unlink(missing_ok=True)
+
+    def __purge_if_record_absent(self, image_name: str, image_subfolder: str) -> None:
+        try:
+            record_exists = self.__invoker.services.image_records.exists(image_name)
+        except Exception as e:
+            # A storage fault must never destroy a live image's files. Keep them: a stale file is
+            # recoverable at the next startup, a deleted one is not.
+            InvokeAILogger.get_logger().error(f"Could not confirm whether {image_name} still exists: {e}")
+            return
+        if record_exists:
+            return
+        self.__purge_files(image_name, image_subfolder)
+
+    def __persist_journal_directory(self, journal_dir: Path) -> None:
+        """Makes a journal directory and its manifest survive a power loss.
+
+        Both fsyncs are needed: the first commits ``manifest.json``'s entry inside the journal
+        directory, the second commits the journal directory's own entry in the output folder.
+        Without the second, the record deletion — which SQLite does fsync — can outlive the journal
+        that is supposed to make it recoverable.
+        """
+        self.__fsync_directory(journal_dir)
+        self.__fsync_directory(self.__output_folder)
+
+    @staticmethod
+    def __fsync_directory(directory: Path) -> None:
+        try:
+            dir_fd = os.open(directory, os.O_RDONLY)
+        except OSError:
+            # Windows cannot open a directory for fsync; the manifest write above is all we get.
+            return
+        try:
+            os.fsync(dir_fd)
+        except OSError:
+            pass
+        finally:
+            os.close(dir_fd)
+
     def get_path(self, image_name: str, thumbnail: bool = False, image_subfolder: str = "") -> Path:
         base_folder = self.__thumbnails_folder if thumbnail else self.__output_folder
         filename = get_thumbnail_name(image_name) if thumbnail else image_name
@@ -315,36 +457,68 @@ def __validate_storage_folders(self) -> None:
         for folder in folders:
             folder.mkdir(parents=True, exist_ok=True)
 
-    def __recover_staged_deletes(self) -> None:
+    def __recover_pending_deletes(self) -> None:
+        """Reconciles every delete journal left behind by an interrupted or failed deletion.
+
+        One rule covers both journal shapes, and the record store decides it: an image whose record
+        survives was never really deleted, so anything staged for it is put back and the journal
+        dropped; an image whose record is gone is an orphan, so its files are purged wherever the
+        interrupted operation left them.
+        """
         logger = InvokeAILogger.get_logger()
-        for staging_dir in self.__output_folder.glob(".delete_*"):
-            manifest_path = staging_dir / "manifest.json"
-            if not manifest_path.is_file():
-                if not any(staging_dir.iterdir()):
-                    staging_dir.rmdir()
-                continue
+        for journal_dir in sorted(self.__output_folder.glob(".delete_*")):
             try:
+                manifest_path = journal_dir / "manifest.json"
+                if not manifest_path.is_file():
+                    # mkdtemp() ran but the manifest never landed, so this directory names nothing
+                    # and cannot be reconciled. Drop it when it is empty; otherwise say so, because
+                    # anything inside it is a staged file that can no longer be put back.
+                    if not journal_dir.is_dir():
+                        logger.warning(f"Ignoring unexpected entry in the outputs folder: {journal_dir}")
+                    elif not any(journal_dir.iterdir()):
+                        journal_dir.rmdir()
+                    else:
+                        logger.warning(f"Image deletion journal {journal_dir} has no manifest and cannot be recovered")
+                    continue
                 with open(manifest_path, encoding="utf-8") as manifest:
                     data = json.load(manifest)
-                image_name = data["image_name"]
-                image_subfolder = data.get("image_subfolder", "")
-                candidates = [
-                    self.get_path(image_name, image_subfolder=image_subfolder),
-                    self.get_path(image_name, thumbnail=True, image_subfolder=image_subfolder),
-                ]
-                token = _StagedDelete(
-                    directory=staging_dir,
-                    files=[(source, staging_dir / str(index)) for index, source in enumerate(candidates)],
-                )
-                self.__invoker.services.image_records.get(image_name)
-                self.rollback_delete(token)
+                records = self.__invoker.services.image_records
+                for image_name, image_subfolder in self.__manifest_images(data):
+                    if records.exists(image_name):
+                        # Put back whatever stage_delete() moved aside. Only a single-image journal
+                        # ever holds staged files, at indices 0 and 1; a pending-delete journal
+                        # moves nothing, so these lookups simply find nothing to restore.
+                        restored = False
+                        for index, source in enumerate(self.__delete_candidates(image_name, image_subfolder)):
+                            staged = journal_dir / str(index)
+                            if staged.exists():
+                                source.parent.mkdir(parents=True, exist_ok=True)
+                                staged.replace(source)
+                                restored = True
+                        # Re-check after a restore, for the same reason rollback_delete() does:
+                        # another Invoke sharing this output folder may have deleted the record
+                        # while the files sat staged, and its own purge found nothing to remove.
+                        # Every deleter purges only after its record is committed as gone, so a
+                        # record still present cannot have been purged before this restore, and a
+                        # record now absent means the files must go — this journal is the last
+                        # thing that can find them. A record-store fault here propagates and keeps
+                        # the journal, like every other lookup in this loop.
+                        if not restored or records.exists(image_name):
+                            continue
+                    self.__purge_files(image_name, image_subfolder)
+                shutil.rmtree(journal_dir, ignore_errors=True)
             except Exception as error:
-                from invokeai.app.services.image_records.image_records_common import ImageRecordNotFoundException
+                # Includes a record-store fault: leave the journal for the next startup rather than
+                # guess. Retrying is always safe; both branches above are idempotent.
+                logger.error(f"Failed to recover image deletion journal {journal_dir}: {error}")
 
-                if isinstance(error, ImageRecordNotFoundException):
-                    shutil.rmtree(staging_dir, ignore_errors=True)
-                else:
-                    logger.error(f"Failed to recover staged image deletion {staging_dir}: {error}")
+    @staticmethod
+    def __manifest_images(data: dict) -> list[tuple[str, str]]:
+        """Reads both journal shapes: a pending delete lists many images, a staged delete names one."""
+        entries = data.get("images")
+        if entries is None:
+            return [(data["image_name"], data.get("image_subfolder", ""))]
+        return [(entry["image_name"], entry.get("image_subfolder", "")) for entry in entries]
 
     def __get_cache(self, image_name: Path) -> Optional[PILImageType]:
         with self.__cache_lock:
diff --git a/invokeai/app/services/image_records/image_records_base.py b/invokeai/app/services/image_records/image_records_base.py
index 64a530b357a..e62345ece4f 100644
--- a/invokeai/app/services/image_records/image_records_base.py
+++ b/invokeai/app/services/image_records/image_records_base.py
@@ -81,8 +81,18 @@ def delete_many(self, image_names: list[str]) -> None:
         pass
 
     @abstractmethod
-    def delete_intermediates(self) -> list[tuple[str, str]]:
-        """Deletes all intermediate image records, returning a list of (image_name, image_subfolder) tuples."""
+    def get_intermediates(self) -> list[tuple[str, str]]:
+        """Gets all intermediate image records as (image_name, image_subfolder) tuples, without deleting them."""
+        pass
+
+    @abstractmethod
+    def delete_intermediates_by_names(self, image_names: list[str]) -> list[str]:
+        """Deletes the named image records, skipping any that are no longer intermediates.
+
+        Returns the names whose records this call actually removed. Names that were already gone, and
+        names whose records survive because they are no longer intermediates, are both excluded, so a
+        caller purges the files of exactly the returned names and touches nothing else.
+        """
         pass
 
     @abstractmethod
diff --git a/invokeai/app/services/image_records/image_records_sqlite.py b/invokeai/app/services/image_records/image_records_sqlite.py
index 2d68967c282..32dd2bef80d 100644
--- a/invokeai/app/services/image_records/image_records_sqlite.py
+++ b/invokeai/app/services/image_records/image_records_sqlite.py
@@ -23,6 +23,10 @@
 
 
 class SqliteImageRecordStorage(ImageRecordStorageBase):
+    # Conservative bound on bound parameters per statement. SQLITE_MAX_VARIABLE_NUMBER defaults to
+    # 999 on SQLite builds older than 3.32, and an image library can hold far more intermediates.
+    _MAX_SQL_VARIABLES = 500
+
     def __init__(self, db: SqliteDatabase) -> None:
         super().__init__()
         self._db = db
@@ -314,30 +318,57 @@ def get_intermediates_count(self, user_id: Optional[str] = None) -> int:
             count = cast(int, cursor.fetchone()[0])
         return count
 
-    def delete_intermediates(self) -> list[tuple[str, str]]:
-        """Deletes all intermediate image records.
+    def get_intermediates(self) -> list[tuple[str, str]]:
+        """Gets all intermediate image records without deleting them.
 
-        Returns a list of (image_name, image_subfolder) tuples for file cleanup.
+        Returns a list of (image_name, image_subfolder) tuples for staged file deletion.
         """
         with self._db.transaction() as cursor:
-            try:
-                cursor.execute(
-                    """--sql
-                    SELECT image_name, image_subfolder FROM images
-                    WHERE is_intermediate = TRUE;
-                    """
-                )
-                result = cast(list[sqlite3.Row], cursor.fetchall())
-                image_name_subfolder_pairs = [(r[0], r[1]) for r in result]
-                cursor.execute(
-                    """--sql
-                    DELETE FROM images
-                    WHERE is_intermediate = TRUE;
-                    """
-                )
-            except sqlite3.Error as e:
-                raise ImageRecordDeleteException from e
-        return image_name_subfolder_pairs
+            cursor.execute(
+                """--sql
+                SELECT image_name, image_subfolder FROM images
+                WHERE is_intermediate = TRUE;
+                """
+            )
+            result = cast(list[sqlite3.Row], cursor.fetchall())
+        return [(r[0], r[1]) for r in result]
+
+    def delete_intermediates_by_names(self, image_names: list[str]) -> list[str]:
+        """Deletes the named image records, skipping any that are no longer intermediates.
+
+        The ``is_intermediate`` predicate rides on the DELETE itself rather than on a preceding
+        SELECT, so an image promoted out of intermediate status keeps its record however the
+        promotion interleaves with this call. (Python's legacy sqlite3 transaction control opens a
+        transaction only before a write, so a SELECT here holds no read lock to rely on.)
+
+        Returns the names whose records this call actually removed. Names that were already gone, and
+        names whose records survive because they are no longer intermediates, are both excluded — the
+        caller purges the files of exactly the returned names and touches nothing else.
+        """
+        deleted: list[str] = []
+        try:
+            with self._db.transaction() as cursor:
+                # Chunked to stay under SQLITE_MAX_VARIABLE_NUMBER; every chunk runs inside the one
+                # transaction above.
+                for start in range(0, len(image_names), self._MAX_SQL_VARIABLES):
+                    chunk = image_names[start : start + self._MAX_SQL_VARIABLES]
+                    placeholders = ",".join("?" for _ in chunk)
+                    select_query = f"SELECT image_name FROM images WHERE image_name IN ({placeholders})"
+
+                    cursor.execute(select_query, chunk)
+                    present_before = {cast(str, r[0]) for r in cursor.fetchall()}
+                    cursor.execute(
+                        f"DELETE FROM images WHERE image_name IN ({placeholders}) AND is_intermediate = TRUE",
+                        chunk,
+                    )
+                    cursor.execute(select_query, chunk)
+                    present_after = {cast(str, r[0]) for r in cursor.fetchall()}
+
+                    deleted.extend(name for name in chunk if name in present_before and name not in present_after)
+        except sqlite3.Error as e:
+            # The try wraps the context manager so a failure in its commit is reported too.
+            raise ImageRecordDeleteException from e
+        return deleted
 
     def save(
         self,
diff --git a/invokeai/app/services/images/images_default.py b/invokeai/app/services/images/images_default.py
index 63b8f3d153f..8913d8470f7 100644
--- a/invokeai/app/services/images/images_default.py
+++ b/invokeai/app/services/images/images_default.py
@@ -106,25 +106,55 @@ def create(
             raise
         except ImageFileSaveException:
             self.__invoker.services.logger.error("Failed to save image file")
-            try:
-                self.__invoker.services.image_files.delete(image_name, image_subfolder=image_subfolder)
-            except Exception as cleanup_error:
-                self.__invoker.services.logger.error(
-                    f"Failed to clean up image files after save failure: {str(cleanup_error)}"
-                )
-            try:
-                # Deleting the record also removes any board association through the database
-                # foreign key cascade. Both cleanup operations are attempted independently.
-                self.__invoker.services.image_records.delete(image_name)
-            except Exception as cleanup_error:
-                self.__invoker.services.logger.error(
-                    f"Failed to clean up image record after save failure: {str(cleanup_error)}"
-                )
+            self.__clean_up_failed_save(image_name, image_subfolder)
             raise
         except Exception as e:
             self.__invoker.services.logger.error(f"Problem saving image record and file: {str(e)}")
             raise e
 
+    def __clean_up_failed_save(self, image_name: str, image_subfolder: str) -> None:
+        """Removes the half-created image left by a failed save, record first.
+
+        Record-then-files is the order every delete path uses, and it is load-bearing rather than
+        cosmetic: a concurrent deleter that has to roll back decides whether to restore an image's
+        files by asking whether its record is still there. Purging files while the record survives
+        would tell that deleter to put them back, stranding them once this cleanup finally removes
+        the record. The journal covers the window in between.
+        """
+        token: object | None = None
+        try:
+            token = self.__invoker.services.image_files.begin_delete([(image_name, image_subfolder)])
+        except Exception as cleanup_error:
+            self.__invoker.services.logger.error(
+                f"Failed to journal the cleanup of {image_name} after a save failure: {str(cleanup_error)}"
+            )
+        try:
+            # Deleting the record also removes any board association through the database foreign
+            # key cascade.
+            self.__invoker.services.image_records.delete(image_name)
+        except Exception as cleanup_error:
+            self.__invoker.services.logger.error(
+                f"Failed to clean up image record after save failure: {str(cleanup_error)}"
+            )
+            # The record survived, so the image is still referenced; its files must stay with it.
+            if token is not None:
+                try:
+                    self.__invoker.services.image_files.abandon_delete(token)
+                except Exception as journal_error:
+                    self.__invoker.services.logger.error(
+                        f"Failed to discard the delete journal for {image_name}: {str(journal_error)}"
+                    )
+            return
+        try:
+            if token is None:
+                self.__invoker.services.image_files.delete(image_name, image_subfolder=image_subfolder)
+            else:
+                self.__invoker.services.image_files.commit_delete(token)
+        except Exception as cleanup_error:
+            self.__invoker.services.logger.error(
+                f"Failed to clean up image files after save failure: {str(cleanup_error)}"
+            )
+
     def update(
         self,
         image_name: str,
@@ -290,11 +320,35 @@ def get_many(
             raise e
 
     def delete(self, image_name: str):
+        # Record first, files second, with a durable journal spanning the two. Deleting the record
+        # first means a database failure leaves the image completely intact, and the only state
+        # that can outlive this call is a file nothing references — which the journal lets startup
+        # recovery find and purge. Nothing is ever moved aside and put back, so a concurrent
+        # deleter of the same image cannot resurrect files whose record has already been removed.
         try:
             record = self.__invoker.services.image_records.get(image_name)
-            self.__invoker.services.image_files.delete(image_name, image_subfolder=record.image_subfolder)
-            self.__invoker.services.image_records.delete(image_name)
+            token = self.__invoker.services.image_files.begin_delete([(image_name, record.image_subfolder)])
+            try:
+                self.__invoker.services.image_records.delete(image_name)
+            except Exception:
+                # The image is still live: drop the journal and leave its files alone.
+                try:
+                    self.__invoker.services.image_files.abandon_delete(token)
+                except Exception as cleanup_error:
+                    self.__invoker.services.logger.error(
+                        f"Failed to discard the delete journal for {image_name}: {cleanup_error}"
+                    )
+                raise
+            try:
+                self.__invoker.services.image_files.commit_delete(token)
+            except Exception as cleanup_error:
+                # The record is committed as gone, so the delete succeeded. The journal stays
+                # behind and startup recovery purges the leftover files.
+                self.__invoker.services.logger.error(f"Failed to purge deleted image files: {cleanup_error}")
             self._on_deleted(image_name)
+        except ImageRecordNotFoundException:
+            # Already deleted by another request; nothing here failed, so nothing to log.
+            raise
         except ImageRecordDeleteException:
             self.__invoker.services.logger.error("Failed to delete image record")
             raise
@@ -361,21 +415,60 @@ def delete_images_on_board(self, board_id: str, user_id: Optional[str] = None) -
             raise e
 
     def delete_intermediates(self) -> int:
+        # Records first, files second, with a durable journal spanning the two. An earlier revision
+        # staged every file, then conditionally deleted the records, then restored the files of any
+        # image that had been promoted out of intermediate status mid-operation. That restore is
+        # unfixably racy: while a promoted image's files sit in a staging directory, a concurrent
+        # single-image or board delete can stage-empty (it finds no files to move) and then remove
+        # the record; the restore then puts the files back with no record referencing them and no
+        # journal to recover from — permanent orphans (JPPhoto, PR #9361).
+        #
+        # Deleting the records first removes that hazard: the conditional DELETE is atomic and
+        # reports exactly which rows it removed, and only the files of already-deleted rows are
+        # touched. A promoted image is never deleted and its files are never moved, so a concurrent
+        # delete of it operates on real files in the output folder and stays consistent. The
+        # journal covers the window the reordering opens: if this process dies, or the filesystem
+        # fails, between the commit and the purge, startup recovery finishes the purge for every
+        # journalled image whose record is gone.
         try:
-            image_name_subfolder_pairs = self.__invoker.services.image_records.delete_intermediates()
-            count = len(image_name_subfolder_pairs)
-            for image_name, image_subfolder in image_name_subfolder_pairs:
-                self.__invoker.services.image_files.delete(image_name, image_subfolder=image_subfolder)
+            image_name_subfolder_pairs = self.__invoker.services.image_records.get_intermediates()
+            if not image_name_subfolder_pairs:
+                return 0
+            subfolders = dict(image_name_subfolder_pairs)
+            token = self.__invoker.services.image_files.begin_delete(list(subfolders.items()))
+            try:
+                # Conditional on the row still being an intermediate: an image promoted between the
+                # snapshot above and this call keeps both its record and its files. Returns exactly
+                # the names this call removed (already-absent and promoted rows are excluded).
+                deleted_image_names = self.__invoker.services.image_records.delete_intermediates_by_names(
+                    list(subfolders.keys())
+                )
+            except Exception:
+                try:
+                    self.__invoker.services.image_files.abandon_delete(token)
+                except Exception as cleanup_error:
+                    self.__invoker.services.logger.error(
+                        f"Failed to discard the intermediates delete journal: {cleanup_error}"
+                    )
+                raise
+            try:
+                # Only the names whose records this call removed are purged; a promoted image keeps
+                # its files. The journal still lists it, which is harmless — recovery re-checks
+                # every entry against the record store and skips the ones that survived.
+                self.__invoker.services.image_files.commit_delete(token, image_names=deleted_image_names)
+            except Exception as cleanup_error:
+                # The records are committed as gone, so the deletion succeeded. A file that could
+                # not be purged keeps its journal entry and is retried at the next startup; it must
+                # neither fail the operation nor undo the committed deletions.
+                self.__invoker.services.logger.error(f"Failed to purge intermediate image files: {cleanup_error}")
+            for image_name in deleted_image_names:
                 self._on_deleted(image_name)
-            return count
+            return len(deleted_image_names)
         except ImageRecordDeleteException:
             self.__invoker.services.logger.error("Failed to delete image records")
             raise
-        except ImageFileDeleteException:
-            self.__invoker.services.logger.error("Failed to delete image files")
-            raise
         except Exception as e:
-            self.__invoker.services.logger.error("Problem deleting image records and files")
+            self.__invoker.services.logger.error("Problem deleting intermediate image records and files")
             raise e
 
     def get_intermediates_count(self, user_id: Optional[str] = None) -> int:
diff --git a/tests/app/routers/test_images.py b/tests/app/routers/test_images.py
index 85761472182..ba4bc84bfa9 100644
--- a/tests/app/routers/test_images.py
+++ b/tests/app/routers/test_images.py
@@ -228,6 +228,181 @@ def test_get_bulk_download_image_image_deleted_after_response(
     assert not (tmp_path / "test.zip").exists()
 
 
+# ── Transactional single-image deletion (DELETE /api/v1/images/i/{image_name}) ──
+
+
+def prepare_delete_image_test(monkeypatch: Any, mock_invoker: Invoker, tmp_path: Path):
+    """Wire the delete route to a real ImageService + real DiskImageFileStorage + real SQLite records."""
+    from invokeai.app.services.image_files.image_files_disk import DiskImageFileStorage
+
+    mock_deps = MockApiDependencies(mock_invoker)
+    monkeypatch.setattr("invokeai.app.api.routers.images.ApiDependencies", mock_deps)
+    monkeypatch.setattr("invokeai.app.api.routers._access.ApiDependencies", mock_deps)
+    monkeypatch.setattr("invokeai.app.api.routers.image_move_maintenance.ApiDependencies", mock_deps)
+    monkeypatch.setattr("invokeai.app.api.auth_dependencies.ApiDependencies", mock_deps)
+
+    mock_invoker.services.urls = MagicMock()
+    mock_invoker.services.urls.get_image_url.return_value = "http://localhost/img.png"
+
+    storage = DiskImageFileStorage(tmp_path / "outputs")
+    mock_invoker.services.image_files = storage
+    storage.start(mock_invoker)
+    mock_invoker.services.images.start(mock_invoker)
+    return storage
+
+
+def _save_deletable_image(mock_invoker: Invoker, storage, image_name: str) -> None:
+    from PIL import Image
+
+    from invokeai.app.services.image_records.image_records_common import ImageCategory, ResourceOrigin
+
+    mock_invoker.services.image_records.save(
+        image_name=image_name,
+        image_origin=ResourceOrigin.INTERNAL,
+        image_category=ImageCategory.GENERAL,
+        width=64,
+        height=64,
+        has_workflow=False,
+    )
+    storage.save(image=Image.new("RGB", (64, 64)), image_name=image_name)
+
+
+def test_delete_image_success_deletes_files_and_record(
+    monkeypatch: Any, mock_invoker: Invoker, tmp_path: Path, client: TestClient
+) -> None:
+    from invokeai.app.services.image_records.image_records_common import ImageRecordNotFoundException
+
+    storage = prepare_delete_image_test(monkeypatch, mock_invoker, tmp_path)
+    _save_deletable_image(mock_invoker, storage, "del.png")
+
+    response = client.delete("/api/v1/images/i/del.png")
+
+    assert response.status_code == 200
+    json_response = response.json()
+    assert json_response["deleted_images"] == ["del.png"]
+    assert json_response["affected_boards"] == ["none"]
+    assert not storage.get_path("del.png").exists()
+    assert not storage.get_path("del.png", thumbnail=True).exists()
+    with pytest.raises(ImageRecordNotFoundException):
+        mock_invoker.services.image_records.get("del.png")
+    assert list(storage.image_root.glob(".delete_*")) == []
+
+
+def test_delete_image_not_found_returns_404(
+    monkeypatch: Any, mock_invoker: Invoker, tmp_path: Path, client: TestClient
+) -> None:
+    prepare_delete_image_test(monkeypatch, mock_invoker, tmp_path)
+
+    response = client.delete("/api/v1/images/i/does-not-exist.png")
+
+    assert response.status_code == 404
+    assert response.json()["detail"] == "Image not found"
+
+
+def test_delete_image_deleted_mid_request_returns_404(
+    monkeypatch: Any, mock_invoker: Invoker, tmp_path: Path, client: TestClient
+) -> None:
+    """An image deleted between the DTO lookup and the service delete is gone, not a server fault.
+
+    Answering 500 here sent the client a failure toast for a postcondition that already held
+    (JPPhoto, PR #9361 round 4).
+    """
+    storage = prepare_delete_image_test(monkeypatch, mock_invoker, tmp_path)
+    _save_deletable_image(mock_invoker, storage, "del.png")
+    real_get_dto = mock_invoker.services.images.get_dto
+
+    def get_dto_then_lose_the_race(image_name: str):
+        dto = real_get_dto(image_name)
+        # Another request completes its delete before this one reaches the service.
+        mock_invoker.services.image_records.delete(image_name)
+        return dto
+
+    monkeypatch.setattr(mock_invoker.services.images, "get_dto", get_dto_then_lose_the_race)
+
+    response = client.delete("/api/v1/images/i/del.png")
+
+    assert response.status_code == 404
+    assert response.json()["detail"] == "Image not found"
+    assert list(storage.image_root.glob(".delete_*")) == []
+
+
+def test_delete_image_lookup_failure_returns_500_not_404(
+    monkeypatch: Any, mock_invoker: Invoker, tmp_path: Path, client: TestClient
+) -> None:
+    """A DTO lookup that fails for a reason other than a missing record is a 500, not a 404.
+
+    Reporting it as 404 would tell the frontend the image is gone and drop a live item from its cache.
+    """
+    storage = prepare_delete_image_test(monkeypatch, mock_invoker, tmp_path)
+    _save_deletable_image(mock_invoker, storage, "del.png")
+
+    def failing_get_dto(image_name: str):
+        raise RuntimeError("database unavailable")
+
+    monkeypatch.setattr(mock_invoker.services.images, "get_dto", failing_get_dto)
+
+    response = client.delete("/api/v1/images/i/del.png")
+
+    assert response.status_code == 500
+    assert response.json()["detail"] == "Failed to delete image"
+    # Nothing was touched: the record and its files are intact.
+    assert storage.get_path("del.png").exists()
+    assert mock_invoker.services.image_records.get("del.png").image_name == "del.png"
+
+
+def test_delete_image_db_fault_during_lookup_returns_500_not_404(
+    monkeypatch: Any, mock_invoker: Invoker, tmp_path: Path, client: TestClient
+) -> None:
+    """A database fault while reading the record is a 500, driven through the real record store.
+
+    The store used to convert every ``sqlite3.Error`` into ``ImageRecordNotFoundException``, which
+    made a database fault indistinguishable from a missing image and produced a 404 for a live one.
+    This drives the real store rather than stubbing it, so the store's translation is what is under
+    test — stubbing ``get`` would bypass the very code that used to be wrong.
+    """
+    storage = prepare_delete_image_test(monkeypatch, mock_invoker, tmp_path)
+    _save_deletable_image(mock_invoker, storage, "del.png")
+
+    # Break the table out from under the query. Any sqlite3.Error would do; this one is deterministic.
+    records = mock_invoker.services.image_records
+    records._db._conn.execute("ALTER TABLE images RENAME TO images_moved;")
+    try:
+        response = client.delete("/api/v1/images/i/del.png")
+    finally:
+        records._db._conn.execute("ALTER TABLE images_moved RENAME TO images;")
+
+    assert response.status_code == 500
+    assert response.json()["detail"] == "Failed to delete image"
+    # The image is still there once the database recovers.
+    assert records.get("del.png").image_name == "del.png"
+    assert storage.get_path("del.png").exists()
+
+
+def test_delete_image_db_failure_returns_500_and_restores_files(
+    monkeypatch: Any, mock_invoker: Invoker, tmp_path: Path, client: TestClient
+) -> None:
+    from invokeai.app.services.image_records.image_records_common import ImageRecordDeleteException
+
+    storage = prepare_delete_image_test(monkeypatch, mock_invoker, tmp_path)
+    _save_deletable_image(mock_invoker, storage, "del.png")
+
+    def failing_delete(image_name: str) -> None:
+        raise ImageRecordDeleteException()
+
+    monkeypatch.setattr(mock_invoker.services.image_records, "delete", failing_delete)
+
+    response = client.delete("/api/v1/images/i/del.png")
+
+    # The route must report the failure, not a success-shaped empty payload.
+    assert response.status_code == 500
+    assert response.json()["detail"] == "Failed to delete image"
+    # The staged files must be rolled back: image and thumbnail restored, record intact.
+    assert storage.get_path("del.png").exists()
+    assert storage.get_path("del.png", thumbnail=True).exists()
+    assert mock_invoker.services.image_records.get("del.png").image_name == "del.png"
+    assert list(storage.image_root.glob(".delete_*")) == []
+
+
 def prepare_image_batch_test(monkeypatch: Any, mock_invoker: Invoker) -> MagicMock:
     """Wires the image router to a MagicMock image service with maintenance inactive.
 
diff --git a/tests/app/routers/test_multiuser_authorization.py b/tests/app/routers/test_multiuser_authorization.py
index adf422f9291..cce5b7d3965 100644
--- a/tests/app/routers/test_multiuser_authorization.py
+++ b/tests/app/routers/test_multiuser_authorization.py
@@ -1500,11 +1500,18 @@ def test_non_owner_can_delete_image_from_public_board(
         _save_image(mock_invoker, "user1-public-delete", user1.user_id)
         mock_invoker.services.board_image_records.add_image_to_board(public_board_id, "user1-public-delete")
 
+        # The delete route no longer swallows service failures, so the test env needs
+        # working urls/image_files services for the deletion to actually succeed.
+        mock_invoker.services.urls = MagicMock()
+        mock_invoker.services.urls.get_image_url.return_value = "http://localhost/img.png"
+        mock_invoker.services.image_files = MagicMock()
+
         r = client.delete(
             "/api/v1/images/i/user1-public-delete",
             headers=_auth(user2_token),
         )
         assert r.status_code == status.HTTP_200_OK
+        assert r.json()["deleted_images"] == ["user1-public-delete"]
 
     def test_clear_intermediates_non_admin_forbidden(self, client: TestClient, user1_token: str):
         r = client.delete("/api/v1/images/intermediates", headers=_auth(user1_token))
diff --git a/tests/app/services/image_files/test_image_files_disk.py b/tests/app/services/image_files/test_image_files_disk.py
index caccb347397..9aaa8a57a3b 100644
--- a/tests/app/services/image_files/test_image_files_disk.py
+++ b/tests/app/services/image_files/test_image_files_disk.py
@@ -7,12 +7,20 @@
 import pytest
 from PIL import Image
 
-from invokeai.app.services.image_files.image_files_common import ImageFileSaveException
+from invokeai.app.services.image_files.image_files_common import ImageFileDeleteException, ImageFileSaveException
 from invokeai.app.services.image_files.image_files_disk import DiskImageFileStorage, _should_use_png_rle
-from invokeai.app.services.image_records.image_records_common import ImageRecordNotFoundException
 from invokeai.app.util.thumbnails import get_thumbnail_name
 
 
+def _restart(storage: DiskImageFileStorage, record_exists: bool) -> DiskImageFileStorage:
+    """Simulates a restart over the same output folder, running journal recovery."""
+    invoker = MagicMock()
+    invoker.services.image_records.exists.return_value = record_exists
+    restarted = DiskImageFileStorage(storage.image_root)
+    restarted.start(invoker)
+    return restarted
+
+
 @pytest.fixture
 def image_names() -> list[str]:
     # Determine the platform and return a path that matches its format
@@ -42,6 +50,9 @@ def disk_storage(tmp_path: Path) -> DiskImageFileStorage:
     # Mock the invoker for save() which needs compress_level
     mock_invoker = MagicMock()
     mock_invoker.services.configuration.pil_compress_level = 6
+    # Deletion asks the record store whether an image is still referenced; say yes unless a test
+    # says otherwise, so nothing here depends on a bare MagicMock happening to be truthy.
+    mock_invoker.services.image_records.exists.return_value = True
     storage._DiskImageFileStorage__invoker = mock_invoker  # type: ignore
     return storage
 
@@ -359,23 +370,302 @@ def test_startup_restores_staged_files_when_record_exists(self, disk_storage: Di
         image_name = "recover.png"
         disk_storage.save(image=Image.new("RGB", (32, 32)), image_name=image_name)
         image_path = disk_storage.get_path(image_name)
+        thumbnail_path = disk_storage.get_path(image_name, thumbnail=True)
         disk_storage.stage_delete(image_name)
 
-        invoker = MagicMock()
-        invoker.services.image_records.get.return_value = object()
-        restarted = DiskImageFileStorage(disk_storage.image_root)
-        restarted.start(invoker)
+        _restart(disk_storage, record_exists=True)
 
         assert image_path.exists()
+        assert thumbnail_path.exists()
+        assert not list(disk_storage.image_root.glob(".delete_*"))
 
     def test_startup_purges_staged_files_when_record_was_deleted(self, disk_storage: DiskImageFileStorage):
         image_name = "purge.png"
         disk_storage.save(image=Image.new("RGB", (32, 32)), image_name=image_name)
+        image_path = disk_storage.get_path(image_name)
+        disk_storage.stage_delete(image_name)
+
+        _restart(disk_storage, record_exists=False)
+
+        assert not image_path.exists()
+        assert not list(disk_storage.image_root.glob(".delete_*"))
+
+    def test_startup_leaves_the_journal_when_the_record_store_is_unreadable(self, disk_storage: DiskImageFileStorage):
+        """A database fault must not decide an image's fate; the journal is retried next startup."""
+        image_name = "unreadable.png"
+        disk_storage.save(image=Image.new("RGB", (32, 32)), image_name=image_name)
+        disk_storage.stage_delete(image_name)
+
+        invoker = MagicMock()
+        invoker.services.image_records.exists.side_effect = RuntimeError("database is locked")
+        restarted = DiskImageFileStorage(disk_storage.image_root)
+        restarted.start(invoker)
+
+        assert list(disk_storage.image_root.glob(".delete_*"))
+
+    def test_startup_purges_restored_files_whose_record_vanished_during_recovery(
+        self, disk_storage: DiskImageFileStorage
+    ):
+        """Recovery re-checks the record after restoring, exactly as rollback_delete() does.
+
+        Another Invoke sharing the output folder can delete the record while the files sit staged;
+        its purge finds nothing, and restoring them afterwards would strand them with no record and
+        no journal left to find them by (JPPhoto, PR #9361 round 4).
+        """
+        image_name = "raced.png"
+        disk_storage.save(image=Image.new("RGB", (32, 32)), image_name=image_name)
+        disk_storage.stage_delete(image_name)
+
+        invoker = MagicMock()
+        # Present when recovery first looks, gone by the time the files are back.
+        invoker.services.image_records.exists.side_effect = [True, False]
+        restarted = DiskImageFileStorage(disk_storage.image_root)
+        restarted.start(invoker)
+
+        assert not disk_storage.get_path(image_name).exists()
+        assert not disk_storage.get_path(image_name, thumbnail=True).exists()
+        assert not list(disk_storage.image_root.glob(".delete_*"))
+
+    def test_startup_keeps_the_journal_when_the_recheck_cannot_read_the_record_store(
+        self, disk_storage: DiskImageFileStorage
+    ):
+        """A fault on the post-restore re-check keeps the files and the journal for the next start."""
+        image_name = "kept.png"
+        disk_storage.save(image=Image.new("RGB", (32, 32)), image_name=image_name)
         disk_storage.stage_delete(image_name)
 
         invoker = MagicMock()
-        invoker.services.image_records.get.side_effect = ImageRecordNotFoundException
+        invoker.services.image_records.exists.side_effect = [True, RuntimeError("database is locked")]
+        restarted = DiskImageFileStorage(disk_storage.image_root)
+        restarted.start(invoker)
+
+        assert disk_storage.get_path(image_name).exists()
+        assert disk_storage.get_path(image_name, thumbnail=True).exists()
+        assert list(disk_storage.image_root.glob(".delete_*"))
+
+    def test_startup_does_not_recheck_a_pending_journal_that_restored_nothing(self, disk_storage: DiskImageFileStorage):
+        """Only a restore can strand files; a journal that moved nothing costs one lookup."""
+        disk_storage.save(image=Image.new("RGB", (32, 32)), image_name="pending.png")
+        disk_storage.begin_delete([("pending.png", "")])
+
+        invoker = MagicMock()
+        invoker.services.image_records.exists.side_effect = [True, AssertionError("unexpected re-check")]
+        restarted = DiskImageFileStorage(disk_storage.image_root)
+        restarted.start(invoker)
+
+        assert disk_storage.get_path("pending.png").exists()
+        assert not list(disk_storage.image_root.glob(".delete_*"))
+
+
+class TestPendingDeleteJournal:
+    """begin_delete() writes the journal that makes records-first deletion recoverable.
+
+    Nothing is moved, so a failure can only ever leave files nothing references — and the journal
+    is what lets the next startup find and purge exactly those.
+    """
+
+    def test_begin_delete_leaves_the_files_in_place(self, disk_storage: DiskImageFileStorage):
+        disk_storage.save(image=Image.new("RGB", (32, 32)), image_name="live.png")
+
+        disk_storage.begin_delete([("live.png", "")])
+
+        assert disk_storage.get_path("live.png").exists()
+        assert disk_storage.get_path("live.png", thumbnail=True).exists()
+        assert len(list(disk_storage.image_root.glob(".delete_*"))) == 1
+
+    def test_begin_delete_rejects_an_unusable_name_before_writing_a_journal(
+        self, disk_storage: DiskImageFileStorage, tmp_path: Path
+    ):
+        """The caller deletes records straight after this returns, so a bad name must fail here."""
+        with pytest.raises(ValueError, match="Invalid image name"):
+            disk_storage.begin_delete([("ok.png", ""), ("../evil.png", "")])
+
+        assert not list(tmp_path.glob(".delete_*"))
+
+    def test_commit_purges_the_files_and_drops_the_journal(self, disk_storage: DiskImageFileStorage):
+        disk_storage.save(image=Image.new("RGB", (32, 32)), image_name="gone.png")
+        token = disk_storage.begin_delete([("gone.png", "")])
+
+        disk_storage.commit_delete(token)
+
+        assert not disk_storage.get_path("gone.png").exists()
+        assert not disk_storage.get_path("gone.png", thumbnail=True).exists()
+        assert not list(disk_storage.image_root.glob(".delete_*"))
+
+    def test_commit_purges_only_the_named_images(self, disk_storage: DiskImageFileStorage):
+        """The journal lists every candidate; only the records that were really deleted are purged."""
+        for name in ("deleted.png", "promoted.png"):
+            disk_storage.save(image=Image.new("RGB", (32, 32)), image_name=name)
+        token = disk_storage.begin_delete([("deleted.png", ""), ("promoted.png", "")])
+
+        disk_storage.commit_delete(token, image_names=["deleted.png"])
+
+        assert not disk_storage.get_path("deleted.png").exists()
+        assert disk_storage.get_path("promoted.png").exists()
+        assert disk_storage.get_path("promoted.png", thumbnail=True).exists()
+
+    def test_commit_keeps_the_journal_when_a_file_cannot_be_purged(self, disk_storage: DiskImageFileStorage):
+        """One unremovable file must not abort the other purges, and must not discard the journal:
+        the entry has to survive so the next startup can retry it."""
+        for name in ("bad.png", "good.png"):
+            disk_storage.save(image=Image.new("RGB", (32, 32)), image_name=name)
+        token = disk_storage.begin_delete([("bad.png", ""), ("good.png", "")])
+        bad_path = disk_storage.get_path("bad.png")
+        real_unlink = Path.unlink
+
+        def unlink(self: Path, missing_ok: bool = False):
+            if self == bad_path:
+                raise OSError("device busy")
+            return real_unlink(self, missing_ok=missing_ok)
+
+        with patch.object(Path, "unlink", unlink), pytest.raises(ImageFileDeleteException):
+            disk_storage.commit_delete(token)
+
+        assert bad_path.exists()
+        # The failure did not stop the rest of the purge...
+        assert not disk_storage.get_path("good.png").exists()
+        # ...and the journal is still there for startup recovery to finish.
+        assert list(disk_storage.image_root.glob(".delete_*"))
+
+        _restart(disk_storage, record_exists=False)
+
+        assert not bad_path.exists()
+        assert not list(disk_storage.image_root.glob(".delete_*"))
+
+    def test_abandon_keeps_the_files_and_drops_the_journal(self, disk_storage: DiskImageFileStorage):
+        """The record delete failed, so the image is still live and must be left completely alone."""
+        disk_storage.save(image=Image.new("RGB", (32, 32)), image_name="kept.png")
+        token = disk_storage.begin_delete([("kept.png", "")])
+
+        disk_storage.abandon_delete(token)
+
+        assert disk_storage.get_path("kept.png").exists()
+        assert disk_storage.get_path("kept.png", thumbnail=True).exists()
+        assert not list(disk_storage.image_root.glob(".delete_*"))
+
+    def test_startup_purges_journalled_files_whose_record_is_gone(self, disk_storage: DiskImageFileStorage):
+        """The crash window records-first opens: records committed as deleted, purge never ran."""
+        for name in ("orphan.png", "survivor.png"):
+            disk_storage.save(image=Image.new("RGB", (32, 32)), image_name=name)
+        disk_storage.begin_delete([("orphan.png", ""), ("survivor.png", "")])
+
+        invoker = MagicMock()
+        invoker.services.image_records.exists.side_effect = lambda name: name == "survivor.png"
         restarted = DiskImageFileStorage(disk_storage.image_root)
         restarted.start(invoker)
 
+        assert not disk_storage.get_path("orphan.png").exists()
+        assert not disk_storage.get_path("orphan.png", thumbnail=True).exists()
+        # The record survived, so this one was never deleted and keeps its files.
+        assert disk_storage.get_path("survivor.png").exists()
+        assert not list(disk_storage.image_root.glob(".delete_*"))
+
+    def test_rollback_purges_instead_of_restoring_when_the_record_is_gone(self, disk_storage: DiskImageFileStorage):
+        """A staged delete that fails must not resurrect files another request has already
+        unreferenced. Restoring them would strand them with no record and no journal to find
+        them by (JPPhoto, PR #9361)."""
+        disk_storage.save(image=Image.new("RGB", (32, 32)), image_name="raced.png")
+        token = disk_storage.stage_delete("raced.png")
+        # Meanwhile another request deleted the record.
+        disk_storage._DiskImageFileStorage__invoker.services.image_records.exists.return_value = False
+
+        disk_storage.rollback_delete(token)
+
+        assert not disk_storage.get_path("raced.png").exists()
+        assert not disk_storage.get_path("raced.png", thumbnail=True).exists()
+        assert not list(disk_storage.image_root.glob(".delete_*"))
+
+    def test_rollback_restores_when_the_record_store_cannot_be_read(self, disk_storage: DiskImageFileStorage):
+        """An unreadable database must never cost a live image its files."""
+        disk_storage.save(image=Image.new("RGB", (32, 32)), image_name="kept.png")
+        token = disk_storage.stage_delete("kept.png")
+        records = disk_storage._DiskImageFileStorage__invoker.services.image_records
+        records.exists.side_effect = RuntimeError("database is locked")
+
+        disk_storage.rollback_delete(token)
+
+        assert disk_storage.get_path("kept.png").exists()
+        assert disk_storage.get_path("kept.png", thumbnail=True).exists()
+
+
+class TestJournalDurability:
+    """The journal only makes a deletion recoverable if it outlives a power loss.
+
+    SQLite fsyncs the record deletion, so a journal that is merely written — and not fsynced, both
+    its manifest and its own directory entry in the output folder — can be lost while the record
+    stays deleted, which is exactly the orphan the journal exists to prevent.
+    """
+
+    def _record_fsyncs(self, monkeypatch) -> list[Path]:
+        fsynced: list[Path] = []
+        monkeypatch.setattr(
+            DiskImageFileStorage,
+            "_DiskImageFileStorage__fsync_directory",
+            staticmethod(lambda directory: fsynced.append(Path(directory))),
+        )
+        return fsynced
+
+    def test_begin_delete_fsyncs_the_journal_and_the_output_folder(
+        self, disk_storage: DiskImageFileStorage, monkeypatch
+    ):
+        fsynced = self._record_fsyncs(monkeypatch)
+
+        token = disk_storage.begin_delete([("img.png", "")])
+
+        assert Path(token.directory) in fsynced
+        assert disk_storage.image_root in [path.resolve() for path in fsynced]
+
+    def test_stage_delete_fsyncs_before_it_moves_the_files(self, disk_storage: DiskImageFileStorage, monkeypatch):
+        """The manifest has to be durable first, or a crash leaves staged files naming nothing."""
+        disk_storage.save(image=Image.new("RGB", (32, 32)), image_name="staged.png")
+        moved: list[str] = []
+        fsynced: list[Path] = []
+
+        def record_fsync(directory):
+            fsynced.append(Path(directory))
+
+        real_replace = Path.replace
+
+        def record_replace(self: Path, target):
+            moved.append(str(target))
+            assert fsynced, "the manifest was not made durable before the files were moved"
+            return real_replace(self, target)
+
+        monkeypatch.setattr(DiskImageFileStorage, "_DiskImageFileStorage__fsync_directory", staticmethod(record_fsync))
+        with patch.object(Path, "replace", record_replace):
+            token = disk_storage.stage_delete("staged.png")
+
+        assert moved
+        assert Path(token.directory) in fsynced
+        assert disk_storage.image_root in [path.resolve() for path in fsynced]
+
+
+class TestRecoveryToleratesStrayEntries:
+    """Recovery runs during start(); anything it cannot make sense of must not stop the app."""
+
+    def test_a_stray_file_does_not_stop_startup(self, disk_storage: DiskImageFileStorage):
+        disk_storage.save(image=Image.new("RGB", (32, 32)), image_name="live.png")
+        (disk_storage.image_root / ".delete_stray").write_text("not a directory")
+
+        _restart(disk_storage, record_exists=True)
+
+        assert disk_storage.get_path("live.png").exists()
+
+    def test_a_journal_with_no_manifest_is_left_alone(self, disk_storage: DiskImageFileStorage):
+        """Its contents cannot be attributed to an image, so removing them would destroy data."""
+        disk_storage.save(image=Image.new("RGB", (32, 32)), image_name="live.png")
+        orphan_journal = disk_storage.image_root / ".delete_nomanifest"
+        orphan_journal.mkdir()
+        (orphan_journal / "0").write_bytes(b"staged image bytes")
+
+        _restart(disk_storage, record_exists=True)
+
+        assert (orphan_journal / "0").read_bytes() == b"staged image bytes"
+        assert disk_storage.get_path("live.png").exists()
+
+    def test_an_empty_journal_directory_is_removed(self, disk_storage: DiskImageFileStorage):
+        (disk_storage.image_root / ".delete_empty").mkdir()
+
+        _restart(disk_storage, record_exists=True)
+
         assert not list(disk_storage.image_root.glob(".delete_*"))
diff --git a/tests/app/services/image_records/test_image_records_sqlite.py b/tests/app/services/image_records/test_image_records_sqlite.py
index 7408dcf3761..62d7c58f5cf 100644
--- a/tests/app/services/image_records/test_image_records_sqlite.py
+++ b/tests/app/services/image_records/test_image_records_sqlite.py
@@ -1,16 +1,23 @@
 """DB-backed tests for SqliteImageRecordStorage.
 
 Verifies that image_subfolder round-trips correctly through save(), get(),
-get_many(), and delete_intermediates() against a real (in-memory) SQLite database,
+get_many(), and get_intermediates() against a real (in-memory) SQLite database,
 and that get_many()/get_image_names() enforce per-user ownership isolation.
 """
 
+import sqlite3
+
 import pytest
 
 from invokeai.app.services.board_image_records.board_image_records_sqlite import SqliteBoardImageRecordStorage
 from invokeai.app.services.board_records.board_records_sqlite import SqliteBoardRecordStorage
 from invokeai.app.services.config.config_default import InvokeAIAppConfig
-from invokeai.app.services.image_records.image_records_common import ImageCategory, ResourceOrigin
+from invokeai.app.services.image_records.image_records_common import (
+    ImageCategory,
+    ImageRecordChanges,
+    ImageRecordNotFoundException,
+    ResourceOrigin,
+)
 from invokeai.app.services.image_records.image_records_sqlite import SqliteImageRecordStorage
 from invokeai.app.services.shared.sqlite.sqlite_common import SQLiteDirection
 from invokeai.backend.util.logging import InvokeAILogger
@@ -106,15 +113,15 @@ def test_get_many_returns_subfolders(self, store: SqliteImageRecordStorage) -> N
         assert by_name["hashed.png"] == "ab"
 
 
-class TestDeleteIntermediatesSubfolder:
-    """delete_intermediates() returns (name, subfolder) pairs and removes rows."""
+class TestGetIntermediatesSubfolder:
+    """get_intermediates() returns (name, subfolder) pairs without deleting rows."""
 
     def test_returns_subfolder_pairs(self, store: SqliteImageRecordStorage) -> None:
         _save(store, "keep.png", subfolder="general", is_intermediate=False)
         _save(store, "tmp1.png", subfolder="intermediate", is_intermediate=True)
         _save(store, "tmp2.png", subfolder="intermediate", is_intermediate=True)
 
-        pairs = store.delete_intermediates()
+        pairs = store.get_intermediates()
 
         # Should return only intermediate images with their subfolders
         assert len(pairs) == 2
@@ -126,16 +133,166 @@ def test_returns_subfolder_pairs(self, store: SqliteImageRecordStorage) -> None:
         record = store.get("keep.png")
         assert record.image_subfolder == "general"
 
-    def test_intermediates_are_deleted(self, store: SqliteImageRecordStorage) -> None:
+    def test_get_intermediates_does_not_delete(self, store: SqliteImageRecordStorage) -> None:
         _save(store, "tmp.png", subfolder="x", is_intermediate=True)
-        store.delete_intermediates()
+        store.get_intermediates()
+
+        # Listing intermediates must not remove them.
+        record = store.get("tmp.png")
+        assert record.image_subfolder == "x"
 
-        from invokeai.app.services.image_records.image_records_common import ImageRecordNotFoundException
+    def test_intermediates_are_deleted_via_delete_intermediates_by_names(self, store: SqliteImageRecordStorage) -> None:
+        _save(store, "tmp.png", subfolder="x", is_intermediate=True)
+        pairs = store.get_intermediates()
+        deleted = store.delete_intermediates_by_names([name for name, _ in pairs])
 
+        assert deleted == ["tmp.png"]
         with pytest.raises(ImageRecordNotFoundException):
             store.get("tmp.png")
 
 
+class TestQueryFaultsAreNotNotFound:
+    """A failing query means the database is unavailable, not that the image is missing.
+
+    Reporting a query fault as "not found" propagates all the way to the API, where it becomes a 404
+    and tells the frontend to drop a live image from its cache.
+    """
+
+    def _break_the_images_table(self, store: SqliteImageRecordStorage) -> None:
+        store._db._conn.execute("ALTER TABLE images RENAME TO images_moved;")
+
+    def test_get_raises_the_db_error_not_not_found(self, store: SqliteImageRecordStorage) -> None:
+        _save(store, "live.png")
+        self._break_the_images_table(store)
+
+        with pytest.raises(sqlite3.Error):
+            store.get("live.png")
+
+    def test_get_metadata_raises_the_db_error_not_not_found(self, store: SqliteImageRecordStorage) -> None:
+        _save(store, "live.png")
+        self._break_the_images_table(store)
+
+        with pytest.raises(sqlite3.Error):
+            store.get_metadata("live.png")
+
+    def test_missing_row_still_raises_not_found(self, store: SqliteImageRecordStorage) -> None:
+        """The genuine not-found path is untouched."""
+        with pytest.raises(ImageRecordNotFoundException):
+            store.get("never-existed.png")
+        with pytest.raises(ImageRecordNotFoundException):
+            store.get_metadata("never-existed.png")
+
+
+class TestDeleteIntermediatesByNames:
+    """delete_intermediates_by_names() deletes only rows that are still intermediates."""
+
+    def test_promoted_image_keeps_its_record(self, store: SqliteImageRecordStorage) -> None:
+        """An image promoted out of intermediate status after the snapshot must survive."""
+        _save(store, "tmp.png", subfolder="x", is_intermediate=True)
+        _save(store, "promoted.png", subfolder="x", is_intermediate=True)
+        snapshot = [name for name, _ in store.get_intermediates()]
+        assert set(snapshot) == {"tmp.png", "promoted.png"}
+
+        # Simulate the race: the image stops being an intermediate between the snapshot and delete.
+        store.update("promoted.png", ImageRecordChanges(is_intermediate=False))
+
+        deleted = store.delete_intermediates_by_names(snapshot)
+
+        assert deleted == ["tmp.png"]
+        # promoted.png is excluded from the returned names, so the caller never purges its files.
+        assert store.get("promoted.png").is_intermediate is False
+        with pytest.raises(ImageRecordNotFoundException):
+            store.get("tmp.png")
+
+    def test_promotion_interleaved_inside_the_call_keeps_the_record(self, store: SqliteImageRecordStorage) -> None:
+        """The is_intermediate predicate must ride on the DELETE, not on a preceding SELECT.
+
+        Python's legacy sqlite3 transaction control opens a transaction only before a write, so a
+        SELECT inside this method holds no read lock. A writer that promotes an image after that
+        SELECT but before the DELETE must still not lose its record.
+        """
+        _save(store, "tmp.png", is_intermediate=True)
+        _save(store, "promoted.png", is_intermediate=True)
+        snapshot = [name for name, _ in store.get_intermediates()]
+
+        # Promote from inside the call, between the first SELECT and the DELETE.
+        real_execute = store._db._conn.execute
+        promoted = False
+
+        def trace(statement: str) -> None:
+            nonlocal promoted
+            # The trace fires when a statement *begins*, so hooking the first SELECT would promote
+            # before that SELECT reads anything — indistinguishable from promoting up front. Hooking
+            # the DELETE puts the promotion after the SELECT has already seen the row as an
+            # intermediate, which is the interleaving that a SELECT-then-unconditional-DELETE
+            # implementation gets wrong.
+            if not promoted and statement.strip().upper().startswith("DELETE FROM IMAGES"):
+                promoted = True
+                real_execute("UPDATE images SET is_intermediate = 0 WHERE image_name = 'promoted.png'")
+
+        store._db._conn.set_trace_callback(trace)
+        try:
+            deleted = store.delete_intermediates_by_names(snapshot)
+        finally:
+            store._db._conn.set_trace_callback(None)
+
+        assert promoted, "the interleaved promotion never ran; the test proves nothing"
+        assert deleted == ["tmp.png"]
+        assert store.get("promoted.png").is_intermediate is False
+
+    def test_unknown_and_empty_names_are_no_ops(self, store: SqliteImageRecordStorage) -> None:
+        _save(store, "keep.png", is_intermediate=False)
+
+        assert store.delete_intermediates_by_names([]) == []
+        # "gone.png" has no record at all and "keep.png" is not an intermediate, so neither is
+        # deleted or returned; keep.png must still be present afterwards.
+        assert store.delete_intermediates_by_names(["gone.png", "keep.png"]) == []
+        assert store.get("keep.png").image_name == "keep.png"
+
+    def test_more_names_than_sql_variable_limit(self, store: SqliteImageRecordStorage) -> None:
+        """Chunking must not lose rows: exercise a name list spanning several chunks."""
+        chunk = SqliteImageRecordStorage._MAX_SQL_VARIABLES
+        names = [f"tmp{i:05d}.png" for i in range(chunk * 2 + 7)]
+        for name in names:
+            _save(store, name, is_intermediate=True)
+        # One image in the middle chunk is promoted and must survive.
+        survivor = names[chunk + 3]
+        store.update(survivor, ImageRecordChanges(is_intermediate=False))
+
+        deleted = store.delete_intermediates_by_names(names)
+
+        assert set(deleted) == set(names) - {survivor}
+        assert survivor not in deleted
+        assert store.get(survivor).is_intermediate is False
+        assert store.get_intermediates() == []
+
+    def test_chunking_stays_within_the_declared_variable_limit(self, store: SqliteImageRecordStorage) -> None:
+        """No statement may bind more parameters than the declared limit."""
+        chunk = SqliteImageRecordStorage._MAX_SQL_VARIABLES
+        names = [f"tmp{i:05d}.png" for i in range(chunk * 2 + 7)]
+        for name in names:
+            _save(store, name, is_intermediate=True)
+
+        # The trace callback reports statements with their parameters already expanded, so count the
+        # bound image names in each one rather than the placeholders.
+        widest = 0
+
+        def trace(statement: str) -> None:
+            nonlocal widest
+            if "images WHERE image_name IN (" in statement:
+                widest = max(widest, statement.count(".png"))
+
+        store._db._conn.set_trace_callback(trace)
+        try:
+            store.delete_intermediates_by_names(names)
+        finally:
+            store._db._conn.set_trace_callback(None)
+
+        # 999 is the SQLITE_MAX_VARIABLE_NUMBER default on builds older than 3.32. Asserting the
+        # literal rather than _MAX_SQL_VARIABLES keeps the test meaningful if that constant is raised.
+        assert 0 < widest <= 999
+
+
 class TestOwnershipFilteringOmittedBoard:
     """get_many()/get_image_names() enforce per-user isolation when board_id is omitted.
 
diff --git a/tests/app/services/images/test_images_default.py b/tests/app/services/images/test_images_default.py
index c10a43edc5f..67f99be04b4 100644
--- a/tests/app/services/images/test_images_default.py
+++ b/tests/app/services/images/test_images_default.py
@@ -1,7 +1,8 @@
 """Tests for ImageService (images_default.py).
 
-Covers subfolder forwarding for all strategies and the delete_images_on_board
-silent-failure contract (Points 2 & 3 from PR review).
+Covers subfolder forwarding for all strategies, the delete_images_on_board
+silent-failure contract (Points 2 & 3 from PR review), and the transactional
+staged-deletion contracts of delete() and delete_intermediates().
 """
 
 from pathlib import Path
@@ -12,12 +13,15 @@
 
 from invokeai.app.services.config.config_default import InvokeAIAppConfig
 from invokeai.app.services.image_files.image_files_common import (
+    ImageFileDeleteException,
     ImageFileSaveException,
 )
 from invokeai.app.services.image_files.image_files_disk import DiskImageFileStorage
 from invokeai.app.services.image_records.image_records_common import (
     ImageCategory,
     ImageRecord,
+    ImageRecordChanges,
+    ImageRecordDeleteException,
     ImageRecordNotFoundException,
     ResourceOrigin,
 )
@@ -26,6 +30,7 @@
 from invokeai.app.services.shared.sqlite.sqlite_util import init_db
 from invokeai.app.util.misc import get_iso_timestamp
 from invokeai.backend.util.logging import InvokeAILogger
+from tests.fixtures.sqlite_database import create_mock_sqlite_database
 
 
 @pytest.fixture
@@ -39,6 +44,8 @@ def image_service() -> ImageService:
     invoker.services.board_image_records.get_board_for_image.return_value = None
     invoker.services.urls.get_image_url.return_value = "http://localhost/img.png"
     invoker.services.configuration.image_subfolder_strategy = "flat"
+    # By default every named intermediate is still an intermediate when the delete runs.
+    invoker.services.image_records.delete_intermediates_by_names.side_effect = lambda names: list(names)
 
     svc.start(invoker)
     return svc
@@ -265,12 +272,15 @@ def test_delete_forwards_subfolder(self, image_service: ImageService):
 
         image_service.delete("test.png")
 
-        invoker.services.image_files.delete.assert_called_once_with("test.png", image_subfolder="2026/04/05")
+        invoker.services.image_files.begin_delete.assert_called_once_with([("test.png", "2026/04/05")])
         invoker.services.image_records.delete.assert_called_once_with("test.png")
+        invoker.services.image_files.commit_delete.assert_called_once_with(
+            invoker.services.image_files.begin_delete.return_value
+        )
 
     def test_delete_intermediates_forwards_subfolder(self, image_service: ImageService):
         invoker = image_service._ImageService__invoker  # type: ignore
-        invoker.services.image_records.delete_intermediates.return_value = [
+        invoker.services.image_records.get_intermediates.return_value = [
             ("img1.png", "intermediate"),
             ("img2.png", "intermediate"),
         ]
@@ -278,11 +288,13 @@ def test_delete_intermediates_forwards_subfolder(self, image_service: ImageServi
         count = image_service.delete_intermediates()
 
         assert count == 2
-        calls = invoker.services.image_files.delete.call_args_list
-        assert calls[0].args == ("img1.png",)
-        assert calls[0].kwargs == {"image_subfolder": "intermediate"}
-        assert calls[1].args == ("img2.png",)
-        assert calls[1].kwargs == {"image_subfolder": "intermediate"}
+        invoker.services.image_files.begin_delete.assert_called_once_with(
+            [("img1.png", "intermediate"), ("img2.png", "intermediate")]
+        )
+        invoker.services.image_records.delete_intermediates_by_names.assert_called_once_with(["img1.png", "img2.png"])
+        invoker.services.image_files.commit_delete.assert_called_once_with(
+            invoker.services.image_files.begin_delete.return_value, image_names=["img1.png", "img2.png"]
+        )
 
 
 # ── Point 3: delete_images_on_board silent-failure contract ──
@@ -359,3 +371,641 @@ def test_database_failure_restores_staged_files(self, image_service: ImageServic
 
         invoker.services.image_files.rollback_delete.assert_called_once_with(token)
         invoker.services.image_files.commit_delete.assert_not_called()
+
+
+# ── Transactional staged deletion (single image and intermediates) ──
+
+
+@pytest.fixture
+def disk_image_service(tmp_path: Path) -> ImageService:
+    """ImageService wired to a real DiskImageFileStorage; all other services are mocks."""
+    svc = ImageService()
+    invoker = MagicMock()
+    invoker.services.configuration.pil_compress_level = 1
+    # By default every named intermediate is still an intermediate when the delete runs.
+    invoker.services.image_records.delete_intermediates_by_names.side_effect = lambda names: list(names)
+    storage = DiskImageFileStorage(tmp_path / "outputs")
+    invoker.services.image_files = storage
+    storage.start(invoker)
+    svc.start(invoker)
+    return svc
+
+
+def _save_image_file(storage: DiskImageFileStorage, image_name: str, image_subfolder: str = "") -> None:
+    storage.save(image=Image.new("RGB", (64, 64)), image_name=image_name, image_subfolder=image_subfolder)
+
+
+def _staging_dirs(storage: DiskImageFileStorage) -> list[Path]:
+    return list(storage.image_root.glob(".delete_*"))
+
+
+@pytest.fixture
+def wired(tmp_path: Path) -> tuple[ImageService, SqliteImageRecordStorage, DiskImageFileStorage]:
+    """ImageService wired to a real record store and a real disk store — no stub decides anything."""
+    config = InvokeAIAppConfig(use_memory_db=True)
+    logger = InvokeAILogger.get_logger(config=config)
+    records = SqliteImageRecordStorage(db=create_mock_sqlite_database(config, logger))
+    storage = DiskImageFileStorage(tmp_path / "outputs")
+
+    svc = ImageService()
+    invoker = MagicMock()
+    invoker.services.configuration.pil_compress_level = 1
+    invoker.services.image_records = records
+    invoker.services.image_files = storage
+    storage.start(invoker)
+    svc.start(invoker)
+    return svc, records, storage
+
+
+def _seed_record(records: SqliteImageRecordStorage, name: str, is_intermediate: bool = True) -> None:
+    records.save(
+        image_name=name,
+        image_origin=ResourceOrigin.INTERNAL,
+        image_category=ImageCategory.GENERAL,
+        width=64,
+        height=64,
+        has_workflow=False,
+        is_intermediate=is_intermediate,
+    )
+
+
+def _restart_file_storage(storage: DiskImageFileStorage, records: SqliteImageRecordStorage) -> DiskImageFileStorage:
+    """Simulates a restart over the same output folder, running delete-journal recovery."""
+    invoker = MagicMock()
+    invoker.services.image_records = records
+    restarted = DiskImageFileStorage(storage.image_root)
+    restarted.start(invoker)
+    return restarted
+
+
+def _unlink_always_fails(path: Path, missing_ok: bool = False) -> None:
+    raise OSError("device busy")
+
+
+class TestDeleteTransactional:
+    """delete() journals its intent, deletes the record, then purges — never losing files on failure."""
+
+    def test_delete_success_removes_files_record_and_fires_callback_once(self, disk_image_service: ImageService):
+        invoker = disk_image_service._ImageService__invoker  # type: ignore
+        storage = invoker.services.image_files
+        _save_image_file(storage, "img.png")
+        invoker.services.image_records.get.return_value = _make_record(image_name="img.png")
+        deleted_callbacks: list[str] = []
+        disk_image_service.on_deleted(deleted_callbacks.append)
+
+        disk_image_service.delete("img.png")
+
+        assert not storage.get_path("img.png").exists()
+        assert not storage.get_path("img.png", thumbnail=True).exists()
+        invoker.services.image_records.delete.assert_called_once_with("img.png")
+        assert deleted_callbacks == ["img.png"]
+        assert _staging_dirs(storage) == []
+
+    def test_delete_journal_failure_keeps_record_and_raises(self, image_service: ImageService):
+        """The journal is written before the record is deleted, so failing to write it aborts."""
+        invoker = image_service._ImageService__invoker  # type: ignore
+        invoker.services.image_files.begin_delete.side_effect = ImageFileDeleteException("disk error")
+        deleted_callbacks: list[str] = []
+        image_service.on_deleted(deleted_callbacks.append)
+
+        with pytest.raises(ImageFileDeleteException):
+            image_service.delete("test.png")
+
+        invoker.services.image_records.delete.assert_not_called()
+        invoker.services.image_files.commit_delete.assert_not_called()
+        assert deleted_callbacks == []
+
+    def test_delete_db_failure_leaves_files_and_raises(self, disk_image_service: ImageService):
+        invoker = disk_image_service._ImageService__invoker  # type: ignore
+        storage = invoker.services.image_files
+        _save_image_file(storage, "img.png")
+        invoker.services.image_records.get.return_value = _make_record(image_name="img.png")
+        invoker.services.image_records.delete.side_effect = ImageRecordDeleteException()
+        deleted_callbacks: list[str] = []
+        disk_image_service.on_deleted(deleted_callbacks.append)
+
+        with pytest.raises(ImageRecordDeleteException):
+            disk_image_service.delete("img.png")
+
+        # Nothing was moved, so the image and its thumbnail are still exactly where they were.
+        assert storage.get_path("img.png").exists()
+        assert storage.get_path("img.png", thumbnail=True).exists()
+        assert deleted_callbacks == []
+        assert _staging_dirs(storage) == []
+
+    def test_delete_journal_cleanup_failure_still_raises_db_error(self, image_service: ImageService):
+        invoker = image_service._ImageService__invoker  # type: ignore
+        invoker.services.image_records.delete.side_effect = ImageRecordDeleteException()
+        invoker.services.image_files.abandon_delete.side_effect = ImageFileDeleteException("journal locked")
+        deleted_callbacks: list[str] = []
+        image_service.on_deleted(deleted_callbacks.append)
+
+        with pytest.raises(ImageRecordDeleteException):
+            image_service.delete("test.png")
+
+        invoker.services.image_files.abandon_delete.assert_called_once_with(
+            invoker.services.image_files.begin_delete.return_value
+        )
+        invoker.services.image_files.commit_delete.assert_not_called()
+        assert deleted_callbacks == []
+
+    def test_delete_commit_failure_is_logged_not_raised(self, image_service: ImageService):
+        invoker = image_service._ImageService__invoker  # type: ignore
+        invoker.services.image_files.commit_delete.side_effect = ImageFileDeleteException("purge failed")
+        deleted_callbacks: list[str] = []
+        image_service.on_deleted(deleted_callbacks.append)
+
+        image_service.delete("test.png")
+
+        invoker.services.image_records.delete.assert_called_once_with("test.png")
+        invoker.services.image_files.abandon_delete.assert_not_called()
+        assert deleted_callbacks == ["test.png"]
+        invoker.services.logger.error.assert_called()
+
+
+class TestDeleteIntermediatesTransactional:
+    """delete_intermediates() deletes records first, then purges the files of exactly the rows it
+    removed. It never stages or restores a promoted image's files, so there is no restore step for a
+    concurrent delete to race (PR #9361, JPPhoto round 2)."""
+
+    def test_success_deletes_multiple_intermediates(self, disk_image_service: ImageService):
+        invoker = disk_image_service._ImageService__invoker  # type: ignore
+        storage = invoker.services.image_files
+        names = ["tmp1.png", "tmp2.png", "tmp3.png"]
+        for name in names:
+            _save_image_file(storage, name)
+        invoker.services.image_records.get_intermediates.return_value = [(name, "") for name in names]
+        deleted_callbacks: list[str] = []
+        disk_image_service.on_deleted(deleted_callbacks.append)
+
+        count = disk_image_service.delete_intermediates()
+
+        assert count == 3
+        for name in names:
+            assert not storage.get_path(name).exists()
+            assert not storage.get_path(name, thumbnail=True).exists()
+        invoker.services.image_records.delete_intermediates_by_names.assert_called_once_with(names)
+        assert deleted_callbacks == names
+        assert _staging_dirs(storage) == []
+
+    def test_promoted_image_keeps_its_files(self, disk_image_service: ImageService):
+        """An image the DB refused to delete (no longer an intermediate) keeps its files untouched."""
+        invoker = disk_image_service._ImageService__invoker  # type: ignore
+        storage = invoker.services.image_files
+        for name in ("tmp1.png", "promoted.png", "tmp2.png"):
+            _save_image_file(storage, name)
+        invoker.services.image_records.get_intermediates.return_value = [
+            ("tmp1.png", ""),
+            ("promoted.png", ""),
+            ("tmp2.png", ""),
+        ]
+        # The store reports it removed everything except promoted.png, so that file is never purged.
+        invoker.services.image_records.delete_intermediates_by_names.side_effect = lambda names: [
+            name for name in names if name != "promoted.png"
+        ]
+        deleted_callbacks: list[str] = []
+        disk_image_service.on_deleted(deleted_callbacks.append)
+
+        count = disk_image_service.delete_intermediates()
+
+        assert count == 2
+        assert storage.get_path("promoted.png").exists()
+        assert storage.get_path("promoted.png", thumbnail=True).exists()
+        for name in ("tmp1.png", "tmp2.png"):
+            assert not storage.get_path(name).exists()
+            assert not storage.get_path(name, thumbnail=True).exists()
+        assert deleted_callbacks == ["tmp1.png", "tmp2.png"]
+        assert _staging_dirs(storage) == []
+
+    def test_only_deleted_rows_are_purged_and_announced(self, image_service: ImageService):
+        invoker = image_service._ImageService__invoker  # type: ignore
+        invoker.services.image_records.get_intermediates.return_value = [("tmp1.png", ""), ("promoted.png", "")]
+        invoker.services.image_records.delete_intermediates_by_names.side_effect = lambda names: ["tmp1.png"]
+        deleted_callbacks: list[str] = []
+        image_service.on_deleted(deleted_callbacks.append)
+
+        count = image_service.delete_intermediates()
+
+        assert count == 1
+        # The promoted row's file is never touched: only the deleted row is purged.
+        invoker.services.image_files.commit_delete.assert_called_once_with(
+            invoker.services.image_files.begin_delete.return_value, image_names=["tmp1.png"]
+        )
+        assert deleted_callbacks == ["tmp1.png"]
+
+    def test_subfolder_is_forwarded_to_the_file_purge(self, image_service: ImageService):
+        invoker = image_service._ImageService__invoker  # type: ignore
+        invoker.services.image_records.get_intermediates.return_value = [("tmp1.png", "a/b")]
+        invoker.services.image_records.delete_intermediates_by_names.side_effect = lambda names: list(names)
+        image_service.delete_intermediates()
+
+        invoker.services.image_files.begin_delete.assert_called_once_with([("tmp1.png", "a/b")])
+
+    def test_file_purge_failure_is_logged_and_does_not_raise(self, image_service: ImageService):
+        """A filesystem failure must not undo the committed record deletions or raise: the records
+        are already gone, and the journal the purge leaves behind is retried at the next startup."""
+        invoker = image_service._ImageService__invoker  # type: ignore
+        invoker.services.image_records.get_intermediates.return_value = [("tmp1.png", ""), ("tmp2.png", "")]
+        invoker.services.image_records.delete_intermediates_by_names.side_effect = lambda names: list(names)
+        invoker.services.image_files.commit_delete.side_effect = ImageFileDeleteException("purge failed")
+        deleted_callbacks: list[str] = []
+        image_service.on_deleted(deleted_callbacks.append)
+
+        count = image_service.delete_intermediates()
+
+        assert count == 2
+        # Both records were deleted, so both deletions are announced despite the file failure.
+        assert deleted_callbacks == ["tmp1.png", "tmp2.png"]
+        invoker.services.image_files.abandon_delete.assert_not_called()
+        invoker.services.logger.error.assert_called()
+
+    def test_db_failure_raises_and_purges_nothing(self, image_service: ImageService):
+        invoker = image_service._ImageService__invoker  # type: ignore
+        invoker.services.image_records.get_intermediates.return_value = [("tmp1.png", ""), ("tmp2.png", "")]
+        invoker.services.image_records.delete_intermediates_by_names.side_effect = ImageRecordDeleteException()
+        deleted_callbacks: list[str] = []
+        image_service.on_deleted(deleted_callbacks.append)
+
+        with pytest.raises(ImageRecordDeleteException):
+            image_service.delete_intermediates()
+
+        # No record was removed, so no file may be purged and the journal must be discarded.
+        invoker.services.image_files.commit_delete.assert_not_called()
+        invoker.services.image_files.abandon_delete.assert_called_once_with(
+            invoker.services.image_files.begin_delete.return_value
+        )
+        assert deleted_callbacks == []
+
+    def test_nothing_deleted_returns_zero_and_fires_no_callbacks(self, image_service: ImageService):
+        invoker = image_service._ImageService__invoker  # type: ignore
+        invoker.services.image_records.get_intermediates.return_value = [("promoted.png", "")]
+        invoker.services.image_records.delete_intermediates_by_names.side_effect = lambda names: []
+        deleted_callbacks: list[str] = []
+        image_service.on_deleted(deleted_callbacks.append)
+
+        assert image_service.delete_intermediates() == 0
+
+        # The journal still lists the promoted image; the purge selects nothing.
+        invoker.services.image_files.commit_delete.assert_called_once_with(
+            invoker.services.image_files.begin_delete.return_value, image_names=[]
+        )
+        assert deleted_callbacks == []
+
+    def test_empty_intermediates_is_a_noop(self, image_service: ImageService):
+        invoker = image_service._ImageService__invoker  # type: ignore
+        invoker.services.image_records.get_intermediates.return_value = []
+        deleted_callbacks: list[str] = []
+        image_service.on_deleted(deleted_callbacks.append)
+
+        assert image_service.delete_intermediates() == 0
+
+        # Nothing to delete: no journal is written and the record store is never asked to delete.
+        invoker.services.image_records.delete_intermediates_by_names.assert_not_called()
+        invoker.services.image_files.begin_delete.assert_not_called()
+        assert deleted_callbacks == []
+
+
+class TestDeleteIntermediatesAgainstRealRecords:
+    """delete_intermediates() wired to a real record store, so no stub stands in for the DB decision.
+
+    The mocked tests above can only assert that the service honours whatever the store reports. These
+    exercise the real store, which is where the promoted-vs-already-gone distinction is actually made,
+    and where the concurrency hazards JPPhoto reported would surface.
+    """
+
+    def _seed(self, records: SqliteImageRecordStorage, storage: DiskImageFileStorage, name: str) -> None:
+        _seed_record(records, name)
+        _save_image_file(storage, name)
+
+    def _promote_after_snapshot(
+        self,
+        records: SqliteImageRecordStorage,
+        monkeypatch,
+        image_name: str,
+    ) -> None:
+        """Promote an image out of intermediate status after the snapshot but before the DB delete.
+
+        Promoting it earlier would drop it from the snapshot entirely; the interesting case is an
+        image that is in the snapshot yet is no longer an intermediate by the time the conditional
+        DELETE runs, so its record (and files) must survive.
+        """
+        real_get_intermediates = records.get_intermediates
+
+        def snapshot_then_promote():
+            pairs = real_get_intermediates()
+            records.update(image_name, ImageRecordChanges(is_intermediate=False))
+            return pairs
+
+        monkeypatch.setattr(records, "get_intermediates", snapshot_then_promote)
+
+    def test_all_intermediates_are_deleted(self, wired) -> None:
+        svc, records, storage = wired
+        self._seed(records, storage, "tmp1.png")
+        self._seed(records, storage, "tmp2.png")
+        deleted_callbacks: list[str] = []
+        svc.on_deleted(deleted_callbacks.append)
+
+        assert svc.delete_intermediates() == 2
+
+        for name in ("tmp1.png", "tmp2.png"):
+            assert not storage.get_path(name).exists()
+            with pytest.raises(ImageRecordNotFoundException):
+                records.get(name)
+        assert sorted(deleted_callbacks) == ["tmp1.png", "tmp2.png"]
+        assert _staging_dirs(storage) == []
+
+    def test_promoted_image_keeps_record_and_files(self, wired, monkeypatch) -> None:
+        svc, records, storage = wired
+        self._seed(records, storage, "tmp1.png")
+        self._seed(records, storage, "promoted.png")
+        self._promote_after_snapshot(records, monkeypatch, "promoted.png")
+        deleted_callbacks: list[str] = []
+        svc.on_deleted(deleted_callbacks.append)
+
+        assert svc.delete_intermediates() == 1
+
+        assert storage.get_path("promoted.png").exists()
+        assert records.get("promoted.png").is_intermediate is False
+        assert not storage.get_path("tmp1.png").exists()
+        assert deleted_callbacks == ["tmp1.png"]
+        assert _staging_dirs(storage) == []
+
+    def test_record_removed_by_another_path_between_snapshot_and_delete(self, wired, monkeypatch) -> None:
+        """An image fully deleted elsewhere after the snapshot is not counted and its (now absent)
+        files are left to the path that owns that deletion — we never touch them."""
+        svc, records, storage = wired
+        self._seed(records, storage, "tmp1.png")
+        self._seed(records, storage, "gone.png")
+
+        real_get_intermediates = records.get_intermediates
+
+        def snapshot_then_delete_gone():
+            pairs = real_get_intermediates()
+            # A single-image delete elsewhere removes gone.png (record and files) after our snapshot.
+            records.delete("gone.png")
+            storage.delete("gone.png")
+            return pairs
+
+        monkeypatch.setattr(records, "get_intermediates", snapshot_then_delete_gone)
+        deleted_callbacks: list[str] = []
+        svc.on_deleted(deleted_callbacks.append)
+
+        count = svc.delete_intermediates()
+
+        assert count == 1
+        assert deleted_callbacks == ["tmp1.png"]
+        assert not storage.get_path("tmp1.png").exists()
+        # gone.png was purged by the other path; we neither resurrect nor re-report it.
+        assert not storage.get_path("gone.png").exists()
+        assert _staging_dirs(storage) == []
+
+    def test_promoted_record_deleted_after_conditional_delete_is_not_resurrected(self, wired, monkeypatch) -> None:
+        """The B3 regression: a promoted image is concurrently deleted (record and files) right after
+        the conditional DELETE keeps it. Because we never staged its files, there is nothing to
+        restore — its files stay deleted and are not stranded on disk with no record.
+        """
+        svc, records, storage = wired
+        self._seed(records, storage, "tmp1.png")
+        self._seed(records, storage, "promoted.png")
+        self._promote_after_snapshot(records, monkeypatch, "promoted.png")
+
+        real_delete_by_names = records.delete_intermediates_by_names
+
+        def delete_then_lose_the_promoted_record(names: list[str]):
+            deleted = real_delete_by_names(names)
+            # A concurrent single-image delete removes the promoted image entirely, right after the
+            # conditional DELETE chose to keep it.
+            records.delete("promoted.png")
+            storage.delete("promoted.png")
+            return deleted
+
+        monkeypatch.setattr(records, "delete_intermediates_by_names", delete_then_lose_the_promoted_record)
+        deleted_callbacks: list[str] = []
+        svc.on_deleted(deleted_callbacks.append)
+
+        count = svc.delete_intermediates()
+
+        assert count == 1
+        assert deleted_callbacks == ["tmp1.png"]
+        assert not storage.get_path("tmp1.png").exists()
+        # promoted.png's files stay deleted — never resurrected into an orphan.
+        assert not storage.get_path("promoted.png").exists()
+        assert not storage.get_path("promoted.png", thumbnail=True).exists()
+        with pytest.raises(ImageRecordNotFoundException):
+            records.get("promoted.png")
+        assert _staging_dirs(storage) == []
+
+
+class TestDeleteAgainstRealRecords:
+    """Single-image delete wired to a real record store, covering the concurrent-delete interleaving
+    JPPhoto reported (PR #9361 round 3)."""
+
+    def test_a_failed_delete_never_resurrects_files_another_request_removed(self, wired, monkeypatch) -> None:
+        """Two requests delete the same image; one commits the record deletion and the other fails.
+
+        The failing request must not put the files back: nothing references them any more, and the
+        journal that would let startup recovery find them is gone with the request that won.
+        """
+        svc, records, storage = wired
+        _seed_record(records, "img.png")
+        _save_image_file(storage, "img.png")
+
+        real_delete = records.delete
+
+        def competing_delete_then_fail(image_name: str) -> None:
+            # The competing request wins the race: it removes the record and purges the files while
+            # this delete is still in flight, and only then does this one's own delete fail.
+            real_delete(image_name)
+            storage.delete(image_name)
+            raise ImageRecordDeleteException()
+
+        monkeypatch.setattr(records, "delete", competing_delete_then_fail)
+        deleted_callbacks: list[str] = []
+        svc.on_deleted(deleted_callbacks.append)
+
+        with pytest.raises(ImageRecordDeleteException):
+            svc.delete("img.png")
+
+        assert not storage.get_path("img.png").exists()
+        assert not storage.get_path("img.png", thumbnail=True).exists()
+        assert deleted_callbacks == []
+        assert _staging_dirs(storage) == []
+
+    def test_a_database_failure_leaves_the_image_completely_intact(self, wired, monkeypatch) -> None:
+        svc, records, storage = wired
+        _seed_record(records, "img.png")
+        _save_image_file(storage, "img.png")
+
+        def failing_delete(image_name: str) -> None:
+            raise ImageRecordDeleteException()
+
+        monkeypatch.setattr(records, "delete", failing_delete)
+
+        with pytest.raises(ImageRecordDeleteException):
+            svc.delete("img.png")
+
+        assert storage.get_path("img.png").exists()
+        assert storage.get_path("img.png", thumbnail=True).exists()
+        assert records.get("img.png").image_name == "img.png"
+        # The journal is discarded: the image is live, so nothing must be left pointing at it.
+        assert _staging_dirs(storage) == []
+
+
+class TestDeleteJournalSurvivesFailedPurges:
+    """Records-first deletion commits the record removal before the files are purged. A crash or a
+    filesystem failure in that window must leave a journal, not a silent orphan (JPPhoto, PR #9361
+    round 3)."""
+
+    def test_intermediates_purge_failure_leaves_a_journal_the_next_startup_finishes(self, wired, monkeypatch) -> None:
+        svc, records, storage = wired
+        for name in ("tmp1.png", "tmp2.png"):
+            _seed_record(records, name)
+            _save_image_file(storage, name)
+        deleted_callbacks: list[str] = []
+        svc.on_deleted(deleted_callbacks.append)
+
+        with patch.object(Path, "unlink", _unlink_always_fails):
+            count = svc.delete_intermediates()
+
+        # The records are committed as gone, so the deletion succeeded and is announced...
+        assert count == 2
+        assert sorted(deleted_callbacks) == ["tmp1.png", "tmp2.png"]
+        for name in ("tmp1.png", "tmp2.png"):
+            with pytest.raises(ImageRecordNotFoundException):
+                records.get(name)
+            # ...but the files could not be removed, so they must still be journalled.
+            assert storage.get_path(name).exists()
+        assert _staging_dirs(storage) != []
+
+        _restart_file_storage(storage, records)
+
+        for name in ("tmp1.png", "tmp2.png"):
+            assert not storage.get_path(name).exists()
+            assert not storage.get_path(name, thumbnail=True).exists()
+        assert _staging_dirs(storage) == []
+
+    def test_a_crash_before_the_purge_leaves_a_journal_the_next_startup_finishes(self, wired) -> None:
+        """The process dies between the committed record deletion and the file purge."""
+        svc, records, storage = wired
+        _seed_record(records, "img.png", is_intermediate=False)
+        _save_image_file(storage, "img.png")
+
+        # Everything delete() does up to the point of no return, and then nothing.
+        record = records.get("img.png")
+        storage.begin_delete([("img.png", record.image_subfolder)])
+        records.delete("img.png")
+
+        assert storage.get_path("img.png").exists()
+
+        _restart_file_storage(storage, records)
+
+        assert not storage.get_path("img.png").exists()
+        assert not storage.get_path("img.png", thumbnail=True).exists()
+        assert _staging_dirs(storage) == []
+
+    def test_a_crash_before_the_record_delete_keeps_the_image(self, wired) -> None:
+        """The mirror case: the journal is written but the record deletion never happened."""
+        svc, records, storage = wired
+        _seed_record(records, "img.png", is_intermediate=False)
+        _save_image_file(storage, "img.png")
+
+        storage.begin_delete([("img.png", "")])
+
+        _restart_file_storage(storage, records)
+
+        assert storage.get_path("img.png").exists()
+        assert storage.get_path("img.png", thumbnail=True).exists()
+        assert records.get("img.png").image_name == "img.png"
+        assert _staging_dirs(storage) == []
+
+
+class TestFailedSaveCleanup:
+    """A save that fails halfway must clean up record-first, like every other delete path.
+
+    The order is load-bearing: a concurrent deleter rolling back decides whether to restore an
+    image's files by asking whether its record is still there, so purging files while the record
+    survives would tell it to put them back and strand them (adversarial review, PR #9361).
+    """
+
+    def test_the_record_is_deleted_before_the_files_are_purged(self, image_service: ImageService) -> None:
+        invoker = image_service._ImageService__invoker  # type: ignore
+        invoker.services.image_files.save.side_effect = ImageFileSaveException()
+        order: list[str] = []
+        token = object()
+
+        def journal(images):
+            order.append("journal")
+            return token
+
+        invoker.services.image_files.begin_delete.side_effect = journal
+        invoker.services.image_records.delete.side_effect = lambda name: order.append("record")
+        invoker.services.image_files.commit_delete.side_effect = lambda t, image_names=None: order.append("purge")
+
+        with pytest.raises(ImageFileSaveException):
+            image_service.create(
+                image=Image.new("RGB", (8, 8)),
+                image_origin=ResourceOrigin.INTERNAL,
+                image_category=ImageCategory.GENERAL,
+            )
+
+        assert order == ["journal", "record", "purge"]
+        invoker.services.image_files.commit_delete.assert_called_once_with(token)
+
+    def test_a_surviving_record_keeps_its_files(self, image_service: ImageService) -> None:
+        """If the record cannot be deleted the image is still referenced, so nothing may be purged."""
+        invoker = image_service._ImageService__invoker  # type: ignore
+        invoker.services.image_files.save.side_effect = ImageFileSaveException()
+        invoker.services.image_records.delete.side_effect = ImageRecordDeleteException()
+
+        with pytest.raises(ImageFileSaveException):
+            image_service.create(
+                image=Image.new("RGB", (8, 8)),
+                image_origin=ResourceOrigin.INTERNAL,
+                image_category=ImageCategory.GENERAL,
+            )
+
+        invoker.services.image_files.commit_delete.assert_not_called()
+        invoker.services.image_files.delete.assert_not_called()
+        invoker.services.image_files.abandon_delete.assert_called_once_with(
+            invoker.services.image_files.begin_delete.return_value
+        )
+
+
+class TestConcurrentBoardDeleteAgainstRealRecords:
+    """delete_images_on_board() still stages, because its per-item contract needs a pre-flight move.
+    Two of them racing for one image must not strand it (adversarial review, PR #9361)."""
+
+    def test_committing_an_empty_token_still_purges_the_files(self, wired, monkeypatch) -> None:
+        """The loser moved the files aside; the winner staged nothing and removed the record.
+
+        The winner's commit is the only thing standing between the loser's restore and a permanent
+        orphan: by the time the loser rolls back, the record is still there, so its own re-check
+        tells it to keep the files it just put back.
+        """
+        svc, records, storage = wired
+        invoker = svc._ImageService__invoker  # type: ignore
+        _seed_record(records, "img.png", is_intermediate=False)
+        _save_image_file(storage, "img.png")
+        invoker.services.board_image_records.get_all_board_image_names_for_board.return_value = ["img.png"]
+
+        # The competing request wins the race to the files, so this delete stages an empty token.
+        competing = storage.stage_delete("img.png", "")
+        real_delete_many = records.delete_many
+
+        def competitor_rolls_back_then_delete(image_names: list[str]) -> None:
+            # The competing request's own record deletion failed, so it restores the files — while
+            # this record is still present, which is what makes its re-check keep them.
+            storage.rollback_delete(competing)
+            real_delete_many(image_names)
+
+        monkeypatch.setattr(records, "delete_many", competitor_rolls_back_then_delete)
+
+        deleted, failed = svc.delete_images_on_board("board-1")
+
+        assert deleted == ["img.png"]
+        assert failed == []
+        with pytest.raises(ImageRecordNotFoundException):
+            records.get("img.png")
+        assert not storage.get_path("img.png").exists()
+        assert not storage.get_path("img.png", thumbnail=True).exists()
+        assert _staging_dirs(storage) == []