Feature: Add the ability to call a saved workflow from another

invokeai/app/api/routers/session_queue.py

@@ -72,6 +72,11 @@ def sanitize_queue_item_for_user(
     sanitized_item.priority = 0
     sanitized_item.field_values = None
     sanitized_item.retried_from_item_id = None
+    sanitized_item.workflow_call_id = None
+    sanitized_item.parent_item_id = None
+    sanitized_item.parent_session_id = None
+    sanitized_item.root_item_id = None
+    sanitized_item.workflow_call_depth = None
     sanitized_item.workflow = None
     sanitized_item.error_type = None
     sanitized_item.error_message = None
@@ -312,20 +317,27 @@ async def retry_items_by_id(
 ) -> RetryItemsResult:
     """Retries the given queue items. Users can only retry their own items unless they are an admin."""
     try:
-        # Check authorization: user must own all items or be an admin
-        if not current_user.is_admin:
-            for item_id in item_ids:
-                try:
-                    queue_item = ApiDependencies.invoker.services.session_queue.get_queue_item(item_id)
-                    if queue_item.user_id != current_user.user_id:
-                        raise HTTPException(
-                            status_code=403, detail=f"You do not have permission to retry queue item {item_id}"
-                        )
-                except SessionQueueItemNotFoundError:
-                    # Skip items that don't exist - they will be handled by retry_items_by_id
-                    continue
+        # Check queue membership for all items and ownership for non-admins.
+        valid_item_ids: list[int] = []
+        for item_id in item_ids:
+            try:
+                queue_item = ApiDependencies.invoker.services.session_queue.get_queue_item(item_id)
+                if queue_item.queue_id != queue_id:
+                    raise HTTPException(
+                        status_code=404, detail=f"Queue item with id {item_id} not found in queue {queue_id}"
+                    )
+                if not current_user.is_admin and queue_item.user_id != current_user.user_id:
+                    raise HTTPException(
+                        status_code=403, detail=f"You do not have permission to retry queue item {item_id}"
+                    )
+                valid_item_ids.append(item_id)
+            except SessionQueueItemNotFoundError:
+                # Skip items that don't exist - they will be handled by retry_items_by_id
+                continue
 
-        return ApiDependencies.invoker.services.session_queue.retry_items_by_id(queue_id=queue_id, item_ids=item_ids)
+        return ApiDependencies.invoker.services.session_queue.retry_items_by_id(
+            queue_id=queue_id, item_ids=valid_item_ids
+        )
     except HTTPException:
         raise
     except Exception as e:
@@ -507,6 +519,8 @@ async def delete_queue_item(
     try:
         # Get the queue item to check ownership
         queue_item = ApiDependencies.invoker.services.session_queue.get_queue_item(item_id)
+        if queue_item.queue_id != queue_id:
+            raise HTTPException(status_code=404, detail=f"Queue item with id {item_id} not found in queue {queue_id}")
 
         # Check authorization: user must own the item or be an admin
         if queue_item.user_id != current_user.user_id and not current_user.is_admin:
@@ -537,6 +551,8 @@ async def cancel_queue_item(
     try:
         # Get the queue item to check ownership
         queue_item = ApiDependencies.invoker.services.session_queue.get_queue_item(item_id)
+        if queue_item.queue_id != queue_id:
+            raise HTTPException(status_code=404, detail=f"Queue item with id {item_id} not found in queue {queue_id}")
 
         # Check authorization: user must own the item or be an admin
         if queue_item.user_id != current_user.user_id and not current_user.is_admin:

invokeai/app/api/routers/workflows.py

@@ -10,6 +10,7 @@
 from invokeai.app.api.dependencies import ApiDependencies
 from invokeai.app.services.shared.pagination import PaginatedResults
 from invokeai.app.services.shared.sqlite.sqlite_common import SQLiteDirection
+from invokeai.app.services.shared.workflow_call_compatibility import get_workflow_call_compatibility
 from invokeai.app.services.workflow_records.workflow_records_common import (
     Workflow,
     WorkflowCategory,
@@ -51,7 +52,18 @@ async def get_workflow(
             raise HTTPException(status_code=403, detail="Not authorized to access this workflow")
 
     thumbnail_url = ApiDependencies.invoker.services.workflow_thumbnails.get_url(workflow_id)
-    return WorkflowRecordWithThumbnailDTO(thumbnail_url=thumbnail_url, **workflow.model_dump())
+    compatibility = get_workflow_call_compatibility(
+        workflow=workflow.workflow.model_dump(),
+        workflow_id=workflow.workflow_id,
+        services=ApiDependencies.invoker.services,
+        user_id=current_user.user_id,
+        maximum_children=ApiDependencies.invoker.services.configuration.max_queue_size,
+    )
+    return WorkflowRecordWithThumbnailDTO(
+        thumbnail_url=thumbnail_url,
+        call_saved_workflow_compatibility=compatibility,
+        **workflow.model_dump(),
+    )
 
 
 @workflows_router.patch(
@@ -66,17 +78,24 @@ 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")
-    # Pass user_id for defense-in-depth SQL scoping; admins pass None to allow any.
     user_id = None if current_user.is_admin else current_user.user_id
-    return ApiDependencies.invoker.services.workflow_records.update(workflow=workflow, user_id=user_id)
+    updated = ApiDependencies.invoker.services.workflow_records.update(workflow=workflow, user_id=user_id)
+    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(
@@ -88,12 +107,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:
@@ -103,6 +123,11 @@ async def delete_workflow(
         pass
     user_id = None if current_user.is_admin else current_user.user_id
     ApiDependencies.invoker.services.workflow_records.delete(workflow_id, user_id=user_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(
@@ -121,9 +146,15 @@ async def create_workflow(
     # workflows remain visible. In multiuser mode, workflows are private to the creator by default.
     config = ApiDependencies.invoker.services.configuration
     is_public = not config.multiuser
-    return ApiDependencies.invoker.services.workflow_records.create(
+    created = ApiDependencies.invoker.services.workflow_records.create(
         workflow=workflow, user_id=current_user.user_id, is_public=is_public
     )
+    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(
@@ -171,16 +202,31 @@ async def list_workflows(
         user_id=user_id_filter,
         is_public=is_public,
     )
+    skipped_missing_workflows = 0
     for workflow in workflows.items:
+        try:
+            full_workflow = ApiDependencies.invoker.services.workflow_records.get(workflow.workflow_id)
+        except WorkflowNotFoundError:
+            skipped_missing_workflows += 1
+            continue
+        compatibility = get_workflow_call_compatibility(
+            workflow=full_workflow.workflow.model_dump(),
+            workflow_id=full_workflow.workflow_id,
+            services=ApiDependencies.invoker.services,
+            user_id=current_user.user_id,
+            maximum_children=ApiDependencies.invoker.services.configuration.max_queue_size,
+            resolve_generator_items=False,
+        )
         workflows_with_thumbnails.append(
             WorkflowRecordListItemWithThumbnailDTO(
                 thumbnail_url=ApiDependencies.invoker.services.workflow_thumbnails.get_url(workflow.workflow_id),
+                call_saved_workflow_compatibility=compatibility,
                 **workflow.model_dump(),
             )
         )
     return PaginatedResults[WorkflowRecordListItemWithThumbnailDTO](
         items=workflows_with_thumbnails,
-        total=workflows.total,
+        total=max(len(workflows_with_thumbnails), workflows.total - skipped_missing_workflows),
         page=workflows.page,
         pages=workflows.pages,
         per_page=workflows.per_page,
@@ -312,9 +358,16 @@ async def update_workflow_is_public(
         raise HTTPException(status_code=403, detail="Not authorized to update this workflow")
 
     user_id = None if current_user.is_admin else current_user.user_id
-    return ApiDependencies.invoker.services.workflow_records.update_is_public(
+    updated = ApiDependencies.invoker.services.workflow_records.update_is_public(
         workflow_id=workflow_id, is_public=is_public, user_id=user_id
     )
+    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")

invokeai/app/api/sockets.py

@@ -35,8 +35,13 @@
     ModelLoadStartedEvent,
     QueueClearedEvent,
     QueueEventBase,
+    QueueItemsRetriedEvent,
     QueueItemStatusChangedEvent,
     RecallParametersUpdatedEvent,
+    WorkflowCreatedEvent,
+    WorkflowDeletedEvent,
+    WorkflowEventBase,
+    WorkflowUpdatedEvent,
     register_events,
 )
 from invokeai.backend.util.logging import InvokeAILogger
@@ -65,6 +70,7 @@ class BulkDownloadSubscriptionEvent(BaseModel):
     InvocationErrorEvent,
     QueueItemStatusChangedEvent,
     BatchEnqueuedEvent,
+    QueueItemsRetriedEvent,
     QueueClearedEvent,
     RecallParametersUpdatedEvent,
 }
@@ -86,6 +92,7 @@ class BulkDownloadSubscriptionEvent(BaseModel):
 }
 
 BULK_DOWNLOAD_EVENTS = {BulkDownloadStartedEvent, BulkDownloadCompleteEvent, BulkDownloadErrorEvent}
+WORKFLOW_EVENTS = {WorkflowCreatedEvent, WorkflowUpdatedEvent, WorkflowDeletedEvent}
 
 
 class SocketIO:
@@ -115,6 +122,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.
@@ -167,6 +175,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
 
         # No valid token provided. In multiuser mode this is not allowed — reject
@@ -183,6 +195,9 @@ async def _handle_connect(self, sid: str, environ: dict, auth: dict | None) -> b
             "is_admin": True,
         }
         logger.debug(f"Socket {sid} connected as system admin (single-user mode)")
+        await self._sio.enter_room(sid, "user:system")
+        await self._sio.enter_room(sid, "workflows:shared")
+        await self._sio.enter_room(sid, "admin")
         return True
 
     @staticmethod
@@ -329,6 +344,26 @@ async def _handle_queue_event(self, event: FastAPIEvent[QueueEventBase]):
                 await self._sio.emit(event=event_name, data=event_data.model_dump(mode="json"), room="admin")
                 logger.debug(f"Emitted private batch_enqueued event to user room {user_room} and admin room")
 
+            # QueueItemsRetriedEvent carries queue item ids that should only be visible
+            # to the affected owners + admins.
+            elif isinstance(event_data, QueueItemsRetriedEvent):
+                for user_id in event_data.user_ids:
+                    user_room = f"user:{user_id}"
+                    owner_event_data = event_data.model_copy(
+                        update={
+                            "retried_item_ids": event_data.retried_item_ids_by_user.get(user_id, []),
+                            "user_ids": [user_id],
+                            "retried_item_ids_by_user": {user_id: event_data.retried_item_ids_by_user.get(user_id, [])},
+                        }
+                    )
+                    await self._sio.emit(
+                        event=event_name, data=owner_event_data.model_dump(mode="json"), room=user_room
+                    )
+                await self._sio.emit(event=event_name, data=event_data.model_dump(mode="json"), room="admin")
+                logger.debug(
+                    f"Emitted private queue_items_retried event to user rooms {event_data.user_ids} and admin room"
+                )
+
             else:
                 # For remaining queue events (e.g. QueueClearedEvent) that do not
                 # carry user identity, emit to all subscribers in the queue room.
@@ -360,3 +395,32 @@ async def _handle_bulk_image_download_event(self, event: FastAPIEvent[BulkDownlo
             await self._sio.emit(
                 event=event_name, data=event_data.model_dump(mode="json"), room=event_data.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")
+
+        if not self._is_multiuser_enabled():
+            await self._sio.emit(event=event_name, data=payload, room="admin")
+            return
+
+        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"
+                )