feat: add configurable image subfolder strategies

docs/configuration.md

@@ -39,7 +39,7 @@ It has two sections - one for internal use and one for user settings:
 
 ```yaml
 # Internal metadata - do not edit:
-schema_version: 4.0.2
+schema_version: 4.0.3
 
 # Put user settings here - see https://invoke-ai.github.io/InvokeAI/features/CONFIGURATION/:
 host: 0.0.0.0 # serve the app on your local network
@@ -144,6 +144,35 @@ 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, all generated images are stored in a single flat directory (`outputs/images/`). This can become unwieldy with a large number of images. The `image_subfolder_strategy` setting lets you organize images into subfolders automatically.
+
+```yaml
+image_subfolder_strategy: flat # default value
+```
+
+Available strategies:
+
+| Strategy | Example Path | Description |
+|----------|-------------|-------------|
+| `flat` | `outputs/images/abc123.png` | **Default.** All images in one directory (current behavior). |
+| `date` | `outputs/images/2026/03/17/abc123.png` | Organized by creation date (YYYY/MM/DD). |
+| `type` | `outputs/images/general/abc123.png` | Organized by image category (`general`, `intermediate`, `mask`, `control`, etc.). |
+| `hash` | `outputs/images/ab/abc123.png` | Uses first 2 characters of the UUID as subfolder. Best for filesystem performance with very large collections (~256 evenly distributed subfolders). |
+
+!!! tip "Switching Strategies"
+
+    You can switch between strategies at any time. Existing images remain in their original location — only newly generated images will use the new subfolder structure. This works because each image's subfolder path is stored in the database.
+
+!!! example "Example: Using date-based organization"
+
+    ```yaml
+    image_subfolder_strategy: date
+    ```
+
+    New images will be saved as `outputs/images/2026/03/17/abc123.png`. Thumbnails mirror the same structure under `outputs/images/thumbnails/2026/03/17/abc123.webp`.
+
 #### Logging
 
 Several different log handler destinations are available, and multiple destinations are supported by providing a list:

invokeai/app/api/routers/recall_parameters.py

@@ -146,17 +146,15 @@ def load_image_file(image_name: str) -> Optional[dict[str, Any]]:
     """
     logger = ApiDependencies.invoker.services.logger
     try:
-        # Prefer using the image_files service to validate & open images
-        image_files = ApiDependencies.invoker.services.image_files
-        # Resolve a safe path inside outputs
-        image_path = image_files.get_path(image_name)
+        images_service = ApiDependencies.invoker.services.images
+        # Use images service which handles subfolder resolution via DB record
+        path = images_service.get_path(image_name)
 
-        if not image_files.validate_path(str(image_path)):
-            logger.warning(f"Image file not found: {image_name} (searched in {image_path.parent})")
+        if not images_service.validate_path(path):
+            logger.warning(f"Image file not found: {image_name}")
             return None
 
-        # Open the image via service to leverage caching
-        pil_image = image_files.get(image_name)
+        pil_image = images_service.get_pil_image(image_name)
         width, height = pil_image.size
         logger.info(f"Found image file: {image_name} ({width}x{height})")
         return {"image_name": image_name, "width": width, "height": height}

invokeai/app/services/config/config_default.py

@@ -29,7 +29,8 @@
 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"]
-CONFIG_SCHEMA_VERSION = "4.0.2"
+IMAGE_SUBFOLDER_STRATEGY = Literal["flat", "date", "type", "hash"]
+CONFIG_SCHEMA_VERSION = "4.0.3"
 
 
 class URLRegexTokenPair(BaseModel):
@@ -69,6 +70,7 @@ class InvokeAIAppConfig(BaseSettings):
         legacy_conf_dir: Path to directory of legacy checkpoint config files.
         db_dir: Path to InvokeAI databases directory.
         outputs_dir: Path to directory for outputs.
+        image_subfolder_strategy: Strategy for organizing images into subfolders. 'flat' stores all images in a single folder. 'date' organizes by YYYY/MM/DD. 'type' organizes by image category. 'hash' uses first 2 characters of UUID for filesystem performance.<br>Valid values: `flat`, `date`, `type`, `hash`
         custom_nodes_dir: Path to directory for custom nodes.
         style_presets_dir: Path to directory for style presets.
         workflow_thumbnails_dir: Path to directory for workflow thumbnails.
@@ -145,6 +147,7 @@ class InvokeAIAppConfig(BaseSettings):
     legacy_conf_dir:               Path = Field(default=Path("configs"), description="Path to directory of legacy checkpoint config files.")
     db_dir:                        Path = Field(default=Path("databases"),  description="Path to InvokeAI databases directory.")
     outputs_dir:                   Path = Field(default=Path("outputs"),    description="Path to directory for outputs.")
+    image_subfolder_strategy: IMAGE_SUBFOLDER_STRATEGY = Field(default="flat", description="Strategy for organizing images into subfolders. 'flat' stores all images in a single folder. 'date' organizes by YYYY/MM/DD. 'type' organizes by image category. 'hash' uses first 2 characters of UUID for filesystem performance.")
     custom_nodes_dir:              Path = Field(default=Path("nodes"),      description="Path to directory for custom nodes.")
     style_presets_dir:      Path = Field(default=Path("style_presets"),      description="Path to directory for style presets.")
     workflow_thumbnails_dir: Path = Field(default=Path("workflow_thumbnails"), description="Path to directory for workflow thumbnails.")
@@ -455,6 +458,20 @@ def migrate_v4_0_1_to_4_0_2_config_dict(config_dict: dict[str, Any]) -> dict[str
     return parsed_config_dict
 
 
+def migrate_v4_0_2_to_4_0_3_config_dict(config_dict: dict[str, Any]) -> dict[str, Any]:
+    """Migrate v4.0.2 config dictionary to a v4.0.3 config dictionary.
+
+    Args:
+        config_dict: A dictionary of settings from a v4.0.2 config file.
+
+    Returns:
+        A config dict with the settings migrated to v4.0.3.
+    """
+    parsed_config_dict: dict[str, Any] = copy.deepcopy(config_dict)
+    parsed_config_dict["schema_version"] = "4.0.3"
+    return parsed_config_dict
+
+
 def load_and_migrate_config(config_path: Path) -> InvokeAIAppConfig:
     """Load and migrate a config file to the latest version.
 
@@ -480,6 +497,9 @@ def load_and_migrate_config(config_path: Path) -> InvokeAIAppConfig:
     if loaded_config_dict["schema_version"] == "4.0.1":
         migrated = True
         loaded_config_dict = migrate_v4_0_1_to_4_0_2_config_dict(loaded_config_dict)
+    if loaded_config_dict["schema_version"] == "4.0.2":
+        migrated = True
+        loaded_config_dict = migrate_v4_0_2_to_4_0_3_config_dict(loaded_config_dict)
 
     if migrated:
         shutil.copy(config_path, config_path.with_suffix(".yaml.bak"))

invokeai/app/services/image_files/image_files_base.py

@@ -9,12 +9,12 @@ class ImageFileStorageBase(ABC):
     """Low-level service responsible for storing and retrieving image files."""
 
     @abstractmethod
-    def get(self, image_name: str) -> PILImageType:
+    def get(self, image_name: str, image_subfolder: str = "") -> PILImageType:
         """Retrieves an image as PIL Image."""
         pass
 
     @abstractmethod
-    def get_path(self, image_name: str, thumbnail: bool = False) -> Path:
+    def get_path(self, image_name: str, thumbnail: bool = False, image_subfolder: str = "") -> Path:
         """Gets the internal path to an image or thumbnail."""
         pass
 
@@ -34,21 +34,22 @@ def save(
         workflow: Optional[str] = None,
         graph: Optional[str] = None,
         thumbnail_size: int = 256,
+        image_subfolder: str = "",
     ) -> None:
         """Saves an image and a 256x256 WEBP thumbnail. Returns a tuple of the image name, thumbnail name, and created timestamp."""
         pass
 
     @abstractmethod
-    def delete(self, image_name: str) -> None:
+    def delete(self, image_name: str, image_subfolder: str = "") -> None:
         """Deletes an image and its thumbnail (if one exists)."""
         pass
 
     @abstractmethod
-    def get_workflow(self, image_name: str) -> Optional[str]:
+    def get_workflow(self, image_name: str, image_subfolder: str = "") -> Optional[str]:
         """Gets the workflow of an image."""
         pass
 
     @abstractmethod
-    def get_graph(self, image_name: str) -> Optional[str]:
+    def get_graph(self, image_name: str, image_subfolder: str = "") -> Optional[str]:
         """Gets the graph of an image."""
         pass

invokeai/app/services/image_files/image_files_disk.py

@@ -32,9 +32,9 @@ def __init__(self, output_folder: Union[str, Path]):
     def start(self, invoker: Invoker) -> None:
         self.__invoker = invoker
 
-    def get(self, image_name: str) -> PILImageType:
+    def get(self, image_name: str, image_subfolder: str = "") -> PILImageType:
         try:
-            image_path = self.get_path(image_name)
+            image_path = self.get_path(image_name, image_subfolder=image_subfolder)
 
             cache_item = self.__get_cache(image_path)
             if cache_item:
@@ -54,10 +54,14 @@ def save(
         workflow: Optional[str] = None,
         graph: Optional[str] = None,
         thumbnail_size: int = 256,
+        image_subfolder: str = "",
     ) -> None:
         try:
             self.__validate_storage_folders()
-            image_path = self.get_path(image_name)
+            image_path = self.get_path(image_name, image_subfolder=image_subfolder)
+
+            # Ensure subfolder directories exist
+            image_path.parent.mkdir(parents=True, exist_ok=True)
 
             pnginfo = PngImagePlugin.PngInfo()
             info_dict = {}
@@ -82,7 +86,11 @@ def save(
             )
 
             thumbnail_name = get_thumbnail_name(image_name)
-            thumbnail_path = self.get_path(thumbnail_name, thumbnail=True)
+            thumbnail_path = self.get_path(thumbnail_name, thumbnail=True, image_subfolder=image_subfolder)
+
+            # Ensure thumbnail subfolder directories exist
+            thumbnail_path.parent.mkdir(parents=True, exist_ok=True)
+
             thumbnail_image = make_thumbnail(image, thumbnail_size)
             thumbnail_image.save(thumbnail_path)
 
@@ -91,17 +99,17 @@ def save(
         except Exception as e:
             raise ImageFileSaveException from e
 
-    def delete(self, image_name: str) -> None:
+    def delete(self, image_name: str, image_subfolder: str = "") -> None:
         try:
-            image_path = self.get_path(image_name)
+            image_path = self.get_path(image_name, image_subfolder=image_subfolder)
 
             if image_path.exists():
                 image_path.unlink()
             if image_path in self.__cache:
                 del self.__cache[image_path]
 
             thumbnail_name = get_thumbnail_name(image_name)
-            thumbnail_path = self.get_path(thumbnail_name, True)
+            thumbnail_path = self.get_path(thumbnail_name, True, image_subfolder=image_subfolder)
 
             if thumbnail_path.exists():
                 thumbnail_path.unlink()
@@ -110,17 +118,21 @@ def delete(self, image_name: str) -> None:
         except Exception as e:
             raise ImageFileDeleteException from e
 
-    def get_path(self, image_name: str, thumbnail: bool = False) -> Path:
+    def get_path(self, image_name: str, thumbnail: bool = False, image_subfolder: str = "") -> Path:
         base_folder = self.__thumbnails_folder if thumbnail else self.__output_folder
         filename = get_thumbnail_name(image_name) if thumbnail else image_name
 
-        # Strip any path information from the filename
+        # Validate the filename itself (no path separators allowed in the filename)
         basename = Path(filename).name
-
         if basename != filename:
             raise ValueError("Invalid image name, potential directory traversal detected")
 
-        image_path = base_folder / basename
+        # Build the full path with optional subfolder
+        if image_subfolder:
+            self._validate_subfolder(image_subfolder)
+            image_path = base_folder / image_subfolder / basename
+        else:
+            image_path = base_folder / basename
 
         # Ensure the image path is within the base folder to prevent directory traversal
         resolved_base = base_folder.resolve()
@@ -131,20 +143,36 @@ def get_path(self, image_name: str, thumbnail: bool = False) -> Path:
 
         return resolved_image_path
 
+    @staticmethod
+    def _validate_subfolder(subfolder: str) -> None:
+        """Validates a subfolder path to prevent directory traversal while allowing controlled subdirectories."""
+        if not subfolder:
+            return
+        if "\\" in subfolder:
+            raise ValueError("Backslashes not allowed in subfolder path")
+        if subfolder.startswith("/"):
+            raise ValueError("Absolute paths not allowed in subfolder path")
+        parts = subfolder.split("/")
+        for part in parts:
+            if part == "..":
+                raise ValueError("Parent directory references not allowed in subfolder path")
+            if part == "":
+                raise ValueError("Empty path segments not allowed in subfolder path")
+
     def validate_path(self, path: Union[str, Path]) -> bool:
         """Validates the path given for an image or thumbnail."""
         path = path if isinstance(path, Path) else Path(path)
         return path.exists()
 
-    def get_workflow(self, image_name: str) -> str | None:
-        image = self.get(image_name)
+    def get_workflow(self, image_name: str, image_subfolder: str = "") -> str | None:
+        image = self.get(image_name, image_subfolder=image_subfolder)
         workflow = image.info.get("invokeai_workflow", None)
         if isinstance(workflow, str):
             return workflow
         return None
 
-    def get_graph(self, image_name: str) -> str | None:
-        image = self.get(image_name)
+    def get_graph(self, image_name: str, image_subfolder: str = "") -> str | None:
+        image = self.get(image_name, image_subfolder=image_subfolder)
         graph = image.info.get("invokeai_graph", None)
         if isinstance(graph, str):
             return graph

invokeai/app/services/image_files/image_subfolder_strategy.py

@@ -0,0 +1,58 @@
+from abc import ABC, abstractmethod
+from datetime import datetime
+
+from invokeai.app.services.image_records.image_records_common import ImageCategory
+
+
+class ImageSubfolderStrategy(ABC):
+    """Base class for image subfolder strategies."""
+
+    @abstractmethod
+    def get_subfolder(self, image_name: str, image_category: ImageCategory, is_intermediate: bool) -> str:
+        """Returns relative subfolder prefix (e.g. '2026/03/17', 'general'), or empty string for flat."""
+        pass
+
+
+class FlatStrategy(ImageSubfolderStrategy):
+    """No subfolders - all images in one directory (default behavior)."""
+
+    def get_subfolder(self, image_name: str, image_category: ImageCategory, is_intermediate: bool) -> str:
+        return ""
+
+
+class DateStrategy(ImageSubfolderStrategy):
+    """Organize images by date: YYYY/MM/DD."""
+
+    def get_subfolder(self, image_name: str, image_category: ImageCategory, is_intermediate: bool) -> str:
+        now = datetime.now()
+        return f"{now.year}/{now.month:02d}/{now.day:02d}"
+
+
+class TypeStrategy(ImageSubfolderStrategy):
+    """Organize images by category/type: general, intermediate, mask, control, etc."""
+
+    def get_subfolder(self, image_name: str, image_category: ImageCategory, is_intermediate: bool) -> str:
+        if is_intermediate:
+            return "intermediate"
+        return image_category.value
+
+
+class HashStrategy(ImageSubfolderStrategy):
+    """Organize images by UUID prefix for filesystem performance (first 2 characters)."""
+
+    def get_subfolder(self, image_name: str, image_category: ImageCategory, is_intermediate: bool) -> str:
+        return image_name[:2]
+
+
+def create_subfolder_strategy(strategy_name: str) -> ImageSubfolderStrategy:
+    """Factory function to create a subfolder strategy by name."""
+    strategies: dict[str, type[ImageSubfolderStrategy]] = {
+        "flat": FlatStrategy,
+        "date": DateStrategy,
+        "type": TypeStrategy,
+        "hash": HashStrategy,
+    }
+    cls = strategies.get(strategy_name)
+    if cls is None:
+        raise ValueError(f"Unknown subfolder strategy: {strategy_name}. Valid options: {', '.join(strategies.keys())}")
+    return cls()

invokeai/app/services/image_records/image_records_base.py

@@ -69,8 +69,8 @@ def delete_many(self, image_names: list[str]) -> None:
         pass
 
     @abstractmethod
-    def delete_intermediates(self) -> list[str]:
-        """Deletes all intermediate image records, returning a list of deleted image names."""
+    def delete_intermediates(self) -> list[tuple[str, str]]:
+        """Deletes all intermediate image records, returning a list of (image_name, image_subfolder) tuples."""
         pass
 
     @abstractmethod
@@ -93,6 +93,7 @@ def save(
         node_id: Optional[str] = None,
         metadata: Optional[str] = None,
         user_id: Optional[str] = None,
+        image_subfolder: str = "",
     ) -> datetime:
         """Saves an image record."""
         pass