Fix session processor crash when a queue item is deleted while running

invokeai/app/api/sockets.py

@@ -306,7 +306,10 @@ async def _handle_queue_event(self, event: FastAPIEvent[QueueEventBase]):
            per-session side effects.
 
         InvocationEventBase events stay private (owner + admins only). RecallParametersUpdatedEvent
-        is also private. QueueClearedEvent has no user identity and is broadcast to the queue room.
+        is also private. QueueClearedEvent is broadcast to the queue room when unscoped (an admin or
+        single-user clear that deleted every user's items); a user-scoped clear goes full to
+        owner + admins with a sanitized companion to the rest of the queue room, so other users
+        refresh their queue lists without treating the clear as their own.
 
         IMPORTANT: Check InvocationEventBase BEFORE QueueItemEventBase since InvocationEventBase
         inherits from QueueItemEventBase. The order of isinstance checks matters!
@@ -445,9 +448,37 @@ async def _handle_queue_event(self, event: FastAPIEvent[QueueEventBase]):
                     f"Emitted private queue_items_retried event to user rooms {event_data.user_ids} and admin room"
                 )
 
+            # QueueClearedEvent: an unscoped clear (user_id=None — admin or single-user mode)
+            # deleted every user's items, so everyone gets the full event. A user-scoped clear
+            # only deleted that user's rows: full event to owner+admin (single emit to a room
+            # list so a socket in both rooms receives it once), sanitized companion to the rest
+            # of the queue room so their queue lists and badge counts refetch without treating
+            # the clear as their own.
+            elif isinstance(event_data, QueueClearedEvent):
+                if event_data.user_id is None:
+                    await self._sio.emit(
+                        event=event_name, data=event_data.model_dump(mode="json"), room=event_data.queue_id
+                    )
+                    logger.debug(f"Emitted unscoped queue_cleared to all subscribers in queue {event_data.queue_id}")
+                else:
+                    user_room = f"user:{event_data.user_id}"
+                    await self._sio.emit(
+                        event=event_name, data=event_data.model_dump(mode="json"), room=[user_room, "admin"]
+                    )
+                    sanitized = event_data.model_copy(update={"user_id": "redacted"})
+                    await self._sio.emit(
+                        event=event_name,
+                        data=sanitized.model_dump(mode="json"),
+                        room=event_data.queue_id,
+                        skip_sid=self._owner_and_admin_sids(event_data.user_id),
+                    )
+                    logger.debug(
+                        f"Emitted queue_cleared: full to {user_room}+admin, sanitized to queue {event_data.queue_id}"
+                    )
+
             else:
-                # For remaining queue events (e.g. QueueClearedEvent) that do not
-                # carry user identity, emit to all subscribers in the queue room.
+                # For remaining queue events that do not carry user identity,
+                # emit to all subscribers in the queue room.
                 await self._sio.emit(
                     event=event_name, data=event_data.model_dump(mode="json"), room=event_data.queue_id
                 )

invokeai/app/services/events/events_base.py

@@ -113,9 +113,9 @@ def emit_queue_items_retried(
         """Emitted when a list of queue items are retried"""
         self.dispatch(QueueItemsRetriedEvent.build(retry_result, user_ids, retried_item_ids_by_user))
 
-    def emit_queue_cleared(self, queue_id: str) -> None:
-        """Emitted when a queue is cleared"""
-        self.dispatch(QueueClearedEvent.build(queue_id))
+    def emit_queue_cleared(self, queue_id: str, user_id: str | None = None) -> None:
+        """Emitted when a queue is cleared. `user_id` scopes the clear to one user's items; None means all."""
+        self.dispatch(QueueClearedEvent.build(queue_id, user_id))
 
     def emit_recall_parameters_updated(self, queue_id: str, user_id: str, parameters: dict) -> None:
         """Emitted when recall parameters are updated"""

invokeai/app/services/events/events_common.py

@@ -331,9 +331,14 @@ class QueueClearedEvent(QueueEventBase):
 
     __event_name__ = "queue_cleared"
 
+    user_id: str | None = Field(
+        default=None,
+        description="The ID of the user whose queue items were cleared, or None if all users' items were cleared",
+    )
+
     @classmethod
-    def build(cls, queue_id: str) -> "QueueClearedEvent":
-        return cls(queue_id=queue_id)
+    def build(cls, queue_id: str, user_id: str | None = None) -> "QueueClearedEvent":
+        return cls(queue_id=queue_id, user_id=user_id)
 
 
 class WorkflowEventBase(EventBase):

invokeai/app/services/session_processor/workflow_call_runtime.py

@@ -8,7 +8,11 @@
 )
 from invokeai.app.invocations.workflow_return import WorkflowReturnOutput
 from invokeai.app.services.session_processor.workflow_call_batch import build_child_workflow_session_results
