feat: Round-robin session queue scheduling across users

docs/src/content/docs/configuration/invokeai-yaml.mdx

@@ -114,6 +114,25 @@ Most common algorithms are supported, like `md5`, `sha256`, and `sha512`. These
 
 These options set the paths of various directories and files used by InvokeAI. Any user-defined paths should be absolute paths.
 
+#### Image Subfolder Strategy
+
+By default, generated images are stored in a single flat directory under `outputs/images/`. The `image_subfolder_strategy` setting lets you organize newly-created images into subfolders automatically. You can edit this setting in `invokeai.yaml` or, as an admin user, in the Settings panel.
+
+```yaml
+image_subfolder_strategy: flat # default value
+```
+
+Available strategies:
+
+| Strategy | Example Path                           | Description                                                                                       |
+| -------- | -------------------------------------- | ------------------------------------------------------------------------------------------------- |
+| `flat`   | `outputs/images/abc123.png`            | Store images directly in the images directory.                                                    |
+| `date`   | `outputs/images/2026/03/17/abc123.png` | Organize images by creation date.                                                                 |
+| `type`   | `outputs/images/general/abc123.png`    | Organize images by image category.                                                                |
+| `hash`   | `outputs/images/ab/abc123.png`         | Use the first two characters of the image UUID for filesystem performance with large collections. |
+
+Changing this setting only affects newly-created images. Existing images remain in their current locations.
+
 #### Logging
 
 Several different log handler destinations are available, and multiple destinations are supported by providing a list:

docs/src/generated/settings.json

@@ -590,6 +590,20 @@
       "type": "<class 'int'>",
       "validation": {}
     },
