invoke-ai · psychedelicious · Sep 23, 2025 · Sep 23, 2025 · Sep 23, 2025 · Sep 23, 2025
@@ -31,7 +31,10 @@
 from invokeai.backend.model_manager import BaseModelType, ModelFormat, ModelType
 from invokeai.backend.model_manager.config import (
     AnyModelConfig,
-    MainCheckpointConfig,
+    Main_Checkpoint_SD1_Config,
+    Main_Checkpoint_SD2_Config,
+    Main_Checkpoint_SDXL_Config,
+    Main_Checkpoint_SDXLRefiner_Config,
 )
 from invokeai.backend.model_manager.load.model_cache.cache_stats import CacheStats
 from invokeai.backend.model_manager.metadata.fetch.huggingface import HuggingFaceMetadataFetch
@@ -741,9 +744,18 @@ async def convert_model(
         logger.error(str(e))
         raise HTTPException(status_code=424, detail=str(e))
 
-    if not isinstance(model_config, MainCheckpointConfig):
-        logger.error(f"The model with key {key} is not a main checkpoint model.")
-        raise HTTPException(400, f"The model with key {key} is not a main checkpoint model.")
+    if isinstance(
+        model_config,
+        (
+            Main_Checkpoint_SD1_Config,
+            Main_Checkpoint_SD2_Config,
+            Main_Checkpoint_SDXL_Config,
+            Main_Checkpoint_SDXLRefiner_Config,
+        ),
+    ):
+        msg = f"The model with key {key} is not a main SD 1/2/XL checkpoint model."
+        logger.error(msg)
+        raise HTTPException(400, msg)
 
     with TemporaryDirectory(dir=ApiDependencies.invoker.services.configuration.models_path) as tmpdir:
         convert_path = pathlib.Path(tmpdir) / pathlib.Path(model_config.path).stem

@@ -21,7 +21,7 @@
 from invokeai.app.invocations.model import UNetField, VAEField
 from invokeai.app.services.shared.invocation_context import InvocationContext
 from invokeai.backend.model_manager import LoadedModel
-from invokeai.backend.model_manager.config import MainConfigBase
+from invokeai.backend.model_manager.config import Main_Config_Base
 from invokeai.backend.model_manager.taxonomy import ModelVariantType
 from invokeai.backend.stable_diffusion.diffusers_pipeline import image_resized_to_grid_as_tensor
 
@@ -182,7 +182,7 @@ def invoke(self, context: InvocationContext) -> GradientMaskOutput:
         if self.unet is not None and self.vae is not None and self.image is not None:
             # all three fields must be present at the same time
             main_model_config = context.models.get_config(self.unet.unet.key)
-            assert isinstance(main_model_config, MainConfigBase)
+            assert isinstance(main_model_config, Main_Config_Base)
             if main_model_config.variant is ModelVariantType.Inpaint:
                 mask = dilated_mask_tensor
                 vae_info: LoadedModel = context.models.load(self.vae.vae)

@@ -48,7 +48,7 @@
     unpack,
 )
 from invokeai.backend.flux.text_conditioning import FluxReduxConditioning, FluxTextConditioning
-from invokeai.backend.model_manager.taxonomy import ModelFormat, ModelVariantType
+from invokeai.backend.model_manager.taxonomy import FluxVariantType, ModelFormat
 from invokeai.backend.patches.layer_patcher import LayerPatcher
 from invokeai.backend.patches.lora_conversions.flux_lora_constants import FLUX_LORA_TRANSFORMER_PREFIX
 from invokeai.backend.patches.model_patch_raw import ModelPatchRaw
@@ -232,7 +232,7 @@ def _run_diffusion(
         )
 
         transformer_config = context.models.get_config(self.transformer.transformer)