-from invokeai.app.services.session_queue.session_queue_common import NodeFieldValue, SessionQueueItem
+from invokeai.app.services.session_queue.session_queue_common import (
+    NodeFieldValue,
+    SessionQueueItem,
+    SessionQueueItemNotFoundError,
+)
 from invokeai.app.services.shared.graph import GraphExecutionState
 
 if TYPE_CHECKING:
@@ -175,15 +179,19 @@ def fail_waiting_workflow_call(self, queue_item: SessionQueueItem, error_message
             error_traceback=error_message,
         )
 
-    def _get_parent_queue_item(self, child_queue_item: SessionQueueItem) -> SessionQueueItem:
+    def _get_parent_queue_item(self, child_queue_item: SessionQueueItem) -> SessionQueueItem | None:
         parent_item_id = child_queue_item.parent_item_id
         if parent_item_id is None:
             raise ValueError("Child workflow queue item is missing parent_item_id metadata.")
-        return self._session_runner._services.session_queue.get_queue_item(parent_item_id)
+        try:
+            return self._session_runner._services.session_queue.get_queue_item(parent_item_id)
+        except SessionQueueItemNotFoundError:
+            # The parent may have been deleted while the child was running (e.g. the queue was cleared).
+            return None
 
     def _resume_parent_from_completed_child(self, child_queue_item: SessionQueueItem) -> None:
         parent_queue_item = self._get_parent_queue_item(child_queue_item)
-        if parent_queue_item.status in ("completed", "failed", "canceled"):
+        if parent_queue_item is None or parent_queue_item.status in ("completed", "failed", "canceled"):
             return
         try:
             output = self.get_child_workflow_return_output(child_queue_item.session)
@@ -200,7 +208,12 @@ def _resume_parent_from_completed_child(self, child_queue_item: SessionQueueItem
                     exclude_item_ids={child_queue_item.item_id},
                 )
             self.fail_waiting_workflow_call(parent_queue_item, str(e))
-            parent_queue_item = self._session_runner._services.session_queue.get_queue_item(parent_queue_item.item_id)
+            try:
+                parent_queue_item = self._session_runner._services.session_queue.get_queue_item(
+                    parent_queue_item.item_id
+                )
+            except SessionQueueItemNotFoundError:
+                return
             if getattr(parent_queue_item, "parent_item_id", None) is not None:
                 self._fail_parent_from_failed_child(parent_queue_item)
             return
@@ -229,7 +242,7 @@ def _resume_parent_from_completed_child(self, child_queue_item: SessionQueueItem
 
     def _fail_parent_from_failed_child(self, child_queue_item: SessionQueueItem) -> None:
         parent_queue_item = self._get_parent_queue_item(child_queue_item)
-        if parent_queue_item.status in ("completed", "failed", "canceled"):
+        if parent_queue_item is None or parent_queue_item.status in ("completed", "failed", "canceled"):
             return
         waiting_frame = parent_queue_item.session.waiting_workflow_call
         if waiting_frame is None:
@@ -244,24 +257,38 @@ def _fail_parent_from_failed_child(self, child_queue_item: SessionQueueItem) ->
             f"The selected saved workflow '{waiting_frame.workflow_id}' failed during child execution."
         )
         self.fail_waiting_workflow_call(parent_queue_item, child_error_message)