+    {
+      "category": "GENERATION",
+      "default": "round_robin",
+      "description": "Session queue mode. Use 'FIFO' for traditional first-in-first-out, or 'round_robin' to serve each user's jobs in turn. In single-user mode, FIFO is always used regardless of this setting.",
+      "env_var": "INVOKEAI_SESSION_QUEUE_MODE",
+      "literal_values": [
+        "FIFO",
+        "round_robin"
+      ],
+      "name": "session_queue_mode",
+      "required": false,
+      "type": "typing.Literal['FIFO', 'round_robin']",
+      "validation": {}
+    },
     {
       "category": "GENERATION",
       "default": false,

invokeai/app/api/routers/app_info.py

@@ -3,17 +3,19 @@
 from importlib.metadata import distributions
 from pathlib import Path as FilePath
 from threading import Lock
+from typing import Any
 
 import torch
 import yaml
 from fastapi import Body, HTTPException, Path
 from fastapi.routing import APIRouter
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, model_validator
 
 from invokeai.app.api.auth_dependencies import AdminUserOrDefault
 from invokeai.app.api.dependencies import ApiDependencies
 from invokeai.app.services.config.config_default import (
     EXTERNAL_PROVIDER_CONFIG_FIELDS,
+    IMAGE_SUBFOLDER_STRATEGY,
     DefaultInvokeAIAppConfig,
     InvokeAIAppConfig,
     get_config,
@@ -105,15 +107,37 @@ class ExternalProviderConfigModel(BaseModel):
 _EXTERNAL_PROVIDER_CONFIG_LOCK = Lock()
 
 
+def _remove_nullable_default_from_schema(schema: dict[str, Any]) -> None:
+    schema.pop("default", None)
+    any_of = schema.pop("anyOf", None)
+    if isinstance(any_of, list):
+        non_null_schemas = [
+            subschema for subschema in any_of if isinstance(subschema, dict) and subschema.get("type") != "null"
+        ]
+        if len(non_null_schemas) == 1:
+            schema.update(non_null_schemas[0])
+
+
 class UpdateAppGenerationSettingsRequest(BaseModel):
     """Writable generation-related app settings."""
 
+    image_subfolder_strategy: IMAGE_SUBFOLDER_STRATEGY | None = Field(
+        default=None,
+        description="Strategy for organizing images into subfolders.",
+        json_schema_extra=_remove_nullable_default_from_schema,
+    )
     max_queue_history: int | None = Field(
         default=None,
         ge=0,
         description="Keep the last N completed, failed, and canceled queue items on startup. Set to 0 to prune all terminal items.",
     )
 
+    @model_validator(mode="after")
+    def validate_explicit_nulls(self) -> "UpdateAppGenerationSettingsRequest":
+        if "image_subfolder_strategy" in self.model_fields_set and self.image_subfolder_strategy is None:
+            raise ValueError("image_subfolder_strategy may not be null")
+        return self
+
 
 @app_router.get(
     "/runtime_config", operation_id="get_runtime_config", status_code=200, response_model=InvokeAIAppConfigWithSetFields
@@ -133,18 +157,19 @@ async def update_runtime_config(
     _: AdminUserOrDefault,
     changes: UpdateAppGenerationSettingsRequest = Body(description="Writable runtime configuration changes"),
 ) -> InvokeAIAppConfigWithSetFields:
-    config = get_config()
-    update_dict = changes.model_dump(exclude_unset=True)
-    config.update_config(update_dict)
+    with _EXTERNAL_PROVIDER_CONFIG_LOCK:
+        config = get_config()
+        update_dict = changes.model_dump(exclude_unset=True)
+        config.update_config(update_dict)
 
-    if config.config_file_path.exists():
-        persisted_config = load_and_migrate_config(config.config_file_path)
-    else:
-        persisted_config = DefaultInvokeAIAppConfig()
+        if config.config_file_path.exists():
+            persisted_config = load_and_migrate_config(config.config_file_path)
+        else:
+            persisted_config = DefaultInvokeAIAppConfig()
 
-    persisted_config.update_config(update_dict)
-    persisted_config.write_file(config.config_file_path)
-    return InvokeAIAppConfigWithSetFields(set_fields=config.model_fields_set, config=config)
+        persisted_config.update_config(update_dict)
+        persisted_config.write_file(config.config_file_path)
+        return InvokeAIAppConfigWithSetFields(set_fields=config.model_fields_set, config=config)
 
 
 @app_router.get(

invokeai/app/api/routers/session_queue.py

@@ -141,12 +141,11 @@ async def get_queue_item_ids(
     queue_id: str = Path(description="The queue id to perform this operation on"),
     order_dir: SQLiteDirection = Query(default=SQLiteDirection.Descending, description="The order of sort"),
 ) -> ItemIdsResult:
-    """Gets all queue item ids that match the given parameters. Non-admin users only see their own items."""
+    """Gets all queue item ids that match the given parameters. The IDs themselves are not sensitive;
+    per-item field redaction is performed when the items are fetched via list_all_queue_items or
+    get_queue_items_by_item_ids."""
     try:
-        user_id = None if current_user.is_admin else current_user.user_id
-        return ApiDependencies.invoker.services.session_queue.get_queue_item_ids(
-            queue_id=queue_id, order_dir=order_dir, user_id=user_id
-        )
+        return ApiDependencies.invoker.services.session_queue.get_queue_item_ids(queue_id=queue_id, order_dir=order_dir)
     except Exception as e:
         raise HTTPException(status_code=500, detail=f"Unexpected error while listing all queue item ids: {e}")
 
@@ -436,10 +435,15 @@ 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. Non-admin users see only their own counts and cannot see current item details unless they own it."""
+    """Gets the status of the session queue. Returns global counts plus the calling user's own
+    pending/in_progress counts (so the UI can show an X/Y badge). 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/sockets.py

@@ -260,20 +260,37 @@ async def _handle_sub_bulk_download(self, sid: str, data: Any) -> None:
     async def _handle_unsub_bulk_download(self, sid: str, data: Any) -> None:
         await self._sio.leave_room(sid, BulkDownloadSubscriptionEvent(**data).bulk_download_id)
 
+    def _owner_and_admin_sids(self, owner_user_id: str) -> list[str]:
+        """Sids belonging to the event's owner or to any admin.
+
+        Used as `skip_sid` when broadcasting a sanitized companion event to the queue room,
+        so the owner and admins (who already received the full event) don't get a second
+        copy that would clobber their cache with redacted values.
+        """
+        return [
+            sid
+            for sid, info in self._socket_users.items()
+            if info.get("user_id") == owner_user_id or info.get("is_admin")
+        ]
+
     async def _handle_queue_event(self, event: FastAPIEvent[QueueEventBase]):
         """Handle queue events with user isolation.
 
-        All queue item events (invocation events AND QueueItemStatusChangedEvent) are
-        private to the owning user and admins. They carry unsanitized user_id, batch_id,
-        session_id, origin, destination and error metadata, and must never be broadcast
-        to the whole queue room — otherwise any other authenticated subscriber could
-        observe cross-user queue activity.
+        Queue events split into two routing paths:
 
-        RecallParametersUpdatedEvent is also private to the owner + admins.
+        1. The owner and admins receive the full unsanitized event in their `user:{id}` /
+           `admin` rooms. The full payload may include batch_id, session_id, origin,
+           destination, error metadata, etc.
 
-        BatchEnqueuedEvent carries the enqueuing user's batch_id/origin/counts and
-        is also routed privately. QueueClearedEvent is the only queue event that
-        is still broadcast to the whole queue room.
+        2. For events that other authenticated users need to know about so their queue list
+           and badge counts stay in sync (QueueItemStatusChangedEvent and BatchEnqueuedEvent),
+           a sanitized companion event is also emitted to the full queue room with the
+           owner's and admins' sids in `skip_sid`. The companion uses `user_id="redacted"`
+           as a sentinel so the frontend handler knows to do tag invalidation only and skip
+           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.
 
         IMPORTANT: Check InvocationEventBase BEFORE QueueItemEventBase since InvocationEventBase
         inherits from QueueItemEventBase. The order of isinstance checks matters!
@@ -302,10 +319,51 @@ async def _handle_queue_event(self, event: FastAPIEvent[QueueEventBase]):
 
                 logger.debug(f"Emitted private invocation event {event_name} to user room {user_room} and admin room")
 
-            # Other queue item events (QueueItemStatusChangedEvent) carry unsanitized
-            # user_id, batch_id, session_id, origin, destination and error metadata.
-            # They are private to the owning user + admins — never broadcast to the
-            # full queue room.
+            # QueueItemStatusChangedEvent: full to owner+admin, sanitized to everyone else in
+            # the queue room so their queue list, badge, and item caches refresh.
+            elif isinstance(event_data, QueueItemStatusChangedEvent):
+                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)
+                await self._sio.emit(event=event_name, data=event_data.model_dump(mode="json"), room="admin")
+
+                sanitized = event_data.model_copy(
+                    update={
+                        "user_id": "redacted",
+                        "batch_id": "redacted",
+                        "session_id": "redacted",
+                        "origin": None,
+                        "destination": None,
+                        "error_type": None,
+                        "error_message": None,
+                        "error_traceback": None,
+                    }
+                )
+                # Strip identifying fields out of the embedded batch_status / queue_status too.
+                sanitized.batch_status = sanitized.batch_status.model_copy(
+                    update={"batch_id": "redacted", "origin": None, "destination": None}
+                )
+                sanitized.queue_status = sanitized.queue_status.model_copy(
+                    update={
+                        "item_id": None,
+                        "session_id": None,
+                        "batch_id": None,
+                        "user_pending": None,
+                        "user_in_progress": None,
+                    }
+                )
+                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_item_status_changed: full to {user_room}+admin, sanitized to queue {event_data.queue_id}"
+                )
+
+            # Other queue item events (currently none beyond QueueItemStatusChangedEvent that
+            # carry user_id) stay private to owner + admins.
             elif isinstance(event_data, QueueItemEventBase) and hasattr(event_data, "user_id"):
                 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)
@@ -320,14 +378,25 @@ 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 recall_parameters_updated event to user room {user_room} and admin room")
 
-            # BatchEnqueuedEvent carries the enqueuing user's batch_id, origin, and
-            # enqueued counts. Route it privately to the owner + admins so other
-            # users do not observe cross-user batch activity.
+            # BatchEnqueuedEvent: full to owner+admin, sanitized to everyone else in the queue
+            # room so their badge total and queue list pick up the new items.
             elif isinstance(event_data, BatchEnqueuedEvent):
                 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)
                 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")
+
+                sanitized = event_data.model_copy(
+                    update={"user_id": "redacted", "batch_id": "redacted", "origin": None}
+                )
+                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 batch_enqueued: full to {user_room}+admin, sanitized to queue {event_data.queue_id}"
+                )
 
             else:
                 # For remaining queue events (e.g. QueueClearedEvent) that do not

invokeai/app/services/config/config_default.py

@@ -30,6 +30,7 @@
 ATTENTION_SLICE_SIZE = Literal["auto", "balanced", "max", 1, 2, 3, 4, 5, 6, 7, 8]
 LOG_FORMAT = Literal["plain", "color", "syslog", "legacy"]
 LOG_LEVEL = Literal["debug", "info", "warning", "error", "critical"]
+SESSION_QUEUE_MODE = Literal["FIFO", "round_robin"]
 IMAGE_SUBFOLDER_STRATEGY = Literal["flat", "date", "type", "hash"]
 CONFIG_SCHEMA_VERSION = "4.0.3"
 EXTERNAL_PROVIDER_CONFIG_FIELDS = (
@@ -114,6 +115,7 @@ class InvokeAIAppConfig(BaseSettings):
         force_tiled_decode: Whether to enable tiled VAE decode (reduces memory consumption with some performance penalty).
         pil_compress_level: The compress_level setting of PIL.Image.save(), used for PNG encoding. All settings are lossless. 0 = no compression, 1 = fastest with slightly larger filesize, 9 = slowest with smallest filesize. 1 is typically the best setting.
         max_queue_size: Maximum number of items in the session queue.
+        session_queue_mode: Session queue mode. Use 'FIFO' for traditional first-in-first-out, or 'round_robin' to serve each user's jobs in turn. In single-user mode, FIFO is always used regardless of this setting.<br>Valid values: `FIFO`, `round_robin`
         clear_queue_on_startup: Empties session queue on startup. If true, disables `max_queue_history`.
         max_queue_history: Keep the last N completed, failed, and canceled queue items. Older items are deleted on startup. Set to 0 to prune all terminal items. Ignored if `clear_queue_on_startup` is true.
         allow_nodes: List of nodes to allow. Omit to allow all.
@@ -214,6 +216,7 @@ class InvokeAIAppConfig(BaseSettings):
     force_tiled_decode:            bool = Field(default=False,              description="Whether to enable tiled VAE decode (reduces memory consumption with some performance penalty).")
     pil_compress_level:             int = Field(default=1,                  description="The compress_level setting of PIL.Image.save(), used for PNG encoding. All settings are lossless. 0 = no compression, 1 = fastest with slightly larger filesize, 9 = slowest with smallest filesize. 1 is typically the best setting.")
     max_queue_size:                 int = Field(default=10000, gt=0,        description="Maximum number of items in the session queue.")
+    session_queue_mode: SESSION_QUEUE_MODE = Field(default="round_robin",   description="Session queue mode. Use 'FIFO' for traditional first-in-first-out, or 'round_robin' to serve each user's jobs in turn. In single-user mode, FIFO is always used regardless of this setting.")
     clear_queue_on_startup:        bool = Field(default=False,              description="Empties session queue on startup. If true, disables `max_queue_history`.")
     max_queue_history:      Optional[int] = Field(default=None, ge=0,        description="Keep the last N completed, failed, and canceled queue items. Older items are deleted on startup. Set to 0 to prune all terminal items. Ignored if `clear_queue_on_startup` is true.")
 

invokeai/app/services/session_queue/session_queue_common.py

@@ -309,6 +309,12 @@ class SessionQueueStatus(BaseModel):
     failed: int = Field(..., description="Number of queue items with status 'error'")
     canceled: int = Field(..., description="Number of queue items with status 'canceled'")
     total: int = Field(..., description="Total number of queue items")
+    user_pending: Optional[int] = Field(
+        default=None, description="Number of pending queue items for the calling user (multiuser only)"
+    )
+    user_in_progress: Optional[int] = Field(
+        default=None, description="Number of in-progress queue items for the calling user (multiuser only)"
+    )
 
 
 class SessionQueueCountsByDestination(BaseModel):