feat(ui): (optional) Protect starred media from accidental deletion

invokeai/app/api/routers/boards.py

@@ -38,6 +38,14 @@ class DeleteBoardResult(BaseModel):
         default_factory=list,
         description="The names of videos that could not be deleted and became uncategorized.",
     )
+    starred_images_skipped: list[str] = Field(
+        default_factory=list,
+        description="The names of starred images that were protected and became uncategorized.",
+    )
+    starred_videos_skipped: list[str] = Field(
+        default_factory=list,
+        description="The names of starred videos that were protected and became uncategorized.",
+    )
 
 
 @boards_router.post(
@@ -125,6 +133,7 @@ def delete_board(
     include_images: Optional[bool] = Query(
         description="Permanently delete all images and videos on the board", default=False
     ),
+    delete_starred: bool = Query(default=True, description="Whether to allow deletion of starred media"),
 ) -> DeleteBoardResult:
     """Deletes a board (user must have access to it)"""
     try:
@@ -141,6 +150,10 @@ def delete_board(
     cascade_user_id: Optional[str] = None if current_user.is_admin else current_user.user_id
     deleted_images: list[str] = []
     deleted_videos: list[str] = []
+    failed_images: list[str] = []
+    failed_videos: list[str] = []
+    starred_images_skipped: list[str] = []
+    starred_videos_skipped: list[str] = []
 
     try:
         if include_images is True:
@@ -151,11 +164,19 @@ def delete_board(
             # truth — reconstructing failures by diffing a router-side board listing
             # against the deleted names would double the DB work and misreport items
             # moved or deleted concurrently between the two queries.
-            deleted_images, failed_images = ApiDependencies.invoker.services.images.delete_images_on_board(
-                board_id=board_id, user_id=cascade_user_id
+            (
+                deleted_images,
+                failed_images,
+                starred_images_skipped,
+            ) = ApiDependencies.invoker.services.images.delete_images_on_board(
+                board_id=board_id, user_id=cascade_user_id, delete_starred=delete_starred
             )
-            deleted_videos, failed_videos = ApiDependencies.invoker.services.videos.delete_videos_on_board(
-                board_id=board_id, user_id=cascade_user_id
+            (
+                deleted_videos,
+                failed_videos,
+                starred_videos_skipped,
+            ) = ApiDependencies.invoker.services.videos.delete_videos_on_board(
+                board_id=board_id, user_id=cascade_user_id, delete_starred=delete_starred
             )
             ApiDependencies.invoker.services.boards.delete(board_id=board_id)
             return DeleteBoardResult(
@@ -166,6 +187,8 @@ def delete_board(
                 deleted_videos=deleted_videos,
                 failed_images=failed_images,
                 failed_videos=failed_videos,
+                starred_images_skipped=starred_images_skipped,
+                starred_videos_skipped=starred_videos_skipped,
             )
         else:
             deleted_board_images = ApiDependencies.invoker.services.board_images.get_all_board_image_names_for_board(
@@ -198,6 +221,10 @@ def delete_board(
                     "message": "Failed to delete board after partially deleting media",
                     "deleted_images": deleted_images,
                     "deleted_videos": deleted_videos,
+                    "failed_images": failed_images,
+                    "failed_videos": failed_videos,
+                    "starred_images_skipped": starred_images_skipped,
+                    "starred_videos_skipped": starred_videos_skipped,
                     "board_deleted": False,
                 },
             )

invokeai/app/api/routers/images.py

@@ -202,27 +202,34 @@ async def create_image_upload_entry(
 async def delete_image(
     current_user: CurrentUserOrDefault,
     image_name: str = Path(description="The name of the image to delete"),
+    delete_starred: bool = Query(default=True, description="Whether to allow deletion of starred images"),
 ) -> DeleteImagesResult:
     """Deletes an image"""
     _assert_image_owner(image_name, current_user)
     assert_image_move_maintenance_inactive()
 
     deleted_images: set[str] = set()
+    failed_images: set[str] = set()
     affected_boards: set[str] = set()
+    starred_skipped: set[str] = set()
 
     try:
-        image_dto = ApiDependencies.invoker.services.images.get_dto(image_name)
-        board_id = image_dto.board_id or "none"
-        ApiDependencies.invoker.services.images.delete(image_name)
-        deleted_images.add(image_name)
+        board_id = ApiDependencies.invoker.services.board_image_records.get_board_for_image(image_name) or "none"
+        was_deleted = ApiDependencies.invoker.services.images.delete(image_name, delete_starred=delete_starred)
+        if was_deleted:
+            deleted_images.add(image_name)
+        else:
+            starred_skipped.add(image_name)
         affected_boards.add(board_id)
-    except Exception:
-        # TODO: Does this need any exception handling at all?
-        pass
+    except Exception as error:
+        failed_images.add(image_name)
+        ApiDependencies.invoker.services.logger.error(f"Failed to delete image {image_name}: {error}")
 
     return DeleteImagesResult(
         deleted_images=list(deleted_images),
+        failed_images=list(failed_images),
         affected_boards=list(affected_boards),
+        starred_skipped=list(starred_skipped),
     )
 
 
@@ -489,6 +496,7 @@ async def list_image_dtos(
 async def delete_images_from_list(
     current_user: CurrentUserOrDefault,
     image_names: list[str] = Body(description="The list of names of images to delete", embed=True),
+    delete_starred: bool = Body(default=True, description="Whether to allow deletion of starred images"),
 ) -> DeleteImagesResult:
     try:
         assert_image_move_maintenance_inactive()
@@ -505,27 +513,34 @@ async def delete_images_from_list(
         deleted_images: set[str] = set()
         failed_images: set[str] = set()
         affected_boards: set[str] = set()
+        starred_skipped: set[str] = set()
         # Dedup while preserving order: a name repeated in the request would otherwise
         # be processed twice, and the second pass's not-found error would land the same
         # name in both deleted_images and failed_images.
         for image_name in dict.fromkeys(image_names):
             try:
                 _assert_image_owner(image_name, current_user)
-                image_dto = ApiDependencies.invoker.services.images.get_dto(image_name)
-                board_id = image_dto.board_id or "none"
-                ApiDependencies.invoker.services.images.delete(image_name)
-                deleted_images.add(image_name)
+                board_id = (
+                    ApiDependencies.invoker.services.board_image_records.get_board_for_image(image_name) or "none"
+                )
+                was_deleted = ApiDependencies.invoker.services.images.delete(image_name, delete_starred=delete_starred)
+                if was_deleted:
+                    deleted_images.add(image_name)
+                else:
+                    starred_skipped.add(image_name)
                 affected_boards.add(board_id)
             except HTTPException:
                 continue
-            except Exception:
+            except Exception as error:
                 # A genuine deletion failure (not an auth/404 skip) — report it so the
                 # client can surface a partial-failure warning, matching the video path.
                 failed_images.add(image_name)
+                ApiDependencies.invoker.services.logger.error(f"Failed to delete image {image_name}: {error}")
         return DeleteImagesResult(
             deleted_images=list(deleted_images),
             failed_images=list(failed_images),
             affected_boards=list(affected_boards),
+            starred_skipped=list(starred_skipped),
         )
     except HTTPException:
         raise
@@ -536,6 +551,7 @@ async def delete_images_from_list(
 @images_router.delete("/uncategorized", operation_id="delete_uncategorized_images", response_model=DeleteImagesResult)
 async def delete_uncategorized_images(
     current_user: CurrentUserOrDefault,
+    delete_starred: bool = Query(default=True, description="Whether to allow deletion of starred images"),
 ) -> DeleteImagesResult:
     """Deletes all uncategorized images owned by the current user (or all if admin)"""
     assert_image_move_maintenance_inactive()
@@ -548,21 +564,27 @@ async def delete_uncategorized_images(
         deleted_images: set[str] = set()
         failed_images: set[str] = set()
         affected_boards: set[str] = set()
+        starred_skipped: set[str] = set()
         for image_name in image_names:
             try:
                 _assert_image_owner(image_name, current_user)
-                ApiDependencies.invoker.services.images.delete(image_name)
-                deleted_images.add(image_name)
+                was_deleted = ApiDependencies.invoker.services.images.delete(image_name, delete_starred=delete_starred)
+                if was_deleted:
+                    deleted_images.add(image_name)
+                else:
+                    starred_skipped.add(image_name)
                 affected_boards.add("none")
             except HTTPException:
                 # Skip images not owned by the current user
-                pass
-            except Exception:
+                continue
+            except Exception as error:
                 failed_images.add(image_name)
+                ApiDependencies.invoker.services.logger.error(f"Failed to delete image {image_name}: {error}")
         return DeleteImagesResult(
             deleted_images=list(deleted_images),
             failed_images=list(failed_images),
             affected_boards=list(affected_boards),
+            starred_skipped=list(starred_skipped),
         )
     except Exception:
         raise HTTPException(status_code=500, detail="Failed to delete images")

invokeai/app/api/routers/videos.py

@@ -74,6 +74,10 @@ class VideoNamesBatch(BaseModel):
     )
 
 
+class DeleteVideosBatch(VideoNamesBatch):
+    delete_starred: bool = Field(default=True, description="Whether to allow deletion of starred videos")
+
+
 def _get_video_cache_control() -> str:
     if ApiDependencies.invoker.services.configuration.multiuser:
         return "private, no-store"
@@ -350,36 +354,37 @@ async def upload_video(
 async def delete_video(
     current_user: CurrentUserOrDefault,
     video_name: str = PathParam(description="The name of the video to delete"),
+    delete_starred: bool = Query(default=True, description="Whether to allow deletion of starred videos"),
 ) -> DeleteVideosResult:
     _assert_video_owner(video_name, current_user)
 
-    # Let service-level failures surface as 500s rather than swallowing them and returning a
-    # success-shaped response. A previous version of this handler caught everything and
-    # returned an empty ``deleted_videos`` list with HTTP 200; the frontend treated that as
-    # success, dropped the item from its cache, and the video stayed on disk — a silent
-    # data-consistency failure that only became visible on the next page reload.
-    try:
-        video_dto = ApiDependencies.invoker.services.videos.get_dto(video_name)
-    except Exception:
-        raise HTTPException(status_code=404, detail="Video not found")
-
-    board_id = video_dto.board_id or "none"
+    # Report service and relation-read failures as explicit failed items. The frontend
+    # must only evict names confirmed in ``deleted_videos``; otherwise a partial failure
+    # can silently hide a video that is still present on disk.
+    board_id: str | None = None
     try:
-        ApiDependencies.invoker.services.videos.delete(video_name)
-    except Exception:
-        raise HTTPException(status_code=500, detail="Failed to delete video")
+        board_id = ApiDependencies.invoker.services.board_video_records.get_board_for_video(video_name) or "none"
+        was_deleted = ApiDependencies.invoker.services.videos.delete(video_name, delete_starred=delete_starred)
+    except Exception as error:
+        ApiDependencies.invoker.services.logger.error(f"Failed to delete video {video_name}: {error}")
+        return DeleteVideosResult(
+            deleted_videos=[],
+            failed_videos=[video_name],
+            affected_boards=[board_id] if board_id is not None else [],
+        )
 
     return DeleteVideosResult(
-        deleted_videos=[video_name],
+        deleted_videos=[video_name] if was_deleted else [],
         failed_videos=[],
         affected_boards=[board_id],
+        starred_skipped=[] if was_deleted else [video_name],
     )
 
 
 @videos_router.post("/delete", operation_id="delete_videos_from_list", response_model=DeleteVideosResult)
 def delete_videos_from_list(
     current_user: CurrentUserOrDefault,
-    batch: VideoNamesBatch,
+    batch: DeleteVideosBatch,
 ) -> DeleteVideosResult:
     # Skip — but do not re-raise — auth failures so a foreign name mid-batch doesn't
     # discard the response payload for items the caller had already legitimately deleted.
@@ -393,31 +398,39 @@ def delete_videos_from_list(
     deleted_videos: set[str] = set()
     failed_videos: set[str] = set()
     affected_boards: set[str] = set()
+    starred_skipped: set[str] = set()
     # Dedup while preserving order: a name repeated in the request would otherwise be
     # processed twice, and the second pass's not-found error would land the same name
     # in both deleted_videos and failed_videos.
     for video_name in dict.fromkeys(batch.video_names):
         try:
             _assert_video_owner(video_name, current_user)
-            video_dto = ApiDependencies.invoker.services.videos.get_dto(video_name)
-            board_id = video_dto.board_id or "none"
-            ApiDependencies.invoker.services.videos.delete(video_name)
-            deleted_videos.add(video_name)
+            board_id = ApiDependencies.invoker.services.board_video_records.get_board_for_video(video_name) or "none"
+            was_deleted = ApiDependencies.invoker.services.videos.delete(
+                video_name, delete_starred=batch.delete_starred
+            )
+            if was_deleted:
+                deleted_videos.add(video_name)
+            else:
+                starred_skipped.add(video_name)
             affected_boards.add(board_id)
         except HTTPException:
             continue
-        except Exception:
+        except Exception as error:
             failed_videos.add(video_name)
+            ApiDependencies.invoker.services.logger.error(f"Failed to delete video {video_name}: {error}")
     return DeleteVideosResult(
         deleted_videos=list(deleted_videos),
         failed_videos=list(failed_videos),
         affected_boards=list(affected_boards),
+        starred_skipped=list(starred_skipped),
     )
 
 
 @videos_router.delete("/uncategorized", operation_id="delete_uncategorized_videos", response_model=DeleteVideosResult)
 def delete_uncategorized_videos(
     current_user: CurrentUserOrDefault,
+    delete_starred: bool = Query(default=True, description="Whether to allow deletion of starred videos"),
 ) -> DeleteVideosResult:
     """Deletes all uncategorized videos owned by the current user (or all if admin).
 
@@ -432,22 +445,28 @@ def delete_uncategorized_videos(
     deleted_videos: set[str] = set()
     failed_videos: set[str] = set()
     affected_boards: set[str] = set()
+    starred_skipped: set[str] = set()
     for video_name in names_result.video_names:
         try:
             _assert_video_owner(video_name, current_user)
-            ApiDependencies.invoker.services.videos.delete(video_name)
-            deleted_videos.add(video_name)
+            was_deleted = ApiDependencies.invoker.services.videos.delete(video_name, delete_starred=delete_starred)
+            if was_deleted:
+                deleted_videos.add(video_name)
+            else:
+                starred_skipped.add(video_name)
             affected_boards.add("none")
         except HTTPException:
             # Skip videos not owned by the current user — an intentional skip, not a
             # failed deletion, so it must not be reported (and toasted) as one.
             continue
-        except Exception:
+        except Exception as error:
             failed_videos.add(video_name)
+            ApiDependencies.invoker.services.logger.error(f"Failed to delete video {video_name}: {error}")
     return DeleteVideosResult(
         deleted_videos=list(deleted_videos),
         failed_videos=list(failed_videos),
         affected_boards=list(affected_boards),
+        starred_skipped=list(starred_skipped),
     )
 
 

invokeai/app/services/images/images_base.py

@@ -133,8 +133,8 @@ def get_many(
         pass
 
     @abstractmethod
-    def delete(self, image_name: str):
-        """Deletes an image."""
+    def delete(self, image_name: str, delete_starred: bool = True) -> bool:
+        """Deletes an image, returning ``False`` when a starred image is protected."""
         pass
 
     @abstractmethod
@@ -148,8 +148,10 @@ def get_intermediates_count(self, user_id: Optional[str] = None) -> int:
         pass
 
     @abstractmethod
-    def delete_images_on_board(self, board_id: str, user_id: Optional[str] = None) -> tuple[list[str], list[str]]:
-        """Deletes all images on a board; returns ``(deleted_names, failed_names)``.
+    def delete_images_on_board(
+        self, board_id: str, user_id: Optional[str] = None, delete_starred: bool = True
+    ) -> tuple[list[str], list[str], list[str]]:
+        """Deletes images on a board; returns ``(deleted_names, failed_names, starred_skipped_names)``.
 
         When ``user_id`` is provided, only images owned by that user are deleted (other users'
         contributions to a public/shared board are preserved). Pass ``None`` for the admin
@@ -158,6 +160,9 @@ def delete_images_on_board(self, board_id: str, user_id: Optional[str] = None) -
         ``failed_names`` is the service's ground truth for per-image failures — callers must
         not reconstruct it by diffing their own board listing against ``deleted_names``, which
         races with concurrent moves/deletes.
+
+        When ``delete_starred`` is ``False``, starred images are preserved and reported in
+        ``starred_skipped_names``.
         """
         pass
 

invokeai/app/services/images/images_common.py

@@ -51,6 +51,10 @@ class DeleteImagesResult(ResultWithAffectedBoards):
         default_factory=list,
         description="The names of authorized images that could not be deleted",
     )
+    starred_skipped: list[str] = Field(
+        default_factory=list,
+        description="The names of starred images that were skipped because deletion protection was enabled",
+    )
 
 
 class StarredImagesResult(ResultWithAffectedBoards):