-        parent_queue_item = self._session_runner._services.session_queue.get_queue_item(parent_queue_item.item_id)
+        try:
+            parent_queue_item = self._session_runner._services.session_queue.get_queue_item(parent_queue_item.item_id)
+        except SessionQueueItemNotFoundError:
+            return
         if getattr(parent_queue_item, "parent_item_id", None) is not None:
             self._fail_parent_from_failed_child(parent_queue_item)
 
     def _cancel_parent_from_canceled_child(self, child_queue_item: SessionQueueItem) -> None:
         parent_queue_item = self._get_parent_queue_item(child_queue_item)
-        if parent_queue_item.status == "canceled":
+        if parent_queue_item is None or parent_queue_item.status == "canceled":
             return
         self._session_runner._services.session_queue.cancel_queue_item(parent_queue_item.item_id)
 
     def run_queue_item(self, queue_item: SessionQueueItem) -> None:
         self._session_runner.run(queue_item)
-        updated_queue_item = self._session_runner._services.session_queue.get_queue_item(queue_item.item_id)
+        try:
+            updated_queue_item = self._session_runner._services.session_queue.get_queue_item(queue_item.item_id)
+        except SessionQueueItemNotFoundError:
+            # The queue item was deleted while it was running (e.g. the queue was cleared or the current item
+            # was deleted). There is no parent lifecycle work to do for a deleted item.
+            return
         if getattr(updated_queue_item, "parent_item_id", None) is None:
             return
-        if updated_queue_item.status == "completed":
-            self._resume_parent_from_completed_child(updated_queue_item)
-        elif updated_queue_item.status == "failed":
-            self._fail_parent_from_failed_child(updated_queue_item)
-        elif updated_queue_item.status == "canceled":
-            self._cancel_parent_from_canceled_child(updated_queue_item)
+        try:
+            if updated_queue_item.status == "completed":
+                self._resume_parent_from_completed_child(updated_queue_item)
+            elif updated_queue_item.status == "failed":
+                self._fail_parent_from_failed_child(updated_queue_item)
+            elif updated_queue_item.status == "canceled":
+                self._cancel_parent_from_canceled_child(updated_queue_item)
+        except SessionQueueItemNotFoundError:
+            # An item in the parent chain was deleted after we looked it up but before we mutated it
+            # (the queue mutations re-read the row and raise when it has disappeared). The chain is
+            # being torn down; there is nothing left to update.
+            return

invokeai/app/services/session_queue/session_queue_sqlite.py

