diff --git a/docs/src/content/docs/features/gallery.mdx b/docs/src/content/docs/features/gallery.mdx
index 2fc37776a19..f4ad2526bbc 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 journal is written to disk before the record is removed and discarded only once the file removal has itself been flushed to disk, so a power loss at any point leaves either the whole image or a journal that finds its remains. If the disk cannot confirm the journal was written, the deletion is refused and the image stays; the same rule governs the cleanup after a failed image upload or save, which keeps the half-created entry (visible after the gallery refreshes) for you to delete later rather than remove its record without a journal. (Windows cannot flush directory entries, so there the journal's durability follows the drive's own write caching.) 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..dc5a17da406 100644
--- a/invokeai/app/services/image_files/image_files_disk.py
+++ b/invokeai/app/services/image_files/image_files_disk.py
@@ -1,4 +1,5 @@
 # Copyright (c) 2022 Kyle Schouviller (https://github.com/kyle0654) and the InvokeAI Team
+import errno
 import io
 import json
 import os
@@ -6,6 +7,7 @@
 import tempfile
 import threading
 import zlib
+from collections.abc import Collection, Iterable, Sequence
 from dataclasses import dataclass
 from pathlib import Path
 from queue import Queue
@@ -32,8 +34,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 +108,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 +221,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 +229,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 +239,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 +250,87 @@ 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)
+            self.__persist_purges([(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)
+        purged: list[tuple[str, str]] = []
+        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)
+                purged.append((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)}")
+        try:
+            self.__persist_purges(purged)
+        except OSError as e:
+            # Same rule: the journal is the only thing that can redo a purge the disk lost.
+            raise ImageFileDeleteException(f"Failed to make the purge of deleted image files durable: {e}") from e
+        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 +339,93 @@ 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)
+            self.__persist_purges([(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. A sync that fails propagates for the same reason:
+        a journal whose durability is unknown must not license a record deletion.
+        """
+        self.__fsync_directory(journal_dir)
+        self.__fsync_directory(self.__output_folder)
+
+    def __persist_purges(self, images: Iterable[tuple[str, str]]) -> None:
+        """Makes the removal (or restoration) of these images' files survive a power loss.
+
+        Runs before the journal that names them is dropped. Unlinks and renames live in the parent
+        directories' entries, and nothing else syncs those; a filesystem that persists the
+        journal's removal ahead of them would bring the files back with no journal left to find
+        them by. Raises ``OSError`` when a sync fails, so the caller keeps the journal.
+        """
+        parents: dict[Path, None] = {}
+        for image_name, image_subfolder in images:
+            for path in self.__delete_candidates(image_name, image_subfolder):
+                parents[path.parent] = None
+        for parent in parents:
+            # A parent that no longer exists has no entries to make durable.
+            self.__fsync_directory(parent, missing_ok=True)
+
+    @staticmethod
+    def __fsync_directory(directory: Path, missing_ok: bool = False) -> None:
+        if os.name == "nt":
+            # Windows cannot open a directory for fsync; the manifest write is all we get.
+            return
+        try:
+            dir_fd = os.open(directory, os.O_RDONLY)
+        except (FileNotFoundError, NotADirectoryError):
+            if missing_ok:
+                return
+            raise
+        try:
+            os.fsync(dir_fd)
+        except OSError as e:
+            # Some filesystems cannot sync a directory at all and say so with one of these; there is
+            # nothing more to be had from them. Anything else (EIO, ENOSPC, ...) means the entries
+            # may not be on disk, and the caller must not proceed as if they were.
+            if e.errno not in (errno.EINVAL, errno.ENOTSUP, errno.EOPNOTSUPP, errno.ENOSYS, errno.EBADF):
+                raise
+        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 +492,72 @@ 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
+                images = self.__manifest_images(data)
+                for image_name, image_subfolder in images:
+                    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)
+                # Every restore and purge above must be on disk before the journal that could redo
+                # them is gone. A failed sync propagates and keeps the journal for the next startup.
+                self.__persist_purges(images)
+                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_moves/image_moves_default.py b/invokeai/app/services/image_moves/image_moves_default.py
index 03077a84a6b..b9e3db0e069 100644
--- a/invokeai/app/services/image_moves/image_moves_default.py
+++ b/invokeai/app/services/image_moves/image_moves_default.py
@@ -1,7 +1,9 @@
 import os
 import tempfile
 import threading
+from collections.abc import Iterator
 from concurrent.futures import Future, ThreadPoolExecutor
+from contextlib import contextmanager
 from dataclasses import dataclass
 from datetime import datetime
 from pathlib import Path
@@ -85,6 +87,9 @@ def __init__(
         self._future: Future | None = None
         self._future_operation: ImageMoveBackgroundOperation | None = None
         self._last_background_error: str | None = None
+        # Serializes the move service's relocate-and-repoint units against the image delete
+        # units in ImageService. See image_mutation_lock() for the interleaving it prevents.
+        self._image_mutation_lock = threading.RLock()
         self._invoker = None
         self._session_queue = None
 
@@ -105,6 +110,27 @@ def set_session_queue(self, session_queue) -> None:
     def stop(self, *args, **kwargs) -> None:
         self._executor.shutdown(wait=True, cancel_futures=False)
 
+    @contextmanager
+    def image_mutation_lock(self) -> Iterator[None]:
+        """Serializes image delete units against subfolder relocation units.
+
+        An image delete reads an image's subfolder, deletes its record, then purges its files
+        at that subfolder. A move unit does the opposite: it relocates the files and repoints
+        the record. If the two interleave, the delete purges the path its snapshot named while
+        the files sit at the new one — permanent orphans, unrecoverable because the record is
+        gone and a clean purge drops the journal (JPPhoto, PR #9361). ``ImageService`` holds
+        this lock across each of its delete units, and the move service holds it across each
+        plan-relocate-repoint cycle below, so neither can observe the other half-done.
+
+        It is a reentrant lock because both sides run their units to completion in one thread;
+        nothing inside a unit may block on another thread that needs this lock. It is also
+        process-local: two Invoke processes sharing one output folder and database are not
+        serialized by it — the same limitation the route guard has, which the delete journal's
+        startup recovery re-check papers over for deletes.
+        """
+        with self._image_mutation_lock:
+            yield
+
     def start_background_move_all(self) -> ImageMoveBackgroundStatus:
         return self._start_background_operation("move_all", self.move_all_images, require_idle_queue=True)
 
@@ -204,27 +230,35 @@ def move_all_images(self) -> ImageMoveResult:
         errors = recovered.errors
 
         while True:
-            moves, plan_errors = self._plan_batch(
-                last_image_name=last_image_name, limit=100, record_missing_errors=True
-            )
-            errors += plan_errors
-            if not moves:
-                next_name = self._next_image_name(last_image_name)
-                if next_name is None:
-                    break
-                last_image_name = next_name
-                continue
-
-            job_id = self.create_move_job(moves)
-            planned += len(moves)
-            try:
-                self.perform_filesystem_moves(job_id)
-                committed += self.commit_database_updates(job_id)
-                errors += self._count_job_errors(job_id)
-            except Exception as e:
-                errors += 1
-                self.record_job_error_message(job_id, str(e))
-                raise
+            # The whole batch cycle — plan, journal the job, relocate files, repoint the
+            # records — holds the image-mutation lock. A delete interleaved inside the cycle
+            # would purge the path its snapshot named while this job's relocate-and-repoint
+            # landed mid-flight, stranding files at the new subfolder with no record and no
+            # journal (JPPhoto, PR #9361). Planning and creating the job are inside the lock
+            # too, so an item whose record a delete removes is never planned in the first
+            # place instead of failing the job after the fact.
+            with self.image_mutation_lock():
+                moves, plan_errors = self._plan_batch(
+                    last_image_name=last_image_name, limit=100, record_missing_errors=True
+                )
+                errors += plan_errors
+                if not moves:
+                    next_name = self._next_image_name(last_image_name)
+                    if next_name is None:
+                        break
+                    last_image_name = next_name
+                    continue
+
+                job_id = self.create_move_job(moves)
+                planned += len(moves)
+                try:
+                    self.perform_filesystem_moves(job_id)
+                    committed += self.commit_database_updates(job_id)
+                    errors += self._count_job_errors(job_id)
+                except Exception as e:
+                    errors += 1
+                    self.record_job_error_message(job_id, str(e))
+                    raise
             last_image_name = moves[-1].image_name
 
         return ImageMoveResult(planned=planned, committed=committed, errors=errors)
@@ -244,9 +278,13 @@ def startup_recovery(self) -> ImageMoveResult:
         errors = 0
         for job_id in job_ids:
             try:
-                self.complete_partial_filesystem_moves(job_id)
-                self.cleanup_empty_source_dirs(job_id)
-                committed += self.commit_database_updates(job_id)
+                # Same unit as a live batch: finishing an interrupted relocation must not
+                # interleave with a concurrent delete, which would otherwise purge the path
+                # its snapshot named while the files land at the new one (JPPhoto, PR #9361).
+                with self.image_mutation_lock():
+                    self.complete_partial_filesystem_moves(job_id)
+                    self.cleanup_empty_source_dirs(job_id)
+                    committed += self.commit_database_updates(job_id)
             except Exception as e:
                 if self._is_unrecoverable_error(e):
                     self.mark_job_unrecoverable(job_id, str(e))
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..975d1c67b3f 100644
--- a/invokeai/app/services/images/images_default.py
+++ b/invokeai/app/services/images/images_default.py
@@ -1,3 +1,5 @@
+from collections.abc import Iterator
+from contextlib import contextmanager
 from typing import Optional
 
 from PIL.Image import Image as PILImageType
@@ -34,6 +36,26 @@ class ImageService(ImageServiceABC):
     def start(self, invoker: Invoker) -> None:
         self.__invoker = invoker
 
+    @contextmanager
+    def _image_mutation_lock(self) -> Iterator[None]:
+        """Holds the image-mutation lock across a delete unit.
+
+        Every delete here reads an image's subfolder and purges its files at that path after
+        the record is gone. A concurrent subfolder move that relocates files and repoints the
+        record mid-unit would leave the purge sweeping a path the files already left: permanent
+        orphans, unrecoverable because the record is gone and a clean purge drops the journal
+        (JPPhoto, PR #9361). The move service takes the same lock around each of its
+        plan-relocate-repoint cycles, so neither unit can observe the other half-done. With no
+        move service configured there is nothing to coordinate with. The lock is process-local:
+        two Invoke processes sharing one output folder and database are not serialized by it.
+        """
+        image_moves = getattr(self.__invoker.services, "image_moves", None)
+        if image_moves is None:
+            yield
+            return
+        with image_moves.image_mutation_lock():
+            yield
+
     def create(
         self,
         image: PILImageType,
@@ -64,39 +86,47 @@ def create(
         (width, height) = image.size
 
         try:
-            # TODO: Consider using a transaction here to ensure consistency between storage and database
-            self.__invoker.services.image_records.save(
-                # Non-nullable fields
-                image_name=image_name,
-                image_origin=image_origin,
-                image_category=image_category,
-                width=width,
-                height=height,
-                has_workflow=workflow is not None or graph is not None,
-                # Meta fields
-                is_intermediate=is_intermediate,
-                # Nullable fields
-                node_id=node_id,
-                metadata=metadata,
-                session_id=session_id,
-                user_id=user_id,
-                image_subfolder=image_subfolder,
-            )
-            if board_id is not None:
-                try:
-                    self.__invoker.services.board_image_records.add_image_to_board(
-                        board_id=board_id, image_name=image_name
-                    )
-                except Exception as e:
-                    self.__invoker.services.logger.warning(f"Failed to add image to board {board_id}: {str(e)}")
-            self.__invoker.services.image_files.save(
-                image_name=image_name,
-                image=image,
-                metadata=metadata,
-                workflow=workflow,
-                graph=graph,
-                image_subfolder=image_subfolder,
-            )
+            # The mutation lock spans the record write through the file write: the moment the
+            # record commits, the row is visible to the move service's planner, and a relocation
+            # that landed before the files did would leave the record naming a subfolder the
+            # files were never written to. Under the date strategy the two can disagree by a day
+            # (the strategy reads the local clock, the move target comes from the record's UTC
+            # created_at), so this is not a rare window. The failure cleanup below takes the
+            # lock again; the reentrant lock makes that safe.
+            with self._image_mutation_lock():
+                # TODO: Consider using a transaction here to ensure consistency between storage and database
+                self.__invoker.services.image_records.save(
+                    # Non-nullable fields
+                    image_name=image_name,
+                    image_origin=image_origin,
+                    image_category=image_category,
+                    width=width,
+                    height=height,
+                    has_workflow=workflow is not None or graph is not None,
+                    # Meta fields
+                    is_intermediate=is_intermediate,
+                    # Nullable fields
+                    node_id=node_id,
+                    metadata=metadata,
+                    session_id=session_id,
+                    user_id=user_id,
+                    image_subfolder=image_subfolder,
+                )
+                if board_id is not None:
+                    try:
+                        self.__invoker.services.board_image_records.add_image_to_board(
+                            board_id=board_id, image_name=image_name
+                        )
+                    except Exception as e:
+                        self.__invoker.services.logger.warning(f"Failed to add image to board {board_id}: {str(e)}")
+                self.__invoker.services.image_files.save(
+                    image_name=image_name,
+                    image=image,
+                    metadata=metadata,
+                    workflow=workflow,
+                    graph=graph,
+                    image_subfolder=image_subfolder,
+                )
             image_dto = self.get_dto(image_name)
 
             self._on_changed(image_dto)
@@ -106,24 +136,77 @@ def create(
             raise
         except ImageFileSaveException:
             self.__invoker.services.logger.error("Failed to save image file")
+            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 — and, like every other delete path,
+        the record is only removed once that journal is durable. If it cannot be, the half-created
+        image is left whole: its record marks the files for a later delete to find, whereas a record
+        removed without a journal would leave them orphaned for good.
+        """
+        with self._image_mutation_lock():
+            # A subfolder move that ran while the file save was failing may have relocated the
+            # just-saved record, and partial files can be left at either path. Journal both the
+            # captured subfolder and the one the record names now. A record that is already gone
+            # (a concurrent delete won the race) or a faulting record store leaves us only the
+            # captured subfolder, which is the best path known either way.
+            delete_subfolders = [image_subfolder]
             try:
-                self.__invoker.services.image_files.delete(image_name, image_subfolder=image_subfolder)
+                record = self.__invoker.services.image_records.get(image_name)
+                if record.image_subfolder != image_subfolder:
+                    delete_subfolders.append(record.image_subfolder)
+            except Exception as lookup_error:
+                # Best effort: a record-store fault here must not mask the ImageFileSaveException
+                # this cleanup is running for. We may then purge only the captured subfolder, so
+                # say so — if the record had been relocated, its files need a manual sweep.
+                self.__invoker.services.logger.warning(
+                    f"Could not confirm the subfolder of {image_name} during save-failure cleanup: {str(lookup_error)}"
+                )
+            try:
+                token = self.__invoker.services.image_files.begin_delete(
+                    [(image_name, subfolder) for subfolder in delete_subfolders]
+                )
             except Exception as cleanup_error:
+                # No durable journal, no record deletion: the record is what keeps these files
+                # findable. Deleting it and then purging blind would fail at the same journal step
+                # and leave whatever survived the save as an orphan nothing can recover.
                 self.__invoker.services.logger.error(
-                    f"Failed to clean up image files after save failure: {str(cleanup_error)}"
+                    f"Failed to journal the cleanup of {image_name} after a save failure; leaving the image in place "
+                    f"for a later delete: {str(cleanup_error)}"
                 )
+                return
             try:
-                # Deleting the record also removes any board association through the database
-                # foreign key cascade. Both cleanup operations are attempted independently.
+                # 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)}"
                 )
-            raise
-        except Exception as e:
-            self.__invoker.services.logger.error(f"Problem saving image record and file: {str(e)}")
-            raise e
+                # The record survived, so the image is still referenced; its files must stay with it.
+                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:
+                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,
@@ -290,93 +373,167 @@ def get_many(
             raise e
 
     def delete(self, image_name: str):
-        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)
-            self._on_deleted(image_name)
-        except ImageRecordDeleteException:
-            self.__invoker.services.logger.error("Failed to delete image record")
-            raise
-        except ImageFileDeleteException:
-            self.__invoker.services.logger.error("Failed to delete image file")
-            raise
-        except Exception as e:
-            self.__invoker.services.logger.error("Problem deleting image record and file")
-            raise e
-
-    def delete_images_on_board(self, board_id: str, user_id: Optional[str] = None) -> tuple[list[str], list[str]]:
-        try:
-            # When ``user_id`` is set the lookup filters to images owned by that user so the
-            # cascade doesn't destroy other users' contributions to a public/shared board.
-            image_names = self.__invoker.services.board_image_records.get_all_board_image_names_for_board(
-                board_id,
-                categories=None,
-                is_intermediate=None,
-                user_id=user_id,
-            )
-            deleted_image_names: list[str] = []
-            failed_image_names: list[str] = []
-            staged_deletes: list[tuple[str, object]] = []
-            for image_name in image_names:
-                try:
-                    record = self.__invoker.services.image_records.get(image_name)
-                    token = self.__invoker.services.image_files.stage_delete(
-                        image_name, image_subfolder=record.image_subfolder
-                    )
-                    staged_deletes.append((image_name, token))
-                    deleted_image_names.append(image_name)
-                except Exception as e:
-                    failed_image_names.append(image_name)
-                    self.__invoker.services.logger.error(
-                        f"Failed to delete image file {image_name}; keeping record: {str(e)}"
-                    )
+        # 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.
+        # The mutation lock spans the record read through the purge so a subfolder move cannot
+        # relocate the files between the two and leave the purge sweeping an abandoned path.
+        with self._image_mutation_lock():
             try:
-                self.__invoker.services.image_records.delete_many(deleted_image_names)
-            except Exception:
-                for image_name, token in staged_deletes:
+                record = self.__invoker.services.image_records.get(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.rollback_delete(token)
-                    except Exception as rollback_error:
+                        self.__invoker.services.image_files.abandon_delete(token)
+                    except Exception as cleanup_error:
                         self.__invoker.services.logger.error(
-                            f"Failed to restore staged image files for {image_name}: {rollback_error}"
+                            f"Failed to discard the delete journal for {image_name}: {cleanup_error}"
                         )
-                raise
-            for _, token in staged_deletes:
+                    raise
                 try:
                     self.__invoker.services.image_files.commit_delete(token)
                 except Exception as cleanup_error:
-                    self.__invoker.services.logger.error(f"Failed to purge staged image files: {cleanup_error}")
-            for image_name in deleted_image_names:
+                    # 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)
-            return deleted_image_names, failed_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(f"Problem deleting image records and files: {str(e)}")
-            raise e
+            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
+            except ImageFileDeleteException:
+                self.__invoker.services.logger.error("Failed to delete image file")
+                raise
+            except Exception as e:
+                self.__invoker.services.logger.error("Problem deleting image record and file")
+                raise e
+
+    def delete_images_on_board(self, board_id: str, user_id: Optional[str] = None) -> tuple[list[str], list[str]]:
+        # The mutation lock spans the per-image record reads through the purges and rollbacks so a
+        # subfolder move cannot relocate files between a record read and its stage or commit.
+        with self._image_mutation_lock():
+            try:
+                # When ``user_id`` is set the lookup filters to images owned by that user so the
+                # cascade doesn't destroy other users' contributions to a public/shared board.
+                image_names = self.__invoker.services.board_image_records.get_all_board_image_names_for_board(
+                    board_id,
+                    categories=None,
+                    is_intermediate=None,
+                    user_id=user_id,
+                )
+                deleted_image_names: list[str] = []
+                failed_image_names: list[str] = []
+                staged_deletes: list[tuple[str, object]] = []
+                for image_name in image_names:
+                    try:
+                        record = self.__invoker.services.image_records.get(image_name)
+                        token = self.__invoker.services.image_files.stage_delete(
+                            image_name, image_subfolder=record.image_subfolder
+                        )
+                        staged_deletes.append((image_name, token))
+                        deleted_image_names.append(image_name)
+                    except Exception as e:
+                        failed_image_names.append(image_name)
+                        self.__invoker.services.logger.error(
+                            f"Failed to delete image file {image_name}; keeping record: {str(e)}"
+                        )
+                try:
+                    self.__invoker.services.image_records.delete_many(deleted_image_names)
+                except Exception:
+                    for image_name, token in staged_deletes:
+                        try:
+                            self.__invoker.services.image_files.rollback_delete(token)
+                        except Exception as rollback_error:
+                            self.__invoker.services.logger.error(
+                                f"Failed to restore staged image files for {image_name}: {rollback_error}"
+                            )
+                    raise
+                for _, token in staged_deletes:
+                    try:
+                        self.__invoker.services.image_files.commit_delete(token)
+                    except Exception as cleanup_error:
+                        self.__invoker.services.logger.error(f"Failed to purge staged image files: {cleanup_error}")
+                for image_name in deleted_image_names:
+                    self._on_deleted(image_name)
+                return deleted_image_names, failed_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(f"Problem deleting image records and files: {str(e)}")
+                raise e
 
     def delete_intermediates(self) -> int:
-        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)
-                self._on_deleted(image_name)
-            return count
-        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")
-            raise e
+        # 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.
+        #
+        # The mutation lock spans the snapshot through the purge so a subfolder move cannot
+        # relocate files between the two and leave the purge sweeping an abandoned path
+        # (JPPhoto, PR #9361).
+        with self._image_mutation_lock():
+            try:
+                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 len(deleted_image_names)
+            except ImageRecordDeleteException:
+                self.__invoker.services.logger.error("Failed to delete image records")
+                raise
+            except Exception as e:
+                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:
         try:
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..273f64151fb 100644
--- a/tests/app/services/image_files/test_image_files_disk.py
+++ b/tests/app/services/image_files/test_image_files_disk.py
@@ -1,5 +1,9 @@
+import errno
 import hashlib
+import os
 import platform
+import shutil
+import stat
 import zlib
 from pathlib import Path
 from unittest.mock import MagicMock, patch
@@ -7,12 +11,35 @@
 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
+
+
+posix_only = pytest.mark.skipif(os.name == "nt", reason="Windows cannot open a directory for fsync")
+
+
+def _failing_directory_fsync(error_number: int):
+    """An ``os.fsync`` that fails for directory descriptors only; file fsyncs still work."""
+    real_fsync = os.fsync
+
+    def fsync(fd: int) -> None:
+        if stat.S_ISDIR(os.fstat(fd).st_mode):
+            raise OSError(error_number, os.strerror(error_number))
+        real_fsync(fd)
+
+    return fsync
+
+
 @pytest.fixture
 def image_names() -> list[str]:
     # Determine the platform and return a path that matches its format
@@ -42,6 +69,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 +389,489 @@ 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]
+
+    @posix_only
+    def test_begin_delete_fails_when_the_journal_cannot_be_made_durable(self, disk_storage: DiskImageFileStorage):
+        """A journal whose durability is unknown must not license a record deletion: the caller
+        deletes records straight after this returns, so the failure has to surface here."""
+        disk_storage.save(image=Image.new("RGB", (32, 32)), image_name="live.png")
+
+        with patch.object(os, "fsync", _failing_directory_fsync(errno.EIO)), pytest.raises(ImageFileDeleteException):
+            disk_storage.begin_delete([("live.png", "")])
+
+        assert not list(disk_storage.image_root.glob(".delete_*"))
+        assert disk_storage.get_path("live.png").exists()
+        assert disk_storage.get_path("live.png", thumbnail=True).exists()
+
+    @posix_only
+    def test_stage_delete_fails_before_moving_anything_when_the_journal_cannot_be_made_durable(
+        self, disk_storage: DiskImageFileStorage
+    ):
+        disk_storage.save(image=Image.new("RGB", (32, 32)), image_name="live.png")
+
+        with patch.object(os, "fsync", _failing_directory_fsync(errno.EIO)), pytest.raises(ImageFileDeleteException):
+            disk_storage.stage_delete("live.png")
+
+        assert not list(disk_storage.image_root.glob(".delete_*"))
+        assert disk_storage.get_path("live.png").exists()
+        assert disk_storage.get_path("live.png", thumbnail=True).exists()
+
+    @posix_only
+    @pytest.mark.parametrize("error_number", [errno.EINVAL, errno.ENOTSUP])
+    def test_a_filesystem_that_cannot_sync_directories_is_not_a_failure(
+        self, disk_storage: DiskImageFileStorage, error_number: int
+    ):
+        """Some filesystems refuse directory fsync outright; there is nothing more to be had from
+        them, and refusing every delete on such a volume would be worse than the risk."""
+        disk_storage.save(image=Image.new("RGB", (32, 32)), image_name="gone.png")
+
+        with patch.object(os, "fsync", _failing_directory_fsync(error_number)):
+            token = disk_storage.begin_delete([("gone.png", "")])
+            disk_storage.commit_delete(token)
+
+        assert not disk_storage.get_path("gone.png").exists()
+        assert not list(disk_storage.image_root.glob(".delete_*"))
+
+
+class TestPurgeDurability:
+    """The journal may only be dropped once the purge it describes is on disk.
+
+    Unlinks live in the parent directories' entries and nothing else syncs those. A filesystem that
+    persists the journal's removal ahead of the unlinks brings the files back after a power loss
+    with no journal left to find them by — the permanent orphan the journal exists to prevent.
+    """
+
+    SUBFOLDER = "2026/09/04"
+
+    def _parents(self, disk_storage: DiskImageFileStorage, image_name: str) -> set[Path]:
+        return {
+            disk_storage.get_path(image_name, image_subfolder=self.SUBFOLDER).parent.resolve(),
+            disk_storage.get_path(image_name, thumbnail=True, image_subfolder=self.SUBFOLDER).parent.resolve(),
+        }
+
+    def _assert_parents_synced_before_journal_removal(self, disk_storage, monkeypatch, image_name: str):
+        """Returns the list the patched fsync records into; rmtree of the journal asserts on it."""
+        fsynced: list[Path] = []
+        monkeypatch.setattr(
+            DiskImageFileStorage,
+            "_DiskImageFileStorage__fsync_directory",
+            staticmethod(lambda directory, **_: fsynced.append(Path(directory).resolve())),
+        )
+        real_rmtree = shutil.rmtree
+        parents = self._parents(disk_storage, image_name)
+
+        def rmtree(path, *args, **kwargs):
+            if Path(path).name.startswith(".delete_"):
+                assert parents <= set(fsynced), "the journal was dropped before the purge was made durable"
+            return real_rmtree(path, *args, **kwargs)
+
+        monkeypatch.setattr(shutil, "rmtree", rmtree)
+        return fsynced
+
+    def test_pending_commit_syncs_the_purged_directories_before_dropping_the_journal(
+        self, disk_storage: DiskImageFileStorage, monkeypatch
+    ):
+        disk_storage.save(image=Image.new("RGB", (32, 32)), image_name="gone.png", image_subfolder=self.SUBFOLDER)
+        token = disk_storage.begin_delete([("gone.png", self.SUBFOLDER)])
+        fsynced = self._assert_parents_synced_before_journal_removal(disk_storage, monkeypatch, "gone.png")
+
+        disk_storage.commit_delete(token)
+
+        assert self._parents(disk_storage, "gone.png") <= set(fsynced)
+        assert not list(disk_storage.image_root.glob(".delete_*"))
+
+    def test_staged_commit_syncs_the_purged_directories_before_dropping_the_journal(
+        self, disk_storage: DiskImageFileStorage, monkeypatch
+    ):
+        disk_storage.save(image=Image.new("RGB", (32, 32)), image_name="gone.png", image_subfolder=self.SUBFOLDER)
+        token = disk_storage.stage_delete("gone.png", image_subfolder=self.SUBFOLDER)
+        fsynced = self._assert_parents_synced_before_journal_removal(disk_storage, monkeypatch, "gone.png")
+
+        disk_storage.commit_delete(token)
+
+        assert self._parents(disk_storage, "gone.png") <= set(fsynced)
+        assert not list(disk_storage.image_root.glob(".delete_*"))
+
+    def test_rollback_syncs_the_restored_directories_before_dropping_the_journal(
+        self, disk_storage: DiskImageFileStorage, monkeypatch
+    ):
+        disk_storage.save(image=Image.new("RGB", (32, 32)), image_name="back.png", image_subfolder=self.SUBFOLDER)
+        token = disk_storage.stage_delete("back.png", image_subfolder=self.SUBFOLDER)
+        fsynced = self._assert_parents_synced_before_journal_removal(disk_storage, monkeypatch, "back.png")
+
+        disk_storage.rollback_delete(token)
+
+        assert self._parents(disk_storage, "back.png") <= set(fsynced)
+        assert disk_storage.get_path("back.png", image_subfolder=self.SUBFOLDER).exists()
+        assert not list(disk_storage.image_root.glob(".delete_*"))
+
+    def test_startup_syncs_before_dropping_a_recovered_journal(self, disk_storage: DiskImageFileStorage, monkeypatch):
+        disk_storage.save(image=Image.new("RGB", (32, 32)), image_name="orphan.png", image_subfolder=self.SUBFOLDER)
+        disk_storage.begin_delete([("orphan.png", self.SUBFOLDER)])
+        fsynced = self._assert_parents_synced_before_journal_removal(disk_storage, monkeypatch, "orphan.png")
+
+        _restart(disk_storage, record_exists=False)
+
+        assert self._parents(disk_storage, "orphan.png") <= set(fsynced)
+        assert not disk_storage.get_path("orphan.png", image_subfolder=self.SUBFOLDER).exists()
+        assert not list(disk_storage.image_root.glob(".delete_*"))
+
+    @posix_only
+    def test_commit_keeps_the_journal_when_the_purge_cannot_be_made_durable(self, disk_storage: DiskImageFileStorage):
+        """The records are already gone, so the journal is the only thing that can redo a purge the
+        disk lost. It stays, and the next startup finishes the job."""
+        disk_storage.save(image=Image.new("RGB", (32, 32)), image_name="gone.png", image_subfolder=self.SUBFOLDER)
+        token = disk_storage.begin_delete([("gone.png", self.SUBFOLDER)])
+
+        with patch.object(os, "fsync", _failing_directory_fsync(errno.EIO)), pytest.raises(ImageFileDeleteException):
+            disk_storage.commit_delete(token)
+
+        assert list(disk_storage.image_root.glob(".delete_*"))
+
+        _restart(disk_storage, record_exists=False)
+
+        assert not disk_storage.get_path("gone.png", image_subfolder=self.SUBFOLDER).exists()
+        assert not disk_storage.get_path("gone.png", thumbnail=True, image_subfolder=self.SUBFOLDER).exists()
+        assert not list(disk_storage.image_root.glob(".delete_*"))
+
+    @posix_only
+    def test_staged_commit_keeps_the_journal_when_the_purge_cannot_be_made_durable(
+        self, disk_storage: DiskImageFileStorage
+    ):
+        disk_storage.save(image=Image.new("RGB", (32, 32)), image_name="gone.png")
+        token = disk_storage.stage_delete("gone.png")
+
+        with patch.object(os, "fsync", _failing_directory_fsync(errno.EIO)), pytest.raises(ImageFileDeleteException):
+            disk_storage.commit_delete(token)
+
+        assert list(disk_storage.image_root.glob(".delete_*"))
+
+        _restart(disk_storage, record_exists=False)
+
+        assert not disk_storage.get_path("gone.png").exists()
+        assert not list(disk_storage.image_root.glob(".delete_*"))
+
+    @posix_only
+    def test_startup_keeps_the_journal_when_the_purge_cannot_be_made_durable(self, disk_storage: DiskImageFileStorage):
+        """Injected fsync failure at the recovery step itself: the journal must survive that
+        startup and be finished by the next one."""
+        disk_storage.save(image=Image.new("RGB", (32, 32)), image_name="orphan.png")
+        disk_storage.begin_delete([("orphan.png", "")])
+
+        with patch.object(os, "fsync", _failing_directory_fsync(errno.EIO)):
+            _restart(disk_storage, record_exists=False)
+
+        assert list(disk_storage.image_root.glob(".delete_*"))
+
+        _restart(disk_storage, record_exists=False)
+
+        assert not disk_storage.get_path("orphan.png").exists()
+        assert not list(disk_storage.image_root.glob(".delete_*"))
+
+    def test_a_subfolder_that_no_longer_exists_does_not_block_the_commit(self, disk_storage: DiskImageFileStorage):
+        """A journal can name a subfolder whose directory is gone entirely (moved or already purged);
+        there are no entries there to make durable, so the commit completes."""
+        token = disk_storage.begin_delete([("ghost.png", "never/made")])
+
+        disk_storage.commit_delete(token)
+
+        assert not list(disk_storage.image_root.glob(".delete_*"))
+
+
+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..3d8595d25f4 100644
--- a/tests/app/services/images/test_images_default.py
+++ b/tests/app/services/images/test_images_default.py
@@ -1,9 +1,14 @@
 """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().
 """
 
+import errno
+import os
+import stat
+import threading
 from pathlib import Path
 from unittest.mock import MagicMock, patch
 
@@ -12,12 +17,16 @@
 
 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_moves.image_moves_default import ImageMoveResult, ImageMoveService
 from invokeai.app.services.image_records.image_records_common import (
     ImageCategory,
     ImageRecord,
+    ImageRecordChanges,
+    ImageRecordDeleteException,
     ImageRecordNotFoundException,
     ResourceOrigin,
 )
@@ -26,6 +35,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 +49,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 +277,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 +293,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 +376,968 @@ 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_*"))
+
+
+def _failing_directory_fsync(error_number: int):
+    """An ``os.fsync`` that fails for directory descriptors only; file fsyncs still work."""
+    real_fsync = os.fsync
+
+    def fsync(fd: int) -> None:
+        if stat.S_ISDIR(os.fstat(fd).st_mode):
+            raise OSError(error_number, os.strerror(error_number))
+        real_fsync(fd)
+
+    return fsync
+
+
+@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
+    invoker.services.image_moves = None
+    storage.start(invoker)
+    svc.start(invoker)
+    return svc, records, storage
+
+
+@pytest.fixture
+def wired_with_move_service(
+    tmp_path: Path,
+) -> tuple[ImageService, SqliteImageRecordStorage, DiskImageFileStorage, ImageMoveService]:
+    """ImageService and ImageMoveService wired to one real db and disk store, as production does."""
+    config = InvokeAIAppConfig(use_memory_db=True, image_subfolder_strategy="flat")
+    logger = InvokeAILogger.get_logger(config=config)
+    db = create_mock_sqlite_database(config, logger)
+    records = SqliteImageRecordStorage(db=db)
+    storage = DiskImageFileStorage(tmp_path / "outputs")
+    moves = ImageMoveService(db=db, image_files=storage, config=config, logger=logger)
+
+    svc = ImageService()
+    invoker = MagicMock()
+    invoker.services.configuration.pil_compress_level = 1
+    invoker.services.image_records = records
+    invoker.services.image_files = storage
+    invoker.services.image_moves = moves
+    invoker.services.names.create_image_name.return_value = "raced.png"
+    invoker.services.urls.get_image_url.return_value = "/api/v1/images/i/raced.png"
+    invoker.services.board_image_records.get_board_for_image.return_value = None
+    storage.start(invoker)
+    svc.start(invoker)
+    return svc, records, storage, moves
+
+
+def _seed_record_at_subfolder(records: SqliteImageRecordStorage, name: str, image_subfolder: str) -> None:
+    records.save(
+        image_name=name,
+        image_origin=ResourceOrigin.INTERNAL,
+        image_category=ImageCategory.GENERAL,
+        width=64,
+        height=64,
+        has_workflow=False,
+        is_intermediate=True,
+        image_subfolder=image_subfolder,
+    )
+
+
+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
+        )
+
+    def test_the_record_is_kept_when_the_cleanup_cannot_be_journaled(self, image_service: ImageService) -> None:
+        """No durable journal, no record deletion. Removing the record and purging blind would fail
+        at the same journal step and leave whatever survived the save as an orphan nothing can find;
+        a surviving record is what lets a later delete find and clear it."""
+        invoker = image_service._ImageService__invoker  # type: ignore
+        invoker.services.image_files.save.side_effect = ImageFileSaveException()
+        invoker.services.image_files.begin_delete.side_effect = ImageFileDeleteException("fsync: I/O error")
+
+        with pytest.raises(ImageFileSaveException):
+            image_service.create(
+                image=Image.new("RGB", (8, 8)),
+                image_origin=ResourceOrigin.INTERNAL,
+                image_category=ImageCategory.GENERAL,
+            )
+
+        invoker.services.image_records.delete.assert_not_called()
+        invoker.services.image_files.commit_delete.assert_not_called()
+        invoker.services.image_files.delete.assert_not_called()
+
+    @pytest.mark.skipif(os.name == "nt", reason="Windows cannot open a directory for fsync")
+    def test_a_leftover_file_is_never_orphaned_without_a_journal(
+        self,
+        real_image_service: tuple[ImageService, SqliteImageRecordStorage, DiskImageFileStorage],
+    ) -> None:
+        """Real storage, real records: the save leaves a file behind and the disk refuses to make
+        the cleanup journal durable. The record must survive, and an ordinary delete once the disk
+        is healthy must be able to clear the whole image."""
+        service, records, storage = real_image_service
+        # save() removes only what it created, so a file that already sits at the image's path is
+        # what survives a failed save.
+        leftover = storage.get_path("uploaded.png")
+        leftover.parent.mkdir(parents=True, exist_ok=True)
+        leftover.write_bytes(b"pre-existing")
+        broken_thumbnail = MagicMock()
+        broken_thumbnail.save.side_effect = OSError("thumbnail filesystem failure")
+
+        with (
+            patch(
+                "invokeai.app.services.image_files.image_files_disk.make_thumbnail",
+                return_value=broken_thumbnail,
+            ),
+            patch.object(os, "fsync", _failing_directory_fsync(errno.EIO)),
+            pytest.raises(ImageFileSaveException),
+        ):
+            service.create(
+                image=Image.new("RGB", (32, 32), "red"),
+                image_origin=ResourceOrigin.EXTERNAL,
+                image_category=ImageCategory.GENERAL,
+            )
+
+        assert records.get("uploaded.png").image_name == "uploaded.png"
+        assert leftover.exists()
+        assert _staging_dirs(storage) == []
+
+        service.delete("uploaded.png")
+
+        with pytest.raises(ImageRecordNotFoundException):
+            records.get("uploaded.png")
+        assert not leftover.exists()
+        assert _staging_dirs(storage) == []
+
+
+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) == []
+
+
+class TestDeleteVersusSubfolderMove:
+    """The image-mutation lock that serializes delete units against subfolder relocations.
+
+    A delete reads an image's subfolder, deletes its record, then purges its files at that
+    subfolder; a move relocates the files and repoints the record. If the two interleave, the
+    delete purges the path its snapshot named while the files sit at the new one — permanent
+    orphans, unrecoverable because the record is gone and a clean purge drops the journal
+    (JPPhoto, PR #9361). These tests drive both real services against one real db and disk
+    store and pin the serialization from both directions.
+    """
+
+    def test_move_cannot_interleave_with_delete_intermediates(
+        self,
+        wired_with_move_service: tuple[ImageService, SqliteImageRecordStorage, DiskImageFileStorage, ImageMoveService],
+        monkeypatch,
+    ) -> None:
+        """While delete_intermediates() runs its unit, a concurrent move_all_images() must wait.
+
+        Without the lock the move relocates the image mid-delete and the delete's purge sweeps
+        the abandoned path: the files survive at the new subfolder with no record and no journal.
+        """
+        svc, records, storage, moves = wired_with_move_service
+        _seed_record_at_subfolder(records, "raced.png", "old")
+        _save_image_file(storage, "raced.png", image_subfolder="old")
+
+        delete_started = threading.Event()
+        delete_may_finish = threading.Event()
+        real_delete_by_names = records.delete_intermediates_by_names
+
+        def delete_after_pause(names: list[str]) -> list[str]:
+            delete_started.set()
+            assert delete_may_finish.wait(timeout=10), "the move never let the delete finish"
+            return real_delete_by_names(names)
+
+        monkeypatch.setattr(records, "delete_intermediates_by_names", delete_after_pause)
+
+        delete_thread = threading.Thread(target=svc.delete_intermediates)
+        delete_thread.start()
+        assert delete_started.wait(timeout=10), "the delete never reached its record deletion"
+
+        # The move finds the row (the delete has not removed it yet) and has a relocation to
+        # perform (the row says "old"; the strategy says the flat root). It must wait for the
+        # delete's unit instead of racing it.
+        move_result: list[object] = []
+        move_thread = threading.Thread(target=lambda: move_result.append(moves.move_all_images()))
+        move_thread.start()
+        move_thread.join(timeout=1.0)
+        assert move_thread.is_alive(), "move_all_images finished while a delete held the mutation lock"
+        assert storage.get_path("raced.png", image_subfolder="old").exists(), (
+            "the move relocated files while a delete was mid-unit"
+        )
+
+        delete_may_finish.set()
+        delete_thread.join(timeout=10)
+        move_thread.join(timeout=10)
+        assert not delete_thread.is_alive()
+        assert not move_thread.is_alive()
+
+        result = move_result[0]
+        assert isinstance(result, ImageMoveResult)
+        assert result.errors == 0
+        with pytest.raises(ImageRecordNotFoundException):
+            records.get("raced.png")
+        assert not storage.get_path("raced.png", image_subfolder="old").exists()
+        assert not storage.get_path("raced.png").exists()
+        assert not storage.get_path("raced.png", thumbnail=True).exists()
+        assert _staging_dirs(storage) == []
+
+    def test_move_holds_the_lock_across_its_batch_cycle(
+        self,
+        wired_with_move_service: tuple[ImageService, SqliteImageRecordStorage, DiskImageFileStorage, ImageMoveService],
+    ) -> None:
+        """While a move unit runs, a concurrent delete must wait for it rather than race it.
+
+        The lock is held across the whole plan-relocate-repoint cycle: nothing may be relocated
+        or repointed, and the cycle may not complete, while another party holds the lock.
+        """
+        svc, records, storage, moves = wired_with_move_service
+        _seed_record_at_subfolder(records, "moved.png", "old")
+        _save_image_file(storage, "moved.png", image_subfolder="old")
+
+        with moves.image_mutation_lock():
+            move_result: list[object] = []
+            move_thread = threading.Thread(target=lambda: move_result.append(moves.move_all_images()))
+            move_thread.start()
+            move_thread.join(timeout=1.0)
+            assert move_thread.is_alive(), "move_all_images completed while the mutation lock was held"
+            assert storage.get_path("moved.png", image_subfolder="old").exists()
+            assert records.get("moved.png").image_subfolder == "old"
+
+        move_thread.join(timeout=10)
+        assert not move_thread.is_alive()
+
+        result = move_result[0]
+        assert isinstance(result, ImageMoveResult)
+        assert (result.planned, result.committed, result.errors) == (1, 1, 0)
+        assert not storage.get_path("moved.png", image_subfolder="old").exists()
+        assert storage.get_path("moved.png").exists()
+        assert records.get("moved.png").image_subfolder == ""
+
+    def test_failed_save_cleanup_purges_the_relocated_subfolder_too(
+        self,
+        wired_with_move_service: tuple[ImageService, SqliteImageRecordStorage, DiskImageFileStorage, ImageMoveService],
+    ) -> None:
+        """A failed save whose record was relocated mid-save must not strand the moved files.
+
+        The cleanup captured the subfolder the save used, but a move that completed before the
+        cleanup began repointed the record; the purge has to cover the path the record names now
+        as well as the one the save wrote to.
+        """
+        svc, records, storage, moves = wired_with_move_service
+        _seed_record_at_subfolder(records, "failed.png", "old")
+        _save_image_file(storage, "failed.png", image_subfolder="old")
+
+        # A subfolder relocation completes before the cleanup runs: files moved, record repointed.
+        # Simulated directly — a real move job would leave item rows whose foreign key blocks the
+        # record delete, a pre-existing limitation on main unrelated to this lock.
+        old_image = storage.get_path("failed.png", image_subfolder="old")
+        old_thumbnail = storage.get_path("failed.png", thumbnail=True, image_subfolder="old")
+        new_image = storage.get_path("failed.png")
+        new_thumbnail = storage.get_path("failed.png", thumbnail=True)
+        new_image.parent.mkdir(parents=True, exist_ok=True)
+        new_thumbnail.parent.mkdir(parents=True, exist_ok=True)
+        old_image.replace(new_image)
+        old_thumbnail.replace(new_thumbnail)
+        with records._db.transaction() as cursor:
+            cursor.execute("UPDATE images SET image_subfolder = '' WHERE image_name = 'failed.png';")
+        assert records.get("failed.png").image_subfolder == ""
+
+        # The failed save's cleanup still holds the subfolder the save captured.
+        svc._ImageService__clean_up_failed_save("failed.png", "old")  # type: ignore[attr-defined]
+
+        with pytest.raises(ImageRecordNotFoundException):
+            records.get("failed.png")
+        assert not new_image.exists()
+        assert not new_thumbnail.exists()
+        assert not old_image.exists()
+        assert not old_thumbnail.exists()
+        assert _staging_dirs(storage) == []
+
+    def test_move_cannot_interleave_with_create(
+        self,
+        wired_with_move_service: tuple[ImageService, SqliteImageRecordStorage, DiskImageFileStorage, ImageMoveService],
+        monkeypatch,
+    ) -> None:
+        """While create() runs its save unit, a concurrent move_all_images() must wait.
+
+        The record becomes visible to the move planner the moment it commits, but its files do
+        not exist until the save lands. Under the date strategy the two subfolders can even
+        disagree by a day — the strategy reads the local clock while the move target comes from
+        the record's UTC created_at — so a relocation landing mid-save leaves the record naming
+        a subfolder the files were never written to.
+        """
+        svc, records, storage, moves = wired_with_move_service
+
+        # The record this create() writes lands at a subfolder the move service's strategy
+        # (flat) does not consider final — the situation the date strategy produces on any
+        # non-UTC host. The subfolder directory must exist, or the move errors in preflight
+        # before it can repoint anything.
+        class _StaleSubfolderStrategy:
+            def get_subfolder(self, image_name, image_category, is_intermediate):
+                return "old"
+
+        monkeypatch.setattr(
+            "invokeai.app.services.images.images_default.create_subfolder_strategy",
+            lambda _strategy_name: _StaleSubfolderStrategy(),
+        )
+        storage.get_path("raced.png", image_subfolder="old").parent.mkdir(parents=True, exist_ok=True)
+
+        save_started = threading.Event()
+        save_may_finish = threading.Event()
+        real_save = storage.save
+
+        def save_then_pause(**kwargs):
+            save_started.set()
+            assert save_may_finish.wait(timeout=10), "the move never let the create finish"
+            real_save(**kwargs)
+
+        monkeypatch.setattr(storage, "save", save_then_pause)
+
+        create_thread = threading.Thread(
+            target=lambda: svc.create(
+                image=Image.new("RGB", (64, 64)),
+                image_origin=ResourceOrigin.EXTERNAL,
+                image_category=ImageCategory.GENERAL,
+            )
+        )
+        create_thread.start()
+        assert save_started.wait(timeout=10), "the create never reached its file save"
+        assert records.get("raced.png").image_subfolder == "old"
+
+        move_result: list[object] = []
+        move_thread = threading.Thread(target=lambda: move_result.append(moves.move_all_images()))
+        move_thread.start()
+        move_thread.join(timeout=1.0)
+        assert move_thread.is_alive(), "move_all_images completed while create() held the mutation lock"
+        assert records.get("raced.png").image_subfolder == "old", (
+            "the move repointed the record while create() was still writing its files"
+        )
+
+        save_may_finish.set()
+        create_thread.join(timeout=10)
+        move_thread.join(timeout=10)
+        assert not create_thread.is_alive()
+        assert not move_thread.is_alive()
+
+        # The move then relocates the completed image: the record and the files end up agreeing.
+        assert records.get("raced.png").image_subfolder == ""
+        assert storage.get_path("raced.png").exists()
+        assert not storage.get_path("raced.png", image_subfolder="old").exists()
+        assert storage.get_path("raced.png", thumbnail=True).exists()
+        assert _staging_dirs(storage) == []