fix(multiuser): do not let progress status of one user's session leak into another's

invokeai/app/api/routers/model_manager.py

@@ -1155,7 +1155,7 @@ async def convert_model(
 
     with TemporaryDirectory(dir=ApiDependencies.invoker.services.configuration.models_path) as tmpdir:
         convert_path = pathlib.Path(tmpdir) / pathlib.Path(model_config.path).stem
-        converted_model = loader.load_model(model_config)
+        converted_model = loader.load_model(model_config, user_id=current_admin.user_id)
         # write the converted file to the convert path
         raw_model = converted_model.model
         assert hasattr(raw_model, "save_pretrained")

invokeai/app/api/routers/session_queue.py

@@ -468,12 +468,14 @@ async def get_queue_status(
     current_user: CurrentUserOrDefault,
     queue_id: str = Path(description="The queue id to perform this operation on"),
 ) -> SessionQueueAndProcessorStatus:
-    """Gets the status of the session queue. Returns global counts; non-admin users additionally
-    get their own pending/in_progress counts (so the UI can show an X/Y badge) and cannot see the
-    current item's identifiers unless they own it."""
+    """Gets the status of the session queue. Returns global counts; every user additionally gets
+    their own pending/in_progress counts (so the UI can show an X/Y badge and scope personal UI
+    like the progress bar to the user's own activity). Non-admin users cannot see the current
+    item's identifiers unless they own it."""
     try:
-        user_id = None if current_user.is_admin else current_user.user_id
-        queue = ApiDependencies.invoker.services.session_queue.get_queue_status(queue_id, user_id=user_id)
+        queue = ApiDependencies.invoker.services.session_queue.get_queue_status(
+            queue_id, user_id=current_user.user_id, is_admin=current_user.is_admin
+        )
         processor = ApiDependencies.invoker.services.session_processor.get_status()
         return SessionQueueAndProcessorStatus(queue=queue, processor=processor)
     except Exception as e:

invokeai/app/api/routers/utilities.py

@@ -103,7 +103,7 @@ def _resolve_model_path(model_config_path: str) -> Path:
     return (base_models_path / model_path).resolve()
 
 
-def _run_expand_prompt(prompt: str, model_key: str, max_tokens: int, system_prompt: str | None) -> str:
+def _run_expand_prompt(prompt: str, model_key: str, max_tokens: int, system_prompt: str | None, user_id: str) -> str:
     """Run text LLM inference synchronously (called from thread)."""
     model_manager = ApiDependencies.invoker.services.model_manager
     model_config = model_manager.store.get_model(model_key)
@@ -112,7 +112,7 @@ def _run_expand_prompt(prompt: str, model_key: str, max_tokens: int, system_prom
         raise ValueError(f"Model '{model_key}' is not a TextLLM model (got {model_config.type})")
 
     with _model_load_lock:
-        loaded_model = model_manager.load.load_model(model_config)
+        loaded_model = model_manager.load.load_model(model_config, user_id=user_id)
 
     with torch.no_grad(), loaded_model.model_on_device() as (_, model):
         model_abs_path = _resolve_model_path(model_config.path)
@@ -147,6 +147,7 @@ async def expand_prompt(current_user: CurrentUserOrDefault, body: ExpandPromptRe
             body.model_key,
             body.max_tokens,
             body.system_prompt,
+            current_user.user_id,
         )
         return ExpandPromptResponse(expanded_prompt=expanded)
     except UnknownModelException:
@@ -172,7 +173,7 @@ class ImageToPromptResponse(BaseModel):
     error: str | None = None
 
 
-def _run_image_to_prompt(image_name: str, model_key: str, instruction: str) -> str:
+def _run_image_to_prompt(image_name: str, model_key: str, instruction: str, user_id: str) -> str:
     """Run LLaVA OneVision inference synchronously (called from thread)."""
     model_manager = ApiDependencies.invoker.services.model_manager
     model_config = model_manager.store.get_model(model_key)
@@ -181,7 +182,7 @@ def _run_image_to_prompt(image_name: str, model_key: str, instruction: str) -> s
         raise ValueError(f"Model '{model_key}' is not a LLaVA OneVision model (got {model_config.type})")
 
     with _model_load_lock:
-        loaded_model = model_manager.load.load_model(model_config)
+        loaded_model = model_manager.load.load_model(model_config, user_id=user_id)
 
     # Load the image from InvokeAI's image store
     image = ApiDependencies.invoker.services.images.get_pil_image(image_name)
@@ -229,6 +230,7 @@ async def image_to_prompt(current_user: CurrentUserOrDefault, body: ImageToPromp
             body.image_name,
             body.model_key,
             body.instruction,
+            current_user.user_id,
         )
         return ImageToPromptResponse(prompt=prompt)
     except UnknownModelException:

invokeai/app/api/sockets.py

@@ -325,18 +325,25 @@ async def _handle_queue_event(self, event: FastAPIEvent[QueueEventBase]):
             if isinstance(event_data, InvocationEventBase) and hasattr(event_data, "user_id"):
                 user_room = f"user:{event_data.user_id}"
 
-                # Emit to the user's room
-                await self._sio.emit(event=event_name, data=event_data.model_dump(mode="json"), room=user_room)
-
-                # Also emit to admin room so admins can see all events, but strip image preview data
-                # from InvocationProgressEvent to prevent admins from seeing other users' image content
                 if isinstance(event_data, InvocationProgressEvent):
-                    admin_event_data = event_data.model_copy(update={"image": None})
-                    await self._sio.emit(event=event_name, data=admin_event_data.model_dump(mode="json"), room="admin")
+                    # Progress events only drive personal UI (the global progress bar and progress
+                    # image previews) and are high-frequency. No admin UI consumes other users'
+                    # progress, so emit to the owner only. This also keeps other users' progress
+                    # from hijacking an admin's progress bar and image previews.
+                    await self._sio.emit(event=event_name, data=event_data.model_dump(mode="json"), room=user_room)
+                    logger.debug(f"Emitted invocation progress event to user room {user_room}")
                 else:
-                    await self._sio.emit(event=event_name, data=event_data.model_dump(mode="json"), room="admin")
-
-                logger.debug(f"Emitted private invocation event {event_name} to user room {user_room} and admin room")
+                    # started/complete/error also feed admins' gallery cache updates, so admins
+                    # receive them for all users. Emit to the union of owner + admin rooms in a
+                    # SINGLE call so an admin owner receives exactly one copy (see the
+                    # RecallParametersUpdatedEvent note below on python-socketio's recipient
+                    # dedup across a room list).
+                    await self._sio.emit(
+                        event=event_name, data=event_data.model_dump(mode="json"), room=[user_room, "admin"]
+                    )
+                    logger.debug(
+                        f"Emitted private invocation event {event_name} to user room {user_room} and admin room"
+                    )
 
             # QueueItemStatusChangedEvent: full to owner+admin, sanitized to everyone else in
             # the queue room so their queue list, badge, and item caches refresh.
@@ -490,7 +497,22 @@ async def _handle_queue_event(self, event: FastAPIEvent[QueueEventBase]):
             logger.error(f"Error handling queue event {event[0]}: {e}", exc_info=True)
 
     async def _handle_model_event(self, event: FastAPIEvent[ModelEventBase | DownloadEventBase]) -> None:
-        await self._sio.emit(event=event[0], data=event[1].model_dump(mode="json"))
+        event_name, event_data = event
+
+        # Model load events only drive personal UI (the loading-models spinner that puts the
+        # progress bar into indeterminate mode), so they are routed to the user whose action
+        # triggered the load. Broadcasting them made every user's progress bar animate whenever
+        # any user's generation loaded a model. In single-user mode the owner is "system" and
+        # every socket is in user:system, preserving the old behavior.
+        if isinstance(event_data, (ModelLoadStartedEvent, ModelLoadCompleteEvent)):
+            await self._sio.emit(
+                event=event_name, data=event_data.model_dump(mode="json"), room=f"user:{event_data.user_id}"
+            )
+            return
+
+        # Model install / download events remain broadcast to all connected sockets - they feed
+        # the model manager UI, which is not per-user.
+        await self._sio.emit(event=event_name, data=event_data.model_dump(mode="json"))
 
     async def _handle_bulk_image_download_event(self, event: FastAPIEvent[BulkDownloadEventBase]) -> None:
         event_name, event_data = event

invokeai/app/services/events/events_base.py

@@ -176,15 +176,17 @@ def emit_download_error(self, job: "DownloadJob") -> None:
 
     # region Model loading
 
-    def emit_model_load_started(self, config: "AnyModelConfig", submodel_type: Optional["SubModelType"] = None) -> None:
+    def emit_model_load_started(
+        self, config: "AnyModelConfig", submodel_type: Optional["SubModelType"] = None, user_id: str = "system"
+    ) -> None:
         """Emitted when a model load is started."""
-        self.dispatch(ModelLoadStartedEvent.build(config, submodel_type))
+        self.dispatch(ModelLoadStartedEvent.build(config, submodel_type, user_id))
 
     def emit_model_load_complete(
-        self, config: "AnyModelConfig", submodel_type: Optional["SubModelType"] = None
+        self, config: "AnyModelConfig", submodel_type: Optional["SubModelType"] = None, user_id: str = "system"
     ) -> None:
         """Emitted when a model load is complete."""
-        self.dispatch(ModelLoadCompleteEvent.build(config, submodel_type))
+        self.dispatch(ModelLoadCompleteEvent.build(config, submodel_type, user_id))
 
     # endregion
 

invokeai/app/services/events/events_common.py

@@ -510,10 +510,13 @@ class ModelLoadStartedEvent(ModelEventBase):
 
     config: AnyModelConfig = Field(description="The model's config")
     submodel_type: Optional[SubModelType] = Field(default=None, description="The submodel type, if any")
+    user_id: str = Field(default="system", description="The ID of the user whose action triggered the load")
 
     @classmethod
-    def build(cls, config: AnyModelConfig, submodel_type: Optional[SubModelType] = None) -> "ModelLoadStartedEvent":
-        return cls(config=config, submodel_type=submodel_type)
+    def build(
+        cls, config: AnyModelConfig, submodel_type: Optional[SubModelType] = None, user_id: str = "system"
+    ) -> "ModelLoadStartedEvent":
+        return cls(config=config, submodel_type=submodel_type, user_id=user_id)
 
 
 @payload_schema.register
@@ -524,10 +527,13 @@ class ModelLoadCompleteEvent(ModelEventBase):
 
     config: AnyModelConfig = Field(description="The model's config")
     submodel_type: Optional[SubModelType] = Field(default=None, description="The submodel type, if any")
+    user_id: str = Field(default="system", description="The ID of the user whose action triggered the load")
 
     @classmethod
-    def build(cls, config: AnyModelConfig, submodel_type: Optional[SubModelType] = None) -> "ModelLoadCompleteEvent":
-        return cls(config=config, submodel_type=submodel_type)
+    def build(
+        cls, config: AnyModelConfig, submodel_type: Optional[SubModelType] = None, user_id: str = "system"
+    ) -> "ModelLoadCompleteEvent":
+        return cls(config=config, submodel_type=submodel_type, user_id=user_id)
 
 
 @payload_schema.register

invokeai/app/services/model_load/model_load_base.py

@@ -15,12 +15,19 @@ class ModelLoadServiceBase(ABC):
     """Wrapper around AnyModelLoader."""
 
     @abstractmethod
-    def load_model(self, model_config: AnyModelConfig, submodel_type: Optional[SubModelType] = None) -> LoadedModel:
+    def load_model(
+        self,
+        model_config: AnyModelConfig,
+        submodel_type: Optional[SubModelType] = None,
+        user_id: Optional[str] = None,
+    ) -> LoadedModel:
         """
         Given a model's configuration, load it and return the LoadedModel object.
 
         :param model_config: Model configuration record (as returned by ModelRecordBase.get_model())
         :param submodel: For main (pipeline models), the submodel to fetch.
+        :param user_id: The user whose action triggered the load, threaded into the model load
+            events so they can be routed to that user's UI (defaults to the system user).
         """
 
     @property

invokeai/app/services/model_load/model_load_default.py

@@ -50,18 +50,25 @@ def ram_cache(self) -> ModelCache:
         """Return the RAM cache used by this loader."""
         return self._ram_cache
 
-    def load_model(self, model_config: AnyModelConfig, submodel_type: Optional[SubModelType] = None) -> LoadedModel:
+    def load_model(
+        self,
+        model_config: AnyModelConfig,
+        submodel_type: Optional[SubModelType] = None,
+        user_id: Optional[str] = None,
+    ) -> LoadedModel:
         """
         Given a model's configuration, load it and return the LoadedModel object.
 
         :param model_config: Model configuration record (as returned by ModelRecordBase.get_model())
         :param submodel: For main (pipeline models), the submodel to fetch.
+        :param user_id: The user whose action triggered the load, threaded into the model load
+            events so they can be routed to that user's UI (defaults to the system user).
         """
 
         # We don't have an invoker during testing
         # TODO(psyche): Mock this method on the invoker in the tests
         if hasattr(self, "_invoker"):
-            self._invoker.services.events.emit_model_load_started(model_config, submodel_type)
+            self._invoker.services.events.emit_model_load_started(model_config, submodel_type, user_id or "system")
 
         implementation, model_config, submodel_type = self._registry.get_implementation(model_config, submodel_type)  # type: ignore
         loaded_model: LoadedModel = implementation(
@@ -71,7 +78,7 @@ def load_model(self, model_config: AnyModelConfig, submodel_type: Optional[SubMo
         ).load_model(model_config, submodel_type)
 
         if hasattr(self, "_invoker"):
-            self._invoker.services.events.emit_model_load_complete(model_config, submodel_type)
+            self._invoker.services.events.emit_model_load_complete(model_config, submodel_type, user_id or "system")
 
         return loaded_model
 

invokeai/app/services/session_queue/session_queue_base.py

@@ -79,18 +79,25 @@ def get_queue_status(
         queue_id: str,
         user_id: Optional[str] = None,
         acting_user_id: Optional[str] = None,
+        is_admin: bool = False,
     ) -> SessionQueueStatus:
         """Gets the status of the queue.
 
         Aggregate counts (pending/in_progress/.../total) are always global across all users.
         If user_id is provided, the requesting user's own counts are additionally returned in
-        the user_pending/user_in_progress fields (left None otherwise).
+        the user_pending/user_in_progress fields (left None otherwise). Admin callers should
+        also pass their user_id so personal UI (e.g. the progress bar) can distinguish their
+        own activity from other users'.
 
         acting_user_id is independent of user_id and controls only current-item redaction:
         when set, the returned status omits item_id/session_id/batch_id unless the
         currently-running item belongs to acting_user_id. The redaction is decided from the
         same get_current() snapshot used to embed those identifiers, so it cannot race against
         a concurrent state change.
+
+        is_admin disables current-item redaction entirely: admins may see the identifiers of
+        any user's current item. Redaction stays fail-closed - a caller that passes user_id
+        without is_admin gets the non-admin behavior.
         """
         pass
 

invokeai/app/services/session_queue/session_queue_sqlite.py

@@ -1143,6 +1143,7 @@ def get_queue_status(
         queue_id: str,
         user_id: Optional[str] = None,
         acting_user_id: Optional[str] = None,
+        is_admin: bool = False,
     ) -> SessionQueueStatus:
         with self._db.transaction() as cursor:
             # Aggregate counts are always global (across all users). This lets a non-admin's
@@ -1190,11 +1191,13 @@ def get_queue_status(
         # so a concurrent transition (e.g. B finishing while A's status changes) cannot leave
         # stale identifiers in the result. The aggregate counts stay global; only the current
         # item's identifiers are gated. acting_user_id (event path) takes precedence over
-        # user_id (API path) when deciding the redaction owner; either being None means an
-        # admin/global caller who may see the current item.
+        # user_id (API path) when deciding the redaction owner; either being None means a
+        # global caller who may see the current item. is_admin disables redaction outright so
+        # admin callers can pass their user_id (for the per-user counts) without losing
+        # visibility of other users' current item.
         owner_user_id = user_id if acting_user_id is None else acting_user_id
         show_current_item = current_item is not None and (
-            owner_user_id is None or current_item.user_id == owner_user_id
+            is_admin or owner_user_id is None or current_item.user_id == owner_user_id
         )
 
         return SessionQueueStatus(

invokeai/app/services/shared/invocation_context.py

@@ -394,7 +394,7 @@ def load(
         if submodel_type:
             message += f" ({submodel_type.value})"
         self._util.signal_progress(message)
-        return self._services.model_manager.load.load_model(model, submodel_type)
+        return self._services.model_manager.load.load_model(model, submodel_type, user_id=self._data.queue_item.user_id)
 
     def load_by_attrs(
         self, name: str, base: BaseModelType, type: ModelType, submodel_type: Optional[SubModelType] = None
@@ -424,7 +424,9 @@ def load_by_attrs(
         if submodel_type:
             message += f" ({submodel_type.value})"
         self._util.signal_progress(message)
-        return self._services.model_manager.load.load_model(configs[0], submodel_type)
+        return self._services.model_manager.load.load_model(
+            configs[0], submodel_type, user_id=self._data.queue_item.user_id
+        )
 
     @staticmethod
     def _raise_if_external(model: AnyModelConfig) -> None:

invokeai/frontend/web/openapi.json

@@ -7738,7 +7738,7 @@
       "get": {
         "tags": ["queue"],
         "summary": "Get Queue Status",
-        "description": "Gets the status of the session queue. Returns global counts; non-admin users additionally\nget their own pending/in_progress counts (so the UI can show an X/Y badge) and cannot see the\ncurrent item's identifiers unless they own it.",
+        "description": "Gets the status of the session queue. Returns global counts; every user additionally gets\ntheir own pending/in_progress counts (so the UI can show an X/Y badge and scope personal UI\nlike the progress bar to the user's own activity). Non-admin users cannot see the current\nitem's identifiers unless they own it.",
         "operationId": "get_queue_status",
         "security": [
           {
@@ -57112,9 +57112,15 @@
             ],
             "default": null,
             "description": "The submodel type, if any"
+          },
+          "user_id": {
+            "default": "system",
+            "description": "The ID of the user whose action triggered the load",
+            "title": "User Id",
+            "type": "string"
           }
         },
-        "required": ["timestamp", "config", "submodel_type"],
+        "required": ["timestamp", "config", "submodel_type", "user_id"],
         "title": "ModelLoadCompleteEvent",
         "type": "object"
       },
@@ -57422,9 +57428,15 @@
             ],
             "default": null,
             "description": "The submodel type, if any"
+          },
+          "user_id": {
+            "default": "system",
+            "description": "The ID of the user whose action triggered the load",
+            "title": "User Id",
+            "type": "string"
           }
         },
-        "required": ["timestamp", "config", "submodel_type"],
+        "required": ["timestamp", "config", "submodel_type", "user_id"],
         "title": "ModelLoadStartedEvent",
         "type": "object"
       },