-        is_schnell = "schnell" in getattr(transformer_config, "config_path", "")
+        is_schnell = transformer_config.variant is FluxVariantType.Schnell
 
         # Calculate the timestep schedule.
         timesteps = get_schedule(
@@ -277,7 +277,7 @@ def _run_diffusion(
 
         # Prepare the extra image conditioning tensor (img_cond) for either FLUX structural control or FLUX Fill.
         img_cond: torch.Tensor | None = None
-        is_flux_fill = transformer_config.variant == ModelVariantType.Inpaint  # type: ignore
+        is_flux_fill = transformer_config.variant is FluxVariantType.DevFill
         if is_flux_fill:
             img_cond = self._prep_flux_fill_img_cond(
                 context, device=TorchDevice.choose_torch_device(), dtype=inference_dtype

@@ -17,8 +17,7 @@
 from invokeai.app.invocations.util import validate_begin_end_step, validate_weights
 from invokeai.app.services.shared.invocation_context import InvocationContext
 from invokeai.backend.model_manager.config import (
-    IPAdapterCheckpointConfig,
-    IPAdapterInvokeAIConfig,
+    IPAdapter_Checkpoint_FLUX_Config,
 )
 from invokeai.backend.model_manager.taxonomy import BaseModelType, ModelType
 
@@ -68,7 +67,7 @@ def validate_begin_end_step_percent(self) -> Self:
     def invoke(self, context: InvocationContext) -> IPAdapterOutput:
         # Lookup the CLIP Vision encoder that is intended to be used with the IP-Adapter model.
         ip_adapter_info = context.models.get_config(self.ip_adapter_model.key)
-        assert isinstance(ip_adapter_info, (IPAdapterInvokeAIConfig, IPAdapterCheckpointConfig))
+        assert isinstance(ip_adapter_info, IPAdapter_Checkpoint_FLUX_Config)
 
         # Note: There is a IPAdapterInvokeAIConfig.image_encoder_model_id field, but it isn't trustworthy.
         image_encoder_starter_model = CLIP_VISION_MODEL_MAP[self.clip_vision_model]

@@ -13,9 +13,9 @@
     preprocess_t5_encoder_model_identifier,
     preprocess_t5_tokenizer_model_identifier,
 )
-from invokeai.backend.flux.util import max_seq_lengths
+from invokeai.backend.flux.util import get_flux_max_seq_length
 from invokeai.backend.model_manager.config import (
-    CheckpointConfigBase,
+    Checkpoint_Config_Base,
 )
 from invokeai.backend.model_manager.taxonomy import BaseModelType, ModelType, SubModelType
 
@@ -87,12 +87,12 @@ def invoke(self, context: InvocationContext) -> FluxModelLoaderOutput:
         t5_encoder = preprocess_t5_encoder_model_identifier(self.t5_encoder_model)
 
         transformer_config = context.models.get_config(transformer)
-        assert isinstance(transformer_config, CheckpointConfigBase)
+        assert isinstance(transformer_config, Checkpoint_Config_Base)
 
         return FluxModelLoaderOutput(
             transformer=TransformerField(transformer=transformer, loras=[]),
             clip=CLIPField(tokenizer=tokenizer, text_encoder=clip_encoder, loras=[], skipped_layers=0),
             t5_encoder=T5EncoderField(tokenizer=tokenizer2, text_encoder=t5_encoder, loras=[]),
             vae=VAEField(vae=vae),
-            max_seq_len=max_seq_lengths[transformer_config.config_path],
+            max_seq_len=get_flux_max_seq_length(transformer_config.variant),
         )
@@ -13,8 +13,8 @@
 from invokeai.app.services.shared.invocation_context import InvocationContext
 from invokeai.backend.model_manager.config import (
     AnyModelConfig,
-    IPAdapterCheckpointConfig,
-    IPAdapterInvokeAIConfig,
+    IPAdapter_Checkpoint_Config_Base,
+    IPAdapter_InvokeAI_Config_Base,
 )
 from invokeai.backend.model_manager.starter_models import (
     StarterModel,
@@ -123,9 +123,9 @@ def validate_begin_end_step_percent(self) -> Self:
     def invoke(self, context: InvocationContext) -> IPAdapterOutput:
         # Lookup the CLIP Vision encoder that is intended to be used with the IP-Adapter model.
         ip_adapter_info = context.models.get_config(self.ip_adapter_model.key)
-        assert isinstance(ip_adapter_info, (IPAdapterInvokeAIConfig, IPAdapterCheckpointConfig))
+        assert isinstance(ip_adapter_info, (IPAdapter_InvokeAI_Config_Base, IPAdapter_Checkpoint_Config_Base))
 
-        if isinstance(ip_adapter_info, IPAdapterInvokeAIConfig):
+        if isinstance(ip_adapter_info, IPAdapter_InvokeAI_Config_Base):
             image_encoder_model_id = ip_adapter_info.image_encoder_model_id
             image_encoder_model_name = image_encoder_model_id.split("/")[-1].strip()
         else:

@@ -24,8 +24,9 @@ class ModelIdentifierField(BaseModel):
     name: str = Field(description="The model's name")
     base: BaseModelType = Field(description="The model's base model type")
     type: ModelType = Field(description="The model's type")
-    submodel_type: Optional[SubModelType] = Field(
-        description="The submodel to load, if this is a main model", default=None
+    submodel_type: SubModelType | None = Field(
+        description="The submodel to load, if this is a main model",
+        default=None,
     )
 
     @classmethod

@@ -5,6 +5,7 @@
 import re
 import threading
 import time
+from copy import deepcopy
 from pathlib import Path
 from queue import Empty, Queue
 from shutil import move, rmtree
@@ -36,11 +37,10 @@
 from invokeai.app.services.model_records.model_records_base import ModelRecordChanges
 from invokeai.backend.model_manager.config import (
     AnyModelConfig,
-    CheckpointConfigBase,
+    Checkpoint_Config_Base,
     InvalidModelConfigException,
-    ModelConfigBase,
+    ModelConfigFactory,
 )
-from invokeai.backend.model_manager.legacy_probe import ModelProbe
 from invokeai.backend.model_manager.metadata import (
     AnyModelRepoMetadata,
     HuggingFaceMetadataFetch,
@@ -370,6 +370,8 @@ def unconditionally_delete(self, key: str) -> None:  # noqa D102
         model_path = self.app_config.models_path / model.path
         if model_path.is_file() or model_path.is_symlink():
             model_path.unlink()
+            assert model_path.parent != self.app_config.models_path
+            os.rmdir(model_path.parent)
         elif model_path.is_dir():
             rmtree(model_path)
         self.unregister(key)
@@ -598,18 +600,11 @@ def _probe(self, model_path: Path, config: Optional[ModelRecordChanges] = None):
         hash_algo = self._app_config.hashing_algorithm
         fields = config.model_dump()
 
-        # WARNING!
-        # The legacy probe relies on the implicit order of tests to determine model classification.
-        # This can lead to regressions between the legacy and new probes.
-        # Do NOT change the order of `probe` and `classify` without implementing one of the following fixes:
-        # Short-term fix: `classify` tests `matches` in the same order as the legacy probe.
-        # Long-term fix: Improve `matches` to be more specific so that only one config matches
-        #   any given model - eliminating ambiguity and removing reliance on order.
-        # After implementing either of these fixes, remove @pytest.mark.xfail from `test_regression_against_model_probe`
-        try:
-            return ModelProbe.probe(model_path=model_path, fields=fields, hash_algo=hash_algo)  # type: ignore
-        except InvalidModelConfigException:
-            return ModelConfigBase.classify(model_path, hash_algo, **fields)
+        return ModelConfigFactory.from_model_on_disk(
+            mod=model_path,
+            overrides=deepcopy(fields),
+            hash_algo=hash_algo,
+        )
 
     def _register(
         self, model_path: Path, config: Optional[ModelRecordChanges] = None, info: Optional[AnyModelConfig] = None
@@ -630,7 +625,7 @@ def _register(
 
         info.path = model_path.as_posix()
 
-        if isinstance(info, CheckpointConfigBase):
+        if isinstance(info, Checkpoint_Config_Base) and info.config_path is not None:
             # Checkpoints have a config file needed for conversion. Same handling as the model weights - if it's in the
             # invoke-managed legacy config dir, we use a relative path.
             legacy_config_path = self.app_config.legacy_conf_path / info.config_path

@@ -21,6 +21,7 @@
 from invokeai.backend.model_manager.taxonomy import (
     BaseModelType,
     ClipVariantType,
+    FluxVariantType,
     ModelFormat,
     ModelSourceType,
     ModelType,
@@ -90,7 +91,9 @@ class ModelRecordChanges(BaseModelExcludeNull):
 
     # Checkpoint-specific changes
     # TODO(MM2): Should we expose these? Feels footgun-y...
-    variant: Optional[ModelVariantType | ClipVariantType] = Field(description="The variant of the model.", default=None)
+    variant: Optional[ModelVariantType | ClipVariantType | FluxVariantType] = Field(
+        description="The variant of the model.", default=None
+    )
     prediction_type: Optional[SchedulerPredictionType] = Field(
         description="The prediction type of the model.", default=None
     )

@@ -141,10 +141,25 @@ def update_model(self, key: str, changes: ModelRecordChanges) -> AnyModelConfig:
         with self._db.transaction() as cursor:
             record = self.get_model(key)
 
-            # Model configs use pydantic's `validate_assignment`, so each change is validated by pydantic.
+            # The changes may mean the model config class changes. So we need to:
+            #
+            # 1. convert the existing record to a dict
+            # 2. apply the changes to the dict
+            # 3. create a new model config from the updated dict
+            #
+            # This way we ensure that the update does not inadvertently create an invalid model config.
+
+            # 1. convert the existing record to a dict
+            record_as_dict = record.model_dump()
+
+            # 2. apply the changes to the dict
             for field_name in changes.model_fields_set:
-                setattr(record, field_name, getattr(changes, field_name))
+                record_as_dict[field_name] = getattr(changes, field_name)
 
+            # 3. create a new model config from the updated dict
+            record = ModelConfigFactory.make_config(record_as_dict)
+
+            # If we get this far, the updated model config is valid, so we can save it to the database.
             json_serialized = record.model_dump_json()
 
             cursor.execute(
@@ -277,14 +292,19 @@ def search_by_attr(
         for row in result:
             try:
                 model_config = ModelConfigFactory.make_config(json.loads(row[0]), timestamp=row[1])
-            except pydantic.ValidationError:
+            except pydantic.ValidationError as e:
                 # We catch this error so that the app can still run if there are invalid model configs in the database.
                 # One reason that an invalid model config might be in the database is if someone had to rollback from a
                 # newer version of the app that added a new model type.
                 row_data = f"{row[0][:64]}..." if len(row[0]) > 64 else row[0]
+                try:
+                    name = json.loads(row[0]).get("name", "<unknown>")
+                except Exception:
+                    name = "<unknown>"
                 self._logger.warning(
-                    f"Found an invalid model config in the database. Ignoring this model. ({row_data})"
+                    f"Skipping invalid model config in the database with name {name}. Ignoring this model. ({row_data})"
                 )
+                self._logger.warning(f"Validation error: {e}")
             else:
                 results.append(model_config)
 

@@ -21,7 +21,7 @@
 from invokeai.app.util.step_callback import diffusion_step_callback
 from invokeai.backend.model_manager.config import (
     AnyModelConfig,
-    ModelConfigBase,
+    Config_Base,
 )
 from invokeai.backend.model_manager.load.load_base import LoadedModel, LoadedModelWithoutConfig
 from invokeai.backend.model_manager.taxonomy import AnyModel, BaseModelType, ModelFormat, ModelType, SubModelType
@@ -558,7 +558,7 @@ def get_absolute_path(self, config_or_path: AnyModelConfig | Path | str) -> Path
             The absolute path to the model.
         """
 
-        model_path = Path(config_or_path.path) if isinstance(config_or_path, ModelConfigBase) else Path(config_or_path)
+        model_path = Path(config_or_path.path) if isinstance(config_or_path, Config_Base) else Path(config_or_path)
 
         if model_path.is_absolute():
             return model_path.resolve()

@@ -8,7 +8,7 @@
 
 from invokeai.app.services.config import InvokeAIAppConfig
 from invokeai.app.services.shared.sqlite_migrator.sqlite_migrator_common import Migration
-from invokeai.backend.model_manager.config import AnyModelConfig, AnyModelConfigValidator
+from invokeai.backend.model_manager.config import AnyModelConfigValidator
 
 
 class NormalizeResult(NamedTuple):
@@ -30,7 +30,7 @@ def __call__(self, cursor: sqlite3.Cursor) -> None:
         for model_id, config_json in rows:
             try:
                 # Get the model config as a pydantic object
-                config = self._load_model_config(config_json)
+                config = AnyModelConfigValidator.validate_json(config_json)
             except ValidationError:
                 # This could happen if the config schema changed in a way that makes old configs invalid. Unlikely
                 # for users, more likely for devs testing out migration paths.
@@ -216,11 +216,6 @@ def _prune_empty_directories(self) -> None:
 
         self._logger.info("Pruned %d empty directories under %s", len(removed_dirs), self._models_dir)
 
-    def _load_model_config(self, config_json: str) -> AnyModelConfig:
-        # The typing of the validator says it returns Unknown, but it's really a AnyModelConfig. This utility function
-        # just makes that clear.
-        return AnyModelConfigValidator.validate_json(config_json)
-
 
 def build_migration_22(app_config: InvokeAIAppConfig, logger: Logger) -> Migration:
     """Builds the migration object for migrating from version 21 to version 22.

@@ -12,6 +12,7 @@
 from invokeai.app.invocations.model import ModelIdentifierField
 from invokeai.app.services.events.events_common import EventBase
 from invokeai.app.services.session_processor.session_processor_common import ProgressImage
+from invokeai.backend.model_manager.config import AnyModelConfigValidator
 from invokeai.backend.util.logging import InvokeAILogger
 
 logger = InvokeAILogger.get_logger()
@@ -115,6 +116,13 @@ def openapi() -> dict[str, Any]:
         # additional_schemas[1] is a dict of $defs that we need to add to the top level of the schema
         move_defs_to_top_level(openapi_schema, additional_schemas[1])
 
+        any_model_config_schema = AnyModelConfigValidator.json_schema(
+            mode="serialization",
+            ref_template="#/components/schemas/{model}",
+        )
+        move_defs_to_top_level(openapi_schema, any_model_config_schema)
+        openapi_schema["components"]["schemas"]["AnyModelConfig"] = any_model_config_schema
+
         if post_transform is not None:
             openapi_schema = post_transform(openapi_schema)
 

@@ -5,7 +5,7 @@
 from invokeai.backend.flux.model import FluxParams
 
 
-def is_state_dict_xlabs_controlnet(sd: Dict[str, Any]) -> bool:
+def is_state_dict_xlabs_controlnet(sd: dict[str | int, Any]) -> bool:
     """Is the state dict for an XLabs ControlNet model?
 
     This is intended to be a reasonably high-precision detector, but it is not guaranteed to have perfect precision.
@@ -25,7 +25,7 @@ def is_state_dict_xlabs_controlnet(sd: Dict[str, Any]) -> bool:
     return False
 
 
-def is_state_dict_instantx_controlnet(sd: Dict[str, Any]) -> bool:
+def is_state_dict_instantx_controlnet(sd: dict[str | int, Any]) -> bool:
     """Is the state dict for an InstantX ControlNet model?
 
     This is intended to be a reasonably high-precision detector, but it is not guaranteed to have perfect precision.