@@ -524,7 +524,7 @@ def clear(self, queue_id: str, user_id: Optional[str] = None) -> ClearResult:
                 """,
                 tuple(params),
             )
-        self.__invoker.services.events.emit_queue_cleared(queue_id)
+        self.__invoker.services.events.emit_queue_cleared(queue_id, user_id)
         return ClearResult(deleted=count)
 
     def delete_queue_items_by_id(self, item_ids: list[int]) -> None:

invokeai/frontend/web/openapi.json

@@ -59604,9 +59604,22 @@
             "description": "The ID of the queue",
             "title": "Queue Id",
             "type": "string"
+          },
+          "user_id": {
+            "anyOf": [
+              {
+                "type": "string"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "default": null,
+            "description": "The ID of the user whose queue items were cleared, or None if all users' items were cleared",
+            "title": "User Id"
           }
         },
-        "required": ["timestamp", "queue_id"],
+        "required": ["timestamp", "queue_id", "user_id"],
         "title": "QueueClearedEvent",
         "type": "object"
       },

invokeai/frontend/web/src/common/hooks/useGlobalHotkeys.ts

@@ -1,8 +1,8 @@
 import { useAppStore } from 'app/store/storeHooks';
 import { useDeleteImageModalApi } from 'features/deleteImageModal/store/state';
 import { selectSelection } from 'features/gallery/store/gallerySelectors';
+import { useCancelCurrentQueueItem } from 'features/queue/hooks/useCancelCurrentQueueItem';
 import { useClearQueue } from 'features/queue/hooks/useClearQueue';
-import { useDeleteCurrentQueueItem } from 'features/queue/hooks/useDeleteCurrentQueueItem';
 import { useInvoke } from 'features/queue/hooks/useInvoke';
 import { useRegisteredHotkeys } from 'features/system/components/HotkeysModal/useHotkeyData';
 import { navigationApi } from 'features/ui/layouts/navigation-api';
@@ -37,17 +37,19 @@ export const useGlobalHotkeys = () => {
     dependencies: [queue],
   });
 
-  const deleteCurrentQueueItem = useDeleteCurrentQueueItem();
+  const cancelCurrentQueueItem = useCancelCurrentQueueItem();
 
   useRegisteredHotkeys({
     id: 'cancelQueueItem',
     category: 'app',
-    callback: deleteCurrentQueueItem.trigger,
+    callback: () => {
+      cancelCurrentQueueItem.trigger();
+    },
     options: {
-      enabled: !deleteCurrentQueueItem.isDisabled && !deleteCurrentQueueItem.isLoading,
+      enabled: !cancelCurrentQueueItem.isDisabled && !cancelCurrentQueueItem.isLoading,
       preventDefault: true,
     },
-    dependencies: [deleteCurrentQueueItem],
+    dependencies: [cancelCurrentQueueItem],
   });
 
   const clearQueue = useClearQueue();

invokeai/frontend/web/src/services/api/schema.ts

@@ -25206,6 +25206,12 @@ export type components = {
              * @description The ID of the queue
              */
             queue_id: string;
+            /**
+             * User Id
+             * @description The ID of the user whose queue items were cleared, or None if all users' items were cleared
+             * @default null
+             */
+            user_id: string | null;
         };
         /**
          * QueueItemStatusChangedEvent

invokeai/frontend/web/src/services/events/setEventListeners.tsx

@@ -4,6 +4,7 @@ import { socketConnected } from 'app/store/middleware/listenerMiddleware/listene
 import type { AppStore } from 'app/store/store';
 import { parseify } from 'common/util/serialize';
 import { isNil, round } from 'es-toolkit/compat';
+import { selectCurrentUser } from 'features/auth/store/authSlice';
 import { getDefaultRefImageConfig } from 'features/controlLayers/hooks/addLayerHooks';
 import { allEntitiesDeleted, controlLayerRecalled } from 'features/controlLayers/store/canvasSlice';
 import { canvasWorkflowIntegrationProcessingCompleted } from 'features/controlLayers/store/canvasWorkflowIntegrationSlice';
@@ -71,6 +72,7 @@ export const setEventListeners = ({ socket, store, setIsConnected }: SetEventLis
     clearCanvasWorkflowIntegrationProcessing: () => dispatch(canvasWorkflowIntegrationProcessingCompleted()),
     completedInvocationKeysByItemId,
     getAllNodeExecutionStates: () => $nodeExecutionStates.get(),
+    getCurrentUserId: () => selectCurrentUser(getState())?.user_id ?? null,
     getNodeExecutionState: (nodeId) => $nodeExecutionStates.get()[nodeId],
     logReconciliationError: (error, itemId) => {
       log.debug({ error: parseify(error) }, `Unable to reconcile workflow queue item ${itemId}`);
@@ -538,6 +540,15 @@ export const setEventListeners = ({ socket, store, setIsConnected }: SetEventLis
 
   socket.on('queue_cleared', (data) => {
     log.debug({ data }, 'Queue cleared');
+    // Clearing the queue deletes the in-progress item without emitting a per-item terminal status
+    // event, so the progress bar must be reset here — and the coordinator must mark its tracked
+    // items terminal so a trailing invocation_progress event cannot repopulate the bar. The
+    // coordinator reports whether the clear applied to this client's items: a user-scoped clear
+    // by another user (multiuser mode) leaves them untouched, and only the queue tags below need
+    // refreshing.
+    if (workflowExecutionCoordinator.onQueueCleared(data)) {
+      $lastProgressEvent.set(null);
+    }
     dispatch(
       queueApi.util.invalidateTags([
         'SessionQueueStatus',