Add workflow live update events

JPPhoto · JPPhoto · commit c2e7a5dec885 · 2026-04-08T13:38:50.000-05:00
Stacked on top of origin PR invoke-ai#9018 (shared/private workflows and boards) for multiuser workflow visibility semantics.
diff --git a/invokeai/app/api/routers/workflows.py b/invokeai/app/api/routers/workflows.py
@@ -66,15 +66,23 @@ async def update_workflow(
     workflow: Workflow = Body(description="The updated workflow", embed=True),
 ) -> WorkflowRecordDTO:
     """Updates a workflow"""
+    try:
+        existing = ApiDependencies.invoker.services.workflow_records.get(workflow.id)
+    except WorkflowNotFoundError:
+        raise HTTPException(status_code=404, detail="Workflow not found")
+
     config = ApiDependencies.invoker.services.configuration
     if config.multiuser:
-        try:
-            existing = ApiDependencies.invoker.services.workflow_records.get(workflow.id)
-        except WorkflowNotFoundError:
-            raise HTTPException(status_code=404, detail="Workflow not found")
         if not current_user.is_admin and existing.user_id != current_user.user_id:
             raise HTTPException(status_code=403, detail="Not authorized to update this workflow")
-    return ApiDependencies.invoker.services.workflow_records.update(workflow=workflow)
+    updated = ApiDependencies.invoker.services.workflow_records.update(workflow=workflow)
+    ApiDependencies.invoker.services.events.emit_workflow_updated(
+        workflow_id=updated.workflow_id,
+        user_id=updated.user_id,
+        old_is_public=existing.is_public,
+        new_is_public=updated.is_public,
+    )
+    return updated
 
 
 @workflows_router.delete(
@@ -86,12 +94,13 @@ async def delete_workflow(
     workflow_id: str = Path(description="The workflow to delete"),
 ) -> None:
     """Deletes a workflow"""
+    try:
+        existing = ApiDependencies.invoker.services.workflow_records.get(workflow_id)
+    except WorkflowNotFoundError:
+        raise HTTPException(status_code=404, detail="Workflow not found")
+
     config = ApiDependencies.invoker.services.configuration
     if config.multiuser:
-        try:
-            existing = ApiDependencies.invoker.services.workflow_records.get(workflow_id)
-        except WorkflowNotFoundError:
-            raise HTTPException(status_code=404, detail="Workflow not found")
         if not current_user.is_admin and existing.user_id != current_user.user_id:
             raise HTTPException(status_code=403, detail="Not authorized to delete this workflow")
     try:
@@ -100,6 +109,11 @@ async def delete_workflow(
         # It's OK if the workflow has no thumbnail file. We can still delete the workflow.
         pass
     ApiDependencies.invoker.services.workflow_records.delete(workflow_id)
+    ApiDependencies.invoker.services.events.emit_workflow_deleted(
+        workflow_id=existing.workflow_id,
+        user_id=existing.user_id,
+        is_public=existing.is_public,
+    )
 
 
 @workflows_router.post(
@@ -114,7 +128,13 @@ async def create_workflow(
     workflow: WorkflowWithoutID = Body(description="The workflow to create", embed=True),
 ) -> WorkflowRecordDTO:
     """Creates a workflow"""
-    return ApiDependencies.invoker.services.workflow_records.create(workflow=workflow, user_id=current_user.user_id)
+    created = ApiDependencies.invoker.services.workflow_records.create(workflow=workflow, user_id=current_user.user_id)
+    ApiDependencies.invoker.services.events.emit_workflow_created(
+        workflow_id=created.workflow_id,
+        user_id=created.user_id,
+        is_public=created.is_public,
+    )
+    return created
 
 
 @workflows_router.get(
@@ -302,9 +322,14 @@ async def update_workflow_is_public(
     if config.multiuser and not current_user.is_admin and existing.user_id != current_user.user_id:
         raise HTTPException(status_code=403, detail="Not authorized to update this workflow")
 
-    return ApiDependencies.invoker.services.workflow_records.update_is_public(
-        workflow_id=workflow_id, is_public=is_public
+    updated = ApiDependencies.invoker.services.workflow_records.update_is_public(workflow_id=workflow_id, is_public=is_public)
+    ApiDependencies.invoker.services.events.emit_workflow_updated(
+        workflow_id=updated.workflow_id,
+        user_id=updated.user_id,
+        old_is_public=existing.is_public,
+        new_is_public=updated.is_public,
     )
+    return updated
 
 
 @workflows_router.get("/tags", operation_id="get_all_tags")
diff --git a/invokeai/app/api/sockets.py b/invokeai/app/api/sockets.py
@@ -37,6 +37,10 @@
     QueueEventBase,
     QueueItemStatusChangedEvent,
     RecallParametersUpdatedEvent,
+    WorkflowCreatedEvent,
+    WorkflowDeletedEvent,
+    WorkflowEventBase,
+    WorkflowUpdatedEvent,
     register_events,
 )
 from invokeai.backend.util.logging import InvokeAILogger
@@ -86,6 +90,7 @@ class BulkDownloadSubscriptionEvent(BaseModel):
 }
 
 BULK_DOWNLOAD_EVENTS = {BulkDownloadStartedEvent, BulkDownloadCompleteEvent, BulkDownloadErrorEvent}
+WORKFLOW_EVENTS = {WorkflowCreatedEvent, WorkflowUpdatedEvent, WorkflowDeletedEvent}
 
 
 class SocketIO:
@@ -115,6 +120,7 @@ def __init__(self, app: FastAPI):
         register_events(QUEUE_EVENTS, self._handle_queue_event)
         register_events(MODEL_EVENTS, self._handle_model_event)
         register_events(BULK_DOWNLOAD_EVENTS, self._handle_bulk_image_download_event)
+        register_events(WORKFLOW_EVENTS, self._handle_workflow_event)
 
     async def _handle_connect(self, sid: str, environ: dict, auth: dict | None) -> bool:
         """Handle socket connection and authenticate the user.
@@ -145,6 +151,10 @@ async def _handle_connect(self, sid: str, environ: dict, auth: dict | None) -> b
                 logger.info(
                     f"Socket {sid} connected with user_id: {token_data.user_id}, is_admin: {token_data.is_admin}"
                 )
+                await self._sio.enter_room(sid, f"user:{token_data.user_id}")
+                await self._sio.enter_room(sid, "workflows:shared")
+                if token_data.is_admin:
+                    await self._sio.enter_room(sid, "admin")
                 return True
 
         # If no valid token, store system user for backward compatibility
@@ -266,3 +276,28 @@ async def _handle_model_event(self, event: FastAPIEvent[ModelEventBase | Downloa
 
     async def _handle_bulk_image_download_event(self, event: FastAPIEvent[BulkDownloadEventBase]) -> None:
         await self._sio.emit(event=event[0], data=event[1].model_dump(mode="json"), room=event[1].bulk_download_id)
+
+    async def _handle_workflow_event(self, event: FastAPIEvent[WorkflowEventBase]) -> None:
+        event_name, event_data = event
+        payload = event_data.model_dump(mode="json")
+
+        await self._sio.emit(event=event_name, data=payload, room=f"user:{event_data.user_id}")
+        await self._sio.emit(event=event_name, data=payload, room="admin")
+
+        if event_name == "workflow_created":
+            if getattr(event_data, "is_public", False):
+                await self._sio.emit(event=event_name, data=payload, room="workflows:shared")
+            return
+
+        if event_name == "workflow_deleted":
+            if getattr(event_data, "is_public", False):
+                await self._sio.emit(event=event_name, data=payload, room="workflows:shared")
+            return
+
+        if event_name == "workflow_updated":
+            if getattr(event_data, "new_is_public", False):
+                await self._sio.emit(event=event_name, data=payload, room="workflows:shared")
+            elif getattr(event_data, "old_is_public", False):
+                await self._sio.emit(
+                    event="workflow_deleted", data={"workflow_id": event_data.workflow_id}, room="workflows:shared"
+                )
diff --git a/invokeai/app/services/events/events_base.py b/invokeai/app/services/events/events_base.py
@@ -32,6 +32,9 @@
     QueueItemsRetriedEvent,
     QueueItemStatusChangedEvent,
     RecallParametersUpdatedEvent,
+    WorkflowCreatedEvent,
+    WorkflowDeletedEvent,
+    WorkflowUpdatedEvent,
 )
 
 if TYPE_CHECKING:
@@ -118,6 +121,29 @@ def emit_recall_parameters_updated(self, queue_id: str, parameters: dict) -> Non
 
     # endregion
 
+    # region Workflow library
+
+    def emit_workflow_created(self, workflow_id: str, user_id: str, is_public: bool) -> None:
+        """Emitted when a workflow is created."""
+        self.dispatch(WorkflowCreatedEvent.build(workflow_id=workflow_id, user_id=user_id, is_public=is_public))
+
+    def emit_workflow_updated(self, workflow_id: str, user_id: str, old_is_public: bool, new_is_public: bool) -> None:
+        """Emitted when a workflow is updated."""
+        self.dispatch(
+            WorkflowUpdatedEvent.build(
+                workflow_id=workflow_id,
+                user_id=user_id,
+                old_is_public=old_is_public,
+                new_is_public=new_is_public,
+            )
+        )
+
+    def emit_workflow_deleted(self, workflow_id: str, user_id: str, is_public: bool) -> None:
+        """Emitted when a workflow is deleted."""
+        self.dispatch(WorkflowDeletedEvent.build(workflow_id=workflow_id, user_id=user_id, is_public=is_public))
+
+    # endregion
+
     # region Download
 
     def emit_download_started(self, job: "DownloadJob") -> None:
diff --git a/invokeai/app/services/events/events_common.py b/invokeai/app/services/events/events_common.py
@@ -321,6 +321,60 @@ def build(cls, queue_id: str) -> "QueueClearedEvent":
         return cls(queue_id=queue_id)
 
 
+class WorkflowEventBase(EventBase):
+    """Base class for workflow library CRUD events."""
+
+    workflow_id: str = Field(description="The ID of the workflow")
+    user_id: str = Field(description="The owner of the workflow")
+
+
+@payload_schema.register
+class WorkflowCreatedEvent(WorkflowEventBase):
+    """Event model for workflow_created"""
+
+    __event_name__ = "workflow_created"
+
+    is_public: bool = Field(description="Whether the workflow is shared with all users")
+
+    @classmethod
+    def build(cls, workflow_id: str, user_id: str, is_public: bool) -> "WorkflowCreatedEvent":
+        return cls(workflow_id=workflow_id, user_id=user_id, is_public=is_public)
+
+
+@payload_schema.register
+class WorkflowUpdatedEvent(WorkflowEventBase):
+    """Event model for workflow_updated"""
+
+    __event_name__ = "workflow_updated"
+
+    old_is_public: bool = Field(description="Whether the workflow was shared before the update")
+    new_is_public: bool = Field(description="Whether the workflow is shared after the update")
+
+    @classmethod
+    def build(
+        cls, workflow_id: str, user_id: str, old_is_public: bool, new_is_public: bool
+    ) -> "WorkflowUpdatedEvent":
+        return cls(
+            workflow_id=workflow_id,
+            user_id=user_id,
+            old_is_public=old_is_public,
+            new_is_public=new_is_public,
+        )
+
+
+@payload_schema.register
+class WorkflowDeletedEvent(WorkflowEventBase):
+    """Event model for workflow_deleted"""
+
+    __event_name__ = "workflow_deleted"
+
+    is_public: bool = Field(description="Whether the workflow was shared when it was deleted")
+
+    @classmethod
+    def build(cls, workflow_id: str, user_id: str, is_public: bool) -> "WorkflowDeletedEvent":
+        return cls(workflow_id=workflow_id, user_id=user_id, is_public=is_public)
+
+
 class DownloadEventBase(EventBase):
     """Base class for events associated with a download"""
 
diff --git a/invokeai/frontend/web/src/services/events/setEventListeners.test.ts b/invokeai/frontend/web/src/services/events/setEventListeners.test.ts
@@ -0,0 +1,99 @@
+import { LIST_TAG } from 'services/api';
+import { describe, expect, it, vi } from 'vitest';
+
+import { setEventListeners } from './setEventListeners';
+
+vi.mock('app/logging/logger', () => ({
+  logger: () => ({
+    debug: vi.fn(),
+    trace: vi.fn(),
+    info: vi.fn(),
+    warn: vi.fn(),
+    error: vi.fn(),
+  }),
+}));
+
+vi.mock('features/toast/toast', () => ({
+  toast: vi.fn(),
+}));
+
+vi.mock('./onInvocationComplete', () => ({
+  buildOnInvocationComplete: () => vi.fn(),
+}));
+
+vi.mock('./onModelInstallError', () => ({
+  buildOnModelInstallError: () => vi.fn(),
+  DiscordLink: () => null,
+  GitHubIssuesLink: () => null,
+}));
+
+const createMockSocket = () => {
+  const handlers = new Map<string, (...args: Array<unknown>) => void>();
+
+  return {
+    on: vi.fn((event: string, handler: (...args: Array<unknown>) => void) => {
+      handlers.set(event, handler);
+    }),
+    emit: vi.fn(),
+    trigger: (event: string, payload?: unknown) => {
+      const handler = handlers.get(event);
+      if (!handler) {
+        throw new Error(`No handler registered for ${event}`);
+      }
+      handler(payload);
+    },
+  };
+};
+
+describe('setEventListeners workflow live updates', () => {
+  it('invalidates workflow list caches on workflow_created', () => {
+    const socket = createMockSocket();
+    const dispatch = vi.fn();
+    const store = {
+      dispatch,
+      getState: vi.fn(() => ({})),
+    };
+
+    setEventListeners({
+      socket: socket as never,
+      store: store as never,
+      setIsConnected: vi.fn(),
+    });
+
+    socket.trigger('workflow_created', { workflow_id: 'wf-1', is_public: true });
+
+    expect(dispatch).toHaveBeenCalledWith(
+      expect.objectContaining({
+        payload: expect.arrayContaining([
+          { type: 'Workflow', id: LIST_TAG },
+          'WorkflowTags',
+          'WorkflowTagCounts',
+          'WorkflowCategoryCounts',
+        ]),
+      })
+    );
+  });
+
+  it('ignores unrelated events for workflow cache invalidation', () => {
+    const socket = createMockSocket();
+    const dispatch = vi.fn();
+    const store = {
+      dispatch,
+      getState: vi.fn(() => ({})),
+    };
+
+    setEventListeners({
+      socket: socket as never,
+      store: store as never,
+      setIsConnected: vi.fn(),
+    });
+
+    socket.trigger('download_started', { source: 'x', download_path: '/tmp/x' });
+
+    expect(dispatch).not.toHaveBeenCalledWith(
+      expect.objectContaining({
+        payload: expect.arrayContaining([{ type: 'Workflow', id: LIST_TAG }]),
+      })
+    );
+  });
+});
diff --git a/invokeai/frontend/web/src/services/events/setEventListeners.tsx b/invokeai/frontend/web/src/services/events/setEventListeners.tsx
@@ -98,6 +98,32 @@ export const setEventListeners = ({ socket, store, setIsConnected }: SetEventLis
     setIsConnected(false);
   });
 
+  const invalidateWorkflowLibrary = () => {
+    dispatch(
+      api.util.invalidateTags([
+        { type: 'Workflow', id: LIST_TAG },
+        'WorkflowTags',
+        'WorkflowTagCounts',
+        'WorkflowCategoryCounts',
+      ])
+    );
+  };
+
+  socket.on('workflow_created', (data) => {
+    log.debug({ data }, 'Workflow created');
+    invalidateWorkflowLibrary();
+  });
+
+  socket.on('workflow_updated', (data) => {
+    log.debug({ data }, 'Workflow updated');
+    invalidateWorkflowLibrary();
+  });
+
+  socket.on('workflow_deleted', (data) => {
+    log.debug({ data }, 'Workflow deleted');
+    invalidateWorkflowLibrary();
+  });
+
   socket.on('invocation_started', (data) => {
     if (finishedQueueItemIds.has(data.item_id)) {
       return;
diff --git a/invokeai/frontend/web/src/services/events/types.ts b/invokeai/frontend/web/src/services/events/types.ts
diff --git a/tests/app/routers/test_workflow_live_updates.py b/tests/app/routers/test_workflow_live_updates.py
diff --git a/tests/app/test_workflow_socketio.py b/tests/app/test_workflow_socketio.py
