Merge pull request #1124 from willmiao/feature/centralized-sidecar-storage

feat: optional centralized storage for sidecar metadata and previews (#1045)

py/config.py

@@ -891,6 +891,17 @@ class Config:
         if self.recipes_path:
             preview_roots.update(self._expand_preview_root(self.recipes_path))
 
+        # Centralized sidecar storage holds preview assets outside the model
+        # roots; allow serving them when the mode is active.
+        try:
+            from .utils.sidecar_paths import get_sidecar_root  # Local import to avoid circular dependency
+
+            sidecar_root = get_sidecar_root()
+        except Exception:  # pragma: no cover - defensive fallback
+            sidecar_root = ""
+        if sidecar_root:
+            preview_roots.update(self._expand_preview_root(sidecar_root))
+
         for target, link in self._path_mappings.items():
             preview_roots.update(self._expand_preview_root(target))
             preview_roots.update(self._expand_preview_root(link))
@@ -1494,6 +1505,15 @@ class Config:
         self.other_roots = self._init_other_paths()
         self._rebuild_preview_roots()
 
+    def refresh_preview_roots(self) -> None:
+        """Rebuild the preview allowlist after path-affecting settings change.
+
+        Called when ``sidecar_storage_mode`` / ``sidecar_storage_path`` are
+        updated so centralized preview assets become servable (or stop being
+        servable) without a restart.
+        """
+        self._rebuild_preview_roots()
+
     def get_other_models_availability(self) -> Dict[str, Any]:
         """Report the other-model folders the host can actually expose.
 

py/metadata_ops/__init__.py

@@ -172,12 +172,16 @@ async def download_preview(
     """
     from ..services.downloader import get_downloader
     from ..utils.exif_utils import ExifUtils
+    from ..utils.sidecar_paths import get_preview_dir
 
     if not url or not url.strip():
         return None
 
     base_name = os.path.splitext(os.path.basename(model_path))[0]
-    preview_dir = os.path.dirname(model_path)
+    preview_dir = get_preview_dir(model_path)
+    # Centralized mirrors may not exist yet (unlike the model's own directory
+    # in alongside mode).
+    os.makedirs(preview_dir, exist_ok=True)
     output_path = os.path.join(preview_dir, base_name + ".webp")
 
     downloader = await get_downloader()

py/routes/handlers/misc_handlers.py

@@ -45,6 +45,8 @@ from ...services.llm_service import (
     get_provider_model_ids,
 )
 from ...services.cache_health_monitor import CacheHealthMonitor, CacheHealthStatus
+from ...services.use_cases.sidecar_migration_use_case import SidecarMigrationUseCase
+from ...services.websocket_progress_callback import WebSocketBroadcastCallback
 from ...utils.models import BaseModelMetadata
 from ...utils.constants import (
     CIVITAI_USER_MODEL_TYPES,
@@ -68,6 +70,7 @@ from ...utils.example_images_paths import (
 )
 from ...utils.lora_metadata import extract_trained_words
 from ...utils.session_logging import get_standalone_session_log_snapshot
+from ...utils.sidecar_paths import get_metadata_path, get_preview_dir
 from ...utils.usage_stats import UsageStats
 from .base_model_handlers import BaseModelHandlerSet
 
@@ -943,15 +946,24 @@ class DoctorHandler:
 
                             os.rename(path, new_path)
 
-                            for suffix in (".metadata.json", ".civitai.info"):
-                                old_sidecar = old_base_no_ext + suffix
-                                new_sidecar = new_base_no_ext + suffix
-                                if os.path.exists(old_sidecar):
-                                    os.rename(old_sidecar, new_sidecar)
+                            old_metadata_path = get_metadata_path(path)
+                            new_metadata_path = get_metadata_path(new_path)
+                            if os.path.exists(old_metadata_path):
+                                os.rename(old_metadata_path, new_metadata_path)
+
+                            old_sidecar = old_base_no_ext + ".civitai.info"
+                            new_sidecar = new_base_no_ext + ".civitai.info"
+                            if os.path.exists(old_sidecar):
+                                os.rename(old_sidecar, new_sidecar)
 
                             for preview_ext in PREVIEW_EXTENSIONS:
-                                old_preview = old_base_no_ext + preview_ext
-                                new_preview = new_base_no_ext + preview_ext
+                                old_preview = os.path.join(
+                                    get_preview_dir(path), base_name + preview_ext
+                                )
+                                new_preview = os.path.join(
+                                    get_preview_dir(new_path),
+                                    candidate_base + preview_ext,
+                                )
                                 if os.path.exists(old_preview):
                                     os.rename(old_preview, new_preview)
 
@@ -963,7 +975,10 @@ class DoctorHandler:
                                     old_preview_url = entry["preview_url"].replace("\\", "/")
                                     preview_ext = os.path.splitext(old_preview_url)[1]
                                     if preview_ext:
-                                        entry["preview_url"] = (new_base_no_ext + preview_ext).replace(os.sep, "/")
+                                        entry["preview_url"] = os.path.join(
+                                            get_preview_dir(new_path),
+                                            candidate_base + preview_ext,
+                                        ).replace(os.sep, "/")
                                 await scanner.update_single_model_cache(
                                     path, new_path, entry
                                 )
@@ -4123,6 +4138,64 @@ class NodeRegistryHandler:
             return web.json_response({"success": False, "error": str(exc)}, status=500)
 
 
+class SidecarMigrationHandler:
+    """Migrate sidecar metadata and previews between storage layouts."""
+
+    _VALID_DIRECTIONS = ("to_centralized", "to_alongside", "relocate_root")
+
+    def __init__(
+        self,
+        *,
+        use_case_factory: Callable[[], SidecarMigrationUseCase] = SidecarMigrationUseCase,
+        progress_callback_factory: Callable[[], Any] = WebSocketBroadcastCallback,
+    ) -> None:
+        self._use_case_factory = use_case_factory
+        self._progress_callback_factory = progress_callback_factory
+
+    async def migrate_sidecars(self, request: web.Request) -> web.Response:
+        """Run a sidecar migration; accepts POST JSON or GET query params."""
+        try:
+            if request.method == "GET":
+                params: Mapping[str, Any] = request.query
+            else:
+                try:
+                    params = await request.json()
+                except Exception:  # empty/invalid body: fall back to query
+                    params = request.query
+
+            direction = str(params.get("direction") or "").strip()
+            if direction not in self._VALID_DIRECTIONS:
+                return web.json_response(
+                    {
+                        "success": False,
+                        "error": "direction must be 'to_centralized', 'to_alongside' or 'relocate_root'",
+                    },
+                    status=400,
+                )
+
+            force = params.get("force") in (True, 1, "true", "1")
+            old_root = str(params.get("old_root") or "").strip()
+            if direction == "relocate_root" and not old_root:
+                return web.json_response(
+                    {"success": False, "error": "old_root is required for relocate_root"},
+                    status=400,
+                )
+
+            use_case = self._use_case_factory()
+            progress_cb = self._progress_callback_factory()
+            result = await use_case.execute_with_error_handling(
+                direction=direction,
+                progress_cb=progress_cb,
+                force=force,
+                old_root=old_root,
+            )
+            status = 200 if result.get("success") else 400
+            return web.json_response(result, status=status)
+        except Exception as exc:
+            logger.error("Sidecar migration failed: %s", exc, exc_info=True)
+            return web.json_response({"success": False, "error": str(exc)}, status=500)
+
+
 class MiscHandlerSet:
     """Aggregate handlers into a lookup compatible with the registrar."""
 
@@ -4149,6 +4222,7 @@ class MiscHandlerSet:
         model_source_handler: Any = None,
         agent_handler: Any = None,
         download_routing: Any = None,
+        sidecar_migration: Any = None,
     ) -> None:
         self.health = health
         self.settings = settings
@@ -4170,6 +4244,7 @@ class MiscHandlerSet:
         self.model_source_handler = model_source_handler
         self.agent_handler = agent_handler
         self.download_routing = download_routing
+        self.sidecar_migration = sidecar_migration
 
     def to_route_mapping(
         self,
@@ -4236,6 +4311,8 @@ class MiscHandlerSet:
             "cancel_agent_skill": self.agent_handler.cancel_agent_skill,
             # Download routing handler
             "get_download_routing": self.download_routing.get_download_routing,
+            # Sidecar migration handler
+            "migrate_sidecars": self.sidecar_migration.migrate_sidecars,
             # Base model handlers
             "get_base_models": self.base_model.get_base_models,
             "refresh_base_models": self.base_model.refresh_base_models,

py/routes/handlers/model_handlers.py

@@ -50,6 +50,7 @@ from ...services.errors import RateLimitError, ResourceNotFoundError
 from ...utils.civitai_utils import resolve_license_payload
 from ...utils.file_utils import calculate_sha256
 from ...utils.metadata_manager import MetadataManager
+from ...utils.sidecar_paths import get_metadata_path
 from ...utils.url_utils import relative_root_prefix
 
 LICENSE_FIELDS = (
@@ -676,7 +677,7 @@ class ModelManagementHandler:
                     status=400,
                 )
 
-            metadata_path = os.path.splitext(file_path)[0] + ".metadata.json"
+            metadata_path = get_metadata_path(file_path)
             local_metadata = await self._metadata_sync.load_local_metadata(
                 metadata_path
             )

py/routes/misc_route_registrar.py

@@ -113,6 +113,13 @@ MISC_ROUTE_DEFINITIONS: tuple[RouteDefinition, ...] = (
     RouteDefinition(
         "POST", "/api/lm/download/routing", "get_download_routing"
     ),
+    # Sidecar storage layout migration (GET supported for the extension)
+    RouteDefinition(
+        "POST", "/api/lm/sidecars/migrate", "migrate_sidecars"
+    ),
+    RouteDefinition(
+        "GET", "/api/lm/sidecars/migrate", "migrate_sidecars"
+    ),
     RouteDefinition(
         "POST", "/api/lm/download-model-source", "download_model_source"
     ),

py/routes/misc_routes.py

@@ -32,6 +32,7 @@ from .handlers.misc_handlers import (
     NodeRegistry,
     NodeRegistryHandler,
     SettingsHandler,
+    SidecarMigrationHandler,
     SupportersHandler,
     TrainedWordsHandler,
     UsageStatsHandler,
@@ -142,6 +143,7 @@ class MiscRoutes:
         model_source_handler = ModelSourceHandler()
         agent_handler = AgentHandler()
         download_routing = DownloadRoutingHandler()
+        sidecar_migration = SidecarMigrationHandler()
 
         return self._handler_set_factory(
             health=health,
@@ -164,6 +166,7 @@ class MiscRoutes:
             model_source_handler=model_source_handler,
             agent_handler=agent_handler,
             download_routing=download_routing,
+            sidecar_migration=sidecar_migration,
         )
 
 

py/services/autov3_backfill_service.py

@@ -27,6 +27,8 @@ import os
 import threading
 from typing import TYPE_CHECKING, Optional
 
+from ..utils.sidecar_paths import get_metadata_path
+
 if TYPE_CHECKING:  # pragma: no cover - type-check only; runtime imports are local
     from .model_scanner import ModelScanner
 
@@ -41,7 +43,7 @@ def _resolve_autov3(file_path: str) -> str:
     safetensors header hash. Returns ``''`` when neither is available.
     """
     try:
-        metadata_path = f"{os.path.splitext(file_path)[0]}.metadata.json"
+        metadata_path = get_metadata_path(file_path)
         if os.path.exists(metadata_path):
             with open(metadata_path, "r", encoding="utf-8") as handle:
                 payload = json.load(handle)

py/services/checkpoint_scanner.py

@@ -12,6 +12,7 @@ from typing import Any, Dict, List, Optional
 from ..utils.models import CheckpointMetadata
 from ..utils.file_utils import find_preview_file, normalize_path, calculate_autov3
 from ..utils.metadata_manager import MetadataManager
+from ..utils.sidecar_paths import get_preview_dir, is_centralized
 from ..config import config
 from .model_scanner import ModelScanner, _is_excluded_dir
 from .model_hash_index import ModelHashIndex
@@ -61,10 +62,9 @@ class CheckpointScanner(ModelScanner):
                 return None
 
             base_name = os.path.splitext(os.path.basename(file_path))[0]
-            dir_path = os.path.dirname(file_path)
 
             # Find preview image
-            preview_url = find_preview_file(base_name, dir_path)
+            preview_url = find_preview_file(base_name, get_preview_dir(file_path))
 
             # AutoV3 reads only the safetensors header, so it is cheap even for
             # large checkpoints; record the checked state at creation time ("" =
@@ -322,6 +322,11 @@ class CheckpointScanner(ModelScanner):
 
     async def _find_pending_models_from_filesystem(self) -> List[Dict[str, Any]]:
         """Scan filesystem for checkpoint metadata files with pending hash status."""
+        # Centralized mode stores sidecars in the mirror tree, not next to the
+        # models; walk the mirror instead of the model folders.
+        if is_centralized():
+            return self._find_pending_models_in_sidecar_mirror()
+
         pending_models = []
 
         for root_path in self.get_model_roots():

py/services/download_manager.py

@@ -38,6 +38,7 @@ from ..utils.preview_selection import resolve_mature_threshold, select_preview_m
 from ..utils.utils import calculate_filename_for_model, sanitize_folder_name
 from ..utils.exif_utils import ExifUtils
 from ..utils.metadata_manager import MetadataManager
+from ..utils.sidecar_paths import get_metadata_path, get_preview_dir
 from .service_registry import ServiceRegistry
 from .download_routing import is_diffusion_model_download, resolve_other_download_sub_type
 from .settings_manager import get_settings_manager
@@ -831,7 +832,7 @@ class DownloadManager:
                 )
 
         for file_path in target_files:
-            metadata_path = os.path.splitext(file_path)[0] + ".metadata.json"
+            metadata_path = get_metadata_path(file_path)
             deleted = await self._delete_file_with_retries(metadata_path)
             if not deleted and os.path.exists(metadata_path):
                 logger.error(f"Error deleting metadata file: {metadata_path}")
@@ -2447,7 +2448,7 @@ class DownloadManager:
                 return {"success": False, "error": save_path}
 
             part_path = save_path + ".part"
-            metadata_path = os.path.splitext(save_path)[0] + ".metadata.json"
+            metadata_path = get_metadata_path(save_path)
 
             pause_control = self._pause_events.get(download_id) if download_id else None
 
@@ -2464,6 +2465,10 @@ class DownloadManager:
             # Download preview image if available
             images = version_info.get("images", [])
             if images:
+                # Centralized preview mirrors may not exist yet (unlike the
+                # model's own directory in alongside mode).
+                os.makedirs(get_preview_dir(save_path), exist_ok=True)
+
                 if progress_callback:
                     await progress_callback(
                         1
@@ -2503,7 +2508,10 @@ class DownloadManager:
 
                     if media_type == "video":
                         preview_ext = _extension_from_url(preview_url, ".mp4")
-                        preview_path = os.path.splitext(save_path)[0] + preview_ext
+                        preview_path = os.path.join(
+                            get_preview_dir(save_path),
+                            os.path.splitext(os.path.basename(save_path))[0] + preview_ext,
+                        )
                         rewritten_url, rewritten = rewrite_preview_url(
                             preview_url, media_type="video"
                         )
@@ -2530,7 +2538,10 @@ class DownloadManager:
                         )
                         if rewritten and rewritten_url:
                             preview_ext = _extension_from_url(preview_url, ".png")
-                            preview_path = os.path.splitext(save_path)[0] + preview_ext
+                            preview_path = os.path.join(
+                                get_preview_dir(save_path),
+                                os.path.splitext(os.path.basename(save_path))[0] + preview_ext,
+                            )
                             success, _ = await downloader.download_file(
                                 rewritten_url, preview_path, use_auth=False
                             )
@@ -2557,8 +2568,9 @@ class DownloadManager:
                                         temp_file_handle.write(
                                             content if isinstance(content, bytes) else content.encode("utf-8")
                                         )
-                                    preview_path = (
-                                        os.path.splitext(save_path)[0] + ".webp"
+                                    preview_path = os.path.join(
+                                        get_preview_dir(save_path),
+                                        os.path.splitext(os.path.basename(save_path))[0] + ".webp",
                                     )
 
                                     optimized_data, _ = ExifUtils.optimize_image(
@@ -2788,9 +2800,7 @@ class DownloadManager:
                             entry = cast(Any, adjusted_entry)
                             metadata_entries[index] = entry
 
-                metadata_file_path = (
-                    os.path.splitext(entry.file_path)[0] + ".metadata.json"
-                )
+                metadata_file_path = get_metadata_path(entry.file_path)
                 metadata_files_for_cleanup.append(metadata_file_path)
 
                 await MetadataManager.save_metadata(entry.file_path, entry)
@@ -3049,7 +3059,11 @@ class DownloadManager:
         extension = os.path.splitext(preview_path)[1] or ".webp"
 
         targets = [
-            os.path.splitext(entry.file_path)[0] + extension for entry in entries
+            os.path.join(
+                get_preview_dir(entry.file_path),
+                os.path.splitext(os.path.basename(entry.file_path))[0] + extension,
+            )
+            for entry in entries
         ]
 
         if not targets:
@@ -3057,10 +3071,12 @@ class DownloadManager:
 
         first_target = targets[0]
         if preview_path != first_target:
+            os.makedirs(os.path.dirname(first_target), exist_ok=True)
             os.replace(preview_path, first_target)
         source_path = first_target
 
         for target in targets[1:]:
+            os.makedirs(os.path.dirname(target), exist_ok=True)
             shutil.copyfile(source_path, target)
 
         return targets

py/services/metadata_sync_service.py

@@ -12,6 +12,7 @@ from ..services.settings_manager import SettingsManager
 from ..utils.civitai_utils import resolve_license_payload
 from ..utils.model_utils import determine_base_model
 from ..utils.models import autov3_from_civitai_files
+from ..utils.sidecar_paths import get_metadata_path
 from .connectivity_guard import OFFLINE_FRIENDLY_MESSAGE, is_expected_offline_error
 from .errors import RateLimitError
 from .model_sources import has_external_source
@@ -216,7 +217,7 @@ class MetadataSyncService:
             logger.error(error)
             return False, error
 
-        metadata_path = os.path.splitext(file_path)[0] + ".metadata.json"
+        metadata_path = get_metadata_path(file_path)
         enable_archive = self._settings.get("enable_metadata_archive_db", False)
         previous_source = model_data.get("metadata_source") or (model_data.get("civitai") or {}).get("source")
 
@@ -485,7 +486,7 @@ class MetadataSyncService:
                 + (f" with version: {model_version_id}" if model_version_id else "")
             )
 
-        metadata_path = os.path.splitext(file_path)[0] + ".metadata.json"
+        metadata_path = get_metadata_path(file_path)
         await self.update_model_metadata(
             metadata_path,
             metadata,
@@ -505,7 +506,7 @@ class MetadataSyncService:
     ) -> Dict[str, Any]:
         """Apply metadata updates and persist to disk and cache."""
 
-        metadata_path = os.path.splitext(file_path)[0] + ".metadata.json"
+        metadata_path = get_metadata_path(file_path)
         metadata = await metadata_loader(metadata_path)
 
         for key, value in updates.items():
@@ -554,7 +555,7 @@ class MetadataSyncService:
         }
 
         expected_hash: Optional[str] = None
-        first_metadata_path = os.path.splitext(file_paths[0])[0] + ".metadata.json"
+        first_metadata_path = get_metadata_path(file_paths[0])
         first_metadata = await metadata_loader(first_metadata_path)
         if first_metadata and "sha256" in first_metadata:
             expected_hash = first_metadata["sha256"].lower()
@@ -565,7 +566,7 @@ class MetadataSyncService:
 
             try:
                 actual_hash = await hash_calculator(path)
-                metadata_path = os.path.splitext(path)[0] + ".metadata.json"
+                metadata_path = get_metadata_path(path)
                 metadata = await metadata_loader(metadata_path)
                 stored_hash = metadata.get("sha256", "").lower()
 

py/services/model_file_service.py

@@ -8,6 +8,7 @@ from abc import ABC, abstractmethod
 
 from ..utils.utils import calculate_relative_path_for_model, remove_empty_dirs
 from ..utils.constants import AUTO_ORGANIZE_BATCH_SIZE, MODEL_FILE_EXTENSIONS
+from ..utils.sidecar_paths import is_centralized, resolve_centralized_dir_for_dir
 from ..services.settings_manager import get_settings_manager
 from ..services.model_lifecycle_service import _require_path_in_library_roots
 from ..services.pending_delete_service import PENDING_DELETE_DIR_NAME
@@ -663,6 +664,21 @@ class ModelMoveService:
 
             shutil.rmtree(absolute_path)
 
+            # Centralized mode: prune the folder's mirror subtree when it no
+            # longer holds any sidecar files (per-model deletes already
+            # removed their sidecars, so only empty directories are expected;
+            # a non-empty mirror keeps its orphan sidecars).
+            if is_centralized():
+                mirror_dir = resolve_centralized_dir_for_dir(absolute_path)
+                if mirror_dir and os.path.isdir(mirror_dir):
+                    for root, _dirs, files in os.walk(mirror_dir, topdown=False):
+                        if files:
+                            continue
+                        try:
+                            os.rmdir(root)
+                        except OSError:  # pragma: no cover - best-effort cleanup
+                            pass
+
             await self._forget_folder(relative_folder)
 
             return {

py/services/model_lifecycle_service.py

@@ -11,6 +11,7 @@ from ..services.service_registry import ServiceRegistry
 from ..services.pending_delete_service import get_pending_delete_service
 from ..utils.constants import PREVIEW_EXTENSIONS
 from ..utils.metadata_manager import MetadataManager
+from ..utils.sidecar_paths import get_metadata_path, get_preview_dir, get_sidecar_dir
 
 logger = logging.getLogger(__name__)
 
@@ -41,16 +42,22 @@ async def load_local_metadata(metadata_path: str) -> Dict[str, Any]:
 async def delete_model_artifacts(
     target_dir: str, file_name: str, main_extension: str | None = None
 ) -> List[str]:
-    """Delete the primary model artefacts within ``target_dir``."""
+    """Delete the primary model artefacts within ``target_dir``.
+
+    Sidecars and previews are taken from the model's sidecar directory — the
+    model's own directory in alongside mode, the centralized mirror otherwise.
+    """
 
     main_extension = ".safetensors" if main_extension is None else main_extension
     main_file = f"{file_name}{main_extension}" if main_extension else file_name
-    patterns = [main_file, f"{file_name}.metadata.json"]
+    model_path = os.path.join(target_dir, main_file)
+    sidecar_dir = get_sidecar_dir(model_path)
+    patterns = [os.path.basename(get_metadata_path(model_path))]
     for ext in PREVIEW_EXTENSIONS:
         patterns.append(f"{file_name}{ext}")
 
     deleted: List[str] = []
-    main_path = os.path.join(target_dir, main_file).replace(os.sep, "/")
+    main_path = model_path.replace(os.sep, "/")
 
     if os.path.exists(main_path):
         os.remove(main_path)
@@ -58,8 +65,8 @@ async def delete_model_artifacts(
     else:
         logger.warning("Model file not found: %s", main_file)
 
-    for pattern in patterns[1:]:
-        path = os.path.join(target_dir, pattern)
+    for pattern in patterns:
+        path = os.path.join(sidecar_dir, pattern)
         if os.path.exists(path):
             try:
                 os.remove(path)
@@ -260,7 +267,7 @@ class ModelLifecycleService:
 
         _require_path_in_library_roots(file_path, self._scanner, label="File path")
 
-        metadata_path = os.path.splitext(file_path)[0] + ".metadata.json"
+        metadata_path = get_metadata_path(file_path)
         metadata = await self._metadata_loader(metadata_path)
         metadata["exclude"] = True
 
@@ -315,7 +322,7 @@ class ModelLifecycleService:
         if not os.path.exists(file_path):
             raise ValueError("Model file does not exist")
 
-        metadata_path = os.path.splitext(file_path)[0] + ".metadata.json"
+        metadata_path = get_metadata_path(file_path)
         metadata_payload = await self._metadata_loader(metadata_path)
         metadata_payload["exclude"] = False
 
@@ -384,21 +391,26 @@ class ModelLifecycleService:
         if os.path.exists(new_file_path):
             raise ValueError("A file with this name already exists")
 
-        patterns = [
-            f"{old_file_name}{old_extension}",
-            f"{old_file_name}.metadata.json",
-            f"{old_file_name}.metadata.json.bak",
+        metadata_filename = os.path.basename(get_metadata_path(file_path))
+        # Sidecars/previews live in the sidecar dir (the model's own dir in
+        # alongside mode, the centralized mirror otherwise); the model file
+        # itself always stays in target_dir.
+        sidecar_dir = get_sidecar_dir(file_path)
+        patterns: List[tuple[str, str]] = [
+            (target_dir, f"{old_file_name}{old_extension}"),
+            (sidecar_dir, metadata_filename),
+            (sidecar_dir, f"{metadata_filename}.bak"),
         ]
         for ext in PREVIEW_EXTENSIONS:
-            patterns.append(f"{old_file_name}{ext}")
+            patterns.append((sidecar_dir, f"{old_file_name}{ext}"))
 
         existing_files: List[tuple[str, str]] = []
-        for pattern in patterns:
-            path = os.path.join(target_dir, pattern)
+        for pattern_dir, pattern in patterns:
+            path = os.path.join(pattern_dir, pattern)
             if os.path.exists(path):
                 existing_files.append((path, pattern))
 
-        metadata_path = os.path.join(target_dir, f"{old_file_name}.metadata.json")
+        metadata_path = get_metadata_path(file_path)
         metadata: Optional[Dict[str, object]] = None
         hash_value: Optional[str] = None
 
@@ -413,9 +425,9 @@ class ModelLifecycleService:
 
         for old_path, pattern in existing_files:
             ext = self._get_multipart_ext(pattern)
-            new_path = os.path.join(target_dir, f"{new_file_name}{ext}").replace(
-                os.sep, "/"
-            )
+            new_path = os.path.join(
+                os.path.dirname(old_path), f"{new_file_name}{ext}"
+            ).replace(os.sep, "/")
             os.rename(old_path, new_path)
             renamed_files.append(new_path)
 
@@ -432,9 +444,9 @@ class ModelLifecycleService:
             if metadata.get("preview_url"):
                 old_preview = str(metadata["preview_url"])
                 ext = self._get_multipart_ext(old_preview)
-                new_preview = os.path.join(target_dir, f"{new_file_name}{ext}").replace(
-                    os.sep, "/"
-                )
+                new_preview = os.path.join(
+                    get_preview_dir(new_file_path), f"{new_file_name}{ext}"
+                ).replace(os.sep, "/")
                 metadata["preview_url"] = new_preview
 
             await self._metadata_manager.save_metadata(new_file_path, metadata)

py/services/model_scanner.py

@@ -11,6 +11,13 @@ from ..utils.models import BaseModelMetadata, autov3_from_civitai_files
 from ..config import config
 from ..utils.file_utils import find_preview_file, get_preview_extension, calculate_sha256, calculate_autov3
 from ..utils.metadata_manager import MetadataManager
+from ..utils.sidecar_paths import (
+    get_metadata_path,
+    get_preview_dir,
+    get_sidecar_dir,
+    is_centralized,
+    resolve_centralized_dir_for_dir,
+)
 from ..utils.civitai_utils import resolve_license_info
 from .model_cache import ModelCache
 from .model_hash_index import ModelHashIndex
@@ -1613,6 +1620,25 @@ class ModelScanner:
         old_abs_prefix = f"{str(previous_path).replace(chr(92), '/').rstrip('/')}/"
         new_abs_prefix = f"{str(new_path).replace(chr(92), '/').rstrip('/')}/"
 
+        # Centralized sidecar mode: sidecars/previews live in the mirror tree,
+        # not under the renamed model directory, so the mirror subtree must
+        # move too and mirror-prefixed preview URLs need their own rekey.
+        old_mirror_dir: Optional[str] = None
+        new_mirror_dir: Optional[str] = None
+        if is_centralized():
+            old_mirror_dir = resolve_centralized_dir_for_dir(str(previous_path))
+            new_mirror_dir = resolve_centralized_dir_for_dir(str(new_path))
+        old_mirror_prefix = (
+            f"{old_mirror_dir.replace(chr(92), '/').rstrip('/')}/"
+            if old_mirror_dir
+            else ""
+        )
+        new_mirror_prefix = (
+            f"{new_mirror_dir.replace(chr(92), '/').rstrip('/')}/"
+            if new_mirror_dir
+            else ""
+        )
+
         cache = self._cache
         if cache is None:
             return False
@@ -1670,8 +1696,24 @@ class ModelScanner:
                 item["preview_url"] = self._rekey_path(
                     item["preview_url"], old_abs_prefix, new_abs_prefix
                 )
+                if old_mirror_prefix:
+                    item["preview_url"] = self._rekey_path(
+                        item["preview_url"], old_mirror_prefix, new_mirror_prefix
+                    )
             touched.append(item)
 
+        if old_mirror_dir and new_mirror_dir and os.path.isdir(old_mirror_dir):
+            try:
+                os.makedirs(os.path.dirname(new_mirror_dir), exist_ok=True)
+                shutil.move(old_mirror_dir, new_mirror_dir)
+            except Exception as exc:  # pragma: no cover - defensive
+                logger.warning(
+                    "Failed to move centralized sidecar mirror %s -> %s: %s",
+                    old_mirror_dir,
+                    new_mirror_dir,
+                    exc,
+                )
+
         if touched:
             changed = True
             await self._rewrite_sidecar_paths(touched)
@@ -1700,7 +1742,9 @@ class ModelScanner:
     async def _rewrite_sidecar_paths(self, entries: List[Dict[str, Any]]) -> None:
         """Point each model's metadata sidecar at its new location.
 
-        Sidecars travel with the renamed directory, so only the recorded
+        In alongside mode sidecars travel with the renamed directory; in
+        centralized mode the mirror subtree has already been moved by the
+        caller (:meth:`rename_known_folder`). Either way only the recorded
         ``file_path``/``preview_url`` inside them need rewriting. Failures are
         logged and skipped — a stale sidecar is repaired by the next metadata
         refresh, and must not abort the rename.
@@ -1709,7 +1753,7 @@ class ModelScanner:
             file_path = item.get("file_path")
             if not file_path:
                 continue
-            metadata_path = f"{os.path.splitext(file_path)[0]}.metadata.json"
+            metadata_path = get_metadata_path(file_path)
             if not os.path.exists(metadata_path):
                 continue
             try:
@@ -1719,6 +1763,94 @@ class ModelScanner:
                     "Failed to rewrite metadata sidecar %s: %s", metadata_path, exc
                 )
 
+    def _find_pending_models_in_sidecar_mirror(self) -> List[Dict[str, Any]]:
+        """Mirror-tree counterpart of the alongside pending-hash filesystem scan.
+
+        Centralized mode stores ``.metadata.json`` sidecars in the mirror
+        tree, so walking the model folders finds nothing. Each mirror base is
+        resolved from a configured model root; a sidecar's recorded
+        ``file_path`` locates its model, with a stem-based probe under the
+        mapped model root as fallback (mirror path components are sanitized,
+        so reverse mapping is best-effort). Orphan sidecars whose model file
+        no longer exists are skipped, matching the alongside scan.
+        """
+
+        pending_models: List[Dict[str, Any]] = []
+
+        for root_path in self.get_model_roots():
+            mirror_base = resolve_centralized_dir_for_dir(root_path)
+            if not mirror_base or not os.path.isdir(mirror_base):
+                continue
+
+            for dirpath, dirnames, filenames in os.walk(mirror_base):
+                dirnames[:] = [d for d in dirnames if not _is_excluded_dir(d)]
+                for filename in filenames:
+                    if not filename.endswith(".metadata.json"):
+                        continue
+
+                    metadata_path = os.path.join(dirpath, filename)
+                    try:
+                        with open(metadata_path, "r", encoding="utf-8") as f:
+                            data = json.load(f)
+
+                        # Check if hash is pending
+                        hash_status = data.get("hash_status", "completed")
+                        sha256 = data.get("sha256", "")
+
+                        if hash_status != "completed" or not sha256:
+                            # Find corresponding model file: prefer the
+                            # sidecar's recorded path, then probe by stem
+                            # under the mapped model root.
+                            model_path = None
+                            recorded_path = data.get("file_path")
+                            if (
+                                isinstance(recorded_path, str)
+                                and recorded_path
+                                and os.path.exists(recorded_path)
+                            ):
+                                model_path = recorded_path
+                            else:
+                                model_name = filename.replace(".metadata.json", "")
+                                rel_dir = os.path.relpath(dirpath, mirror_base)
+                                candidate_dir = (
+                                    root_path
+                                    if rel_dir == os.curdir
+                                    else os.path.join(root_path, rel_dir)
+                                )
+                                for ext in self.file_extensions:
+                                    potential_path = os.path.join(
+                                        candidate_dir, model_name + ext
+                                    )
+                                    if os.path.exists(potential_path):
+                                        model_path = potential_path
+                                        break
+
+                            if model_path:
+                                pending_models.append(
+                                    {
+                                        "file_path": model_path.replace(os.sep, "/"),
+                                        "hash_status": hash_status,
+                                        "sha256": sha256,
+                                        **{
+                                            k: v
+                                            for k, v in data.items()
+                                            if k
+                                            not in [
+                                                "file_path",
+                                                "hash_status",
+                                                "sha256",
+                                            ]
+                                        },
+                                    }
+                                )
+                    except (json.JSONDecodeError, Exception) as e:
+                        logger.debug(
+                            f"Error reading metadata file {metadata_path}: {e}"
+                        )
+                        continue
+
+        return pending_models
+
     def _schedule_all_folders_backfill(self) -> None:
         """Kick off a one-shot background folder walk if none is running."""
         if self._all_folders_backfill_running:
@@ -1889,7 +2021,7 @@ class ModelScanner:
                         file_info['name'] = os.path.basename(file_path)
 
                         metadata = cast(Any, self.model_class).from_civitai_info(version_info, file_info, file_path)
-                        metadata.preview_url = find_preview_file(local_stem, os.path.dirname(file_path))
+                        metadata.preview_url = find_preview_file(local_stem, get_preview_dir(file_path))
                         await MetadataManager.save_metadata(file_path, metadata)
                         logger.info(f"Created metadata from .civitai.info for {file_path} (Reason: .civitai.info was found but .metadata.json was missing)")
                 except Exception as e:
@@ -2326,38 +2458,51 @@ class ModelScanner:
             # Move all associated files with the same base name
             source_metadata = None
             moved_metadata_path = None
-            
-            # Find all files with the same base name in the source directory
+
+            # Associated files (sidecar metadata, previews) sit next to the
+            # model in alongside mode and in the mirror tree in centralized
+            # mode; collect from every directory that holds them.
+            source_sidecar_dir = get_sidecar_dir(source_path)
+            target_sidecar_dir = get_sidecar_dir(target_file)
+            associated_dirs = [(source_dir, target_path)]
+            if os.path.normpath(source_sidecar_dir) != os.path.normpath(source_dir):
+                associated_dirs.append((source_sidecar_dir, target_sidecar_dir))
+
+            # Find all files with the same base name in the source directories
             files_to_move = []
-            try:
-                for file in os.listdir(source_dir):
-                    if file.startswith(base_name + ".") and file != os.path.basename(source_path):
-                        source_file_path = os.path.join(source_dir, file)
-                        # Generate new filename with the same base name as the model file
-                        file_suffix = file[len(base_name):]  # Get the part after base_name (e.g., ".metadata.json", ".preview.png")
-                        new_associated_filename = f"{final_base_name}{file_suffix}"
-                        target_associated_path = os.path.join(target_path, new_associated_filename)
-                        
-                        # Store metadata file path for special handling
-                        if file == f"{base_name}.metadata.json":
-                            source_metadata = source_file_path
-                            moved_metadata_path = target_associated_path
-                        else:
-                            files_to_move.append((source_file_path, target_associated_path))
-            except Exception as e:
-                logger.error(f"Error listing files in {source_dir}: {e}")
-            
+            metadata_filename = os.path.basename(get_metadata_path(source_path))
+            for assoc_source_dir, assoc_target_dir in associated_dirs:
+                try:
+                    for file in os.listdir(assoc_source_dir):
+                        if file.startswith(base_name + ".") and file != os.path.basename(source_path):
+                            source_file_path = os.path.join(assoc_source_dir, file)
+                            # Generate new filename with the same base name as the model file
+                            file_suffix = file[len(base_name):]  # Get the part after base_name (e.g., ".metadata.json", ".preview.png")
+                            new_associated_filename = f"{final_base_name}{file_suffix}"
+                            target_associated_path = os.path.join(assoc_target_dir, new_associated_filename)
+
+                            # Store metadata file path for special handling
+                            if file == metadata_filename:
+                                source_metadata = source_file_path
+                                moved_metadata_path = target_associated_path
+                            else:
+                                files_to_move.append((source_file_path, target_associated_path))
+                except Exception as e:
+                    logger.error(f"Error listing files in {assoc_source_dir}: {e}")
+
             # Move all associated files
             metadata = None
             for source_file, target_file_path in files_to_move:
                 try:
+                    os.makedirs(os.path.dirname(target_file_path), exist_ok=True)
                     shutil.move(source_file, target_file_path)
                 except Exception as e:
                     logger.error(f"Error moving associated file {source_file}: {e}")
-            
+
             # Handle metadata file specially to update paths
             if source_metadata and moved_metadata_path and os.path.exists(source_metadata):
                 try:
+                    os.makedirs(os.path.dirname(moved_metadata_path), exist_ok=True)
                     shutil.move(source_metadata, moved_metadata_path)
                     metadata = await self._update_metadata_paths(moved_metadata_path, target_file)
                 except Exception as e:
@@ -2399,7 +2544,7 @@ class ModelScanner:
             metadata['file_name'] = os.path.splitext(os.path.basename(model_path))[0]
             
             if 'preview_url' in metadata and metadata['preview_url']:
-                preview_dir = os.path.dirname(model_path)
+                preview_dir = get_preview_dir(model_path)
                 # Update preview filename to match the new base name
                 new_base_name = os.path.splitext(os.path.basename(model_path))[0]
                 preview_ext = get_preview_extension(metadata['preview_url'])
@@ -2759,7 +2904,7 @@ class ModelScanner:
 
             # Sidecar write-back: JSON null encodes the checked-unavailable
             # state. Skip silently when the sidecar does not exist.
-            metadata_path = f"{os.path.splitext(file_path)[0]}.metadata.json"
+            metadata_path = get_metadata_path(file_path)
             if os.path.exists(metadata_path):
                 with open(metadata_path, 'r', encoding='utf-8') as handle:
                     payload = json.load(handle)
@@ -2825,7 +2970,7 @@ class ModelScanner:
         if not file_path:
             return None
 
-        dir_path = os.path.dirname(file_path)
+        dir_path = get_preview_dir(file_path)
         base_name = os.path.splitext(os.path.basename(file_path))[0]
         preview_path = find_preview_file(base_name, dir_path)
         if preview_path:

py/services/other_scanner.py

@@ -12,6 +12,7 @@ from typing import Any, Dict, List, Optional
 from ..utils.models import OtherModelMetadata
 from ..utils.file_utils import find_preview_file, normalize_path, calculate_autov3
 from ..utils.metadata_manager import MetadataManager
+from ..utils.sidecar_paths import get_preview_dir, is_centralized
 from ..config import config
 from .model_scanner import ModelScanner, _is_excluded_dir
 from .model_hash_index import ModelHashIndex
@@ -72,10 +73,9 @@ class OtherScanner(ModelScanner):
                 return None
 
             base_name = os.path.splitext(os.path.basename(file_path))[0]
-            dir_path = os.path.dirname(file_path)
 
             # Find preview image
-            preview_url = find_preview_file(base_name, dir_path)
+            preview_url = find_preview_file(base_name, get_preview_dir(file_path))
 
             # AutoV3 reads only the safetensors header, so it is cheap even for
             # large files; record the checked state at creation time ("" =
@@ -333,6 +333,11 @@ class OtherScanner(ModelScanner):
 
     async def _find_pending_models_from_filesystem(self) -> List[Dict[str, Any]]:
         """Scan filesystem for other-model metadata files with pending hash status."""
+        # Centralized mode stores sidecars in the mirror tree, not next to the
+        # models; walk the mirror instead of the model folders.
+        if is_centralized():
+            return self._find_pending_models_in_sidecar_mirror()
+
         pending_models = []
 
         for root_path in self.get_model_roots():

py/services/pending_delete_service.py

@@ -38,6 +38,7 @@ from typing import (
 )
 
 from ..utils.constants import PREVIEW_EXTENSIONS
+from ..utils.sidecar_paths import get_metadata_path, get_sidecar_dir
 from ..utils import settings_paths
 
 logger = logging.getLogger(__name__)
@@ -669,13 +670,22 @@ class PendingDeleteService:
         """Enumerate existing artifacts exactly like delete_model_artifacts."""
         main_extension = ".safetensors" if main_extension is None else main_extension
         main_file = f"{file_name}{main_extension}" if main_extension else file_name
-        patterns = [main_file, f"{file_name}.metadata.json"]
+        model_path = os.path.join(target_dir, main_file)
+
+        artifacts: List[str] = []
+        main_path = os.path.abspath(model_path)
+        if os.path.exists(main_path):
+            artifacts.append(main_path)
+
+        # Sidecars/previews live in the sidecar dir (the model's own dir in
+        # alongside mode, the centralized mirror otherwise).
+        sidecar_dir = get_sidecar_dir(model_path)
+        patterns = [os.path.basename(get_metadata_path(model_path))]
         for ext in PREVIEW_EXTENSIONS:
             patterns.append(f"{file_name}{ext}")
 
-        artifacts: List[str] = []
         for pattern in patterns:
-            path = os.path.abspath(os.path.join(target_dir, pattern))
+            path = os.path.abspath(os.path.join(sidecar_dir, pattern))
             if os.path.exists(path):
                 artifacts.append(path)
         return artifacts
@@ -694,7 +704,9 @@ class PendingDeleteService:
         """
         for original_path in artifacts:
             staged_path = os.path.join(batch_dir, os.path.basename(original_path))
-            os.rename(original_path, staged_path)
+            # EXDEV-tolerant: centralized sidecars may live on a different
+            # filesystem than the staging batch dir under the model root.
+            self._restore_file(original_path, staged_path)
             staged_pairs.append(
                 {
                     "staged": os.path.abspath(staged_path),
@@ -736,13 +748,14 @@ class PendingDeleteService:
         return staged_pairs
 
     def _restore_file(self, staged_path: str, original_path: str) -> None:
-        """Restore a staged file to its original path, tolerating EXDEV.
+        """Move a file between staging and library paths, tolerating EXDEV.
 
         ``os.rename`` is atomic and preferred (model staging and most recipe
         restores are same-volume). Recipe staging copies into the settings-dir
-        staging parent, which may live on a DIFFERENT filesystem than the
-        recipes dir; rename then raises EXDEV. Fall back to ``shutil.copy2`` +
-        ``os.remove`` so the bytes are restored and the staged copy removed.
+        staging parent, and centralized sidecars live under the configured
+        sidecar root; both may live on a DIFFERENT filesystem than the target
+        dir, so rename can raise EXDEV. Fall back to ``shutil.copy2`` +
+        ``os.remove`` so the bytes are moved and the source copy removed.
         """
         try:
             os.rename(staged_path, original_path)
@@ -764,7 +777,10 @@ class PendingDeleteService:
             if not os.path.exists(staged_path):
                 continue
             try:
-                os.rename(staged_path, original_path)
+                # EXDEV-tolerant: centralized sidecars may have been copied
+                # across filesystems into staging, so plain os.rename would
+                # fail here and strand the only copy.
+                self._restore_file(staged_path, original_path)
             except OSError as exc:  # pragma: no cover - best-effort rollback
                 logger.warning(
                     "Failed to roll back staged file %s -> %s: %s",

py/services/preview_asset_service.py

@@ -10,6 +10,7 @@ from urllib.parse import urlparse
 from ..utils.constants import CARD_PREVIEW_WIDTH, PREVIEW_EXTENSIONS
 from ..utils.civitai_utils import rewrite_preview_url
 from ..utils.preview_selection import resolve_mature_threshold, select_preview_media
+from ..utils.sidecar_paths import get_metadata_path, get_preview_dir
 from .settings_manager import get_settings_manager
 
 logger = logging.getLogger(__name__)
@@ -63,6 +64,9 @@ class PreviewAssetService:
 
         base_name = os.path.splitext(os.path.splitext(os.path.basename(metadata_path))[0])[0]
         preview_dir = os.path.dirname(metadata_path)
+        # Centralized mirrors may not exist yet (unlike the model's own
+        # directory in alongside mode).
+        os.makedirs(preview_dir, exist_ok=True)
         is_video = first_preview.get("type") == "video"
         preview_url = first_preview.get("url")
 
@@ -159,7 +163,10 @@ class PreviewAssetService:
         """Replace an existing preview asset for a model."""
 
         base_name = os.path.splitext(os.path.basename(model_path))[0]
-        folder = os.path.dirname(model_path)
+        folder = get_preview_dir(model_path)
+        # Centralized mirrors may not exist yet (unlike the model's own
+        # directory in alongside mode).
+        os.makedirs(folder, exist_ok=True)
 
         extension, optimized_data = await self._convert_preview(
             preview_data, content_type, original_filename
@@ -179,7 +186,7 @@ class PreviewAssetService:
         with open(preview_path, "wb") as handle:
             handle.write(optimized_data)
 
-        metadata_path = os.path.splitext(model_path)[0] + ".metadata.json"
+        metadata_path = get_metadata_path(model_path)
         metadata = await metadata_loader(metadata_path)
         metadata["preview_url"] = preview_path
         metadata["preview_nsfw_level"] = nsfw_level

py/services/settings_manager.py

@@ -99,6 +99,8 @@ DEFAULT_SETTINGS: Dict[str, Any] = {
     "enable_other_models": False,
     "enabled_other_sub_types": list(DEFAULT_ENABLED_OTHER_SUB_TYPES),
     "recipes_path": "",
+    "sidecar_storage_mode": "alongside",
+    "sidecar_storage_path": "",
     "base_model_path_mappings": {},
     "download_path_templates": {},
     "download_filename_templates": {},
@@ -1616,9 +1618,30 @@ class SettingsManager:
 
         return os.path.abspath(os.path.normpath(os.path.expanduser(stripped)))
 
+    @staticmethod
+    def _normalize_sidecar_storage_mode(value: Any) -> str:
+        """Return a valid sidecar storage mode, falling back to ``alongside``."""
+
+        if isinstance(value, str):
+            normalized = value.strip().lower()
+            if normalized in ("alongside", "centralized"):
+                return normalized
+        return "alongside"
+
+    def _refresh_sidecar_storage_config(self) -> None:
+        """Rebuild dependent config state after sidecar storage settings change."""
+
+        try:
+            from ..config import config  # Local import to avoid circular dependency
+
+            config.refresh_preview_roots()
+        except Exception as exc:  # pragma: no cover - defensive logging
+            logger.debug(
+                "Failed to refresh config after sidecar storage change: %s", exc
+            )
+
     def _get_effective_recipes_dir(self, recipes_path: Optional[str] = None) -> str:
         """Resolve the effective recipes directory for the active library."""
-
         normalized_custom = self._normalize_recipes_path_value(
             self.settings.get("recipes_path", "")
             if recipes_path is None
@@ -1815,6 +1838,10 @@ class SettingsManager:
             target_recipes_dir = self._get_effective_recipes_dir(value)
             self._validate_recipes_storage_path(target_recipes_dir)
             self._migrate_recipes_directory(current_recipes_dir, target_recipes_dir)
+        elif key == "sidecar_storage_mode":
+            value = self._normalize_sidecar_storage_mode(value)
+        elif key == "sidecar_storage_path":
+            value = self._normalize_recipes_path_value(value)
         self.settings[key] = value
         portable_switch_pending = False
         if key == "use_portable_settings" and isinstance(value, bool):
@@ -1845,6 +1872,8 @@ class SettingsManager:
         self._save_settings()
         if key == "recipes_path":
             self._notify_library_change(self.get_active_library_name())
+        if key in ("sidecar_storage_mode", "sidecar_storage_path"):
+            self._refresh_sidecar_storage_config()
         if key in ("enable_other_models", "enabled_other_sub_types"):
             self._apply_other_model_settings_change()
         if portable_switch_pending:

py/services/tag_update_service.py

@@ -2,10 +2,9 @@
 
 from __future__ import annotations
 
-import os
-
 from typing import Awaitable, Callable, Dict, List, Sequence, Tuple
 
+from ..utils.sidecar_paths import get_metadata_path
 from .auto_tag_service import extract_auto_tags
 
 
@@ -24,8 +23,7 @@ class TagUpdateService:
         update_cache: Callable[[str, str, Dict[str, object]], Awaitable[bool]],
     ) -> Tuple[List[str], List[str]]:
         """Add tags to a metadata entry and return updated tags and auto_tags."""
-        base, _ = os.path.splitext(file_path)
-        metadata_path = f"{base}.metadata.json"
+        metadata_path = get_metadata_path(file_path)
         metadata = await metadata_loader(metadata_path)
 
         raw_tags = metadata.get("tags", [])

py/services/use_cases/__init__.py

@@ -21,6 +21,10 @@ from .example_images import (
     ImportExampleImagesValidationError,
 )
 from .filename_template_use_case import FilenameTemplateUseCase
+from .sidecar_migration_use_case import (
+    SidecarMigrationProgressReporter,
+    SidecarMigrationUseCase,
+)
 
 __all__ = [
     "AutoOrganizeInProgressError",
@@ -36,4 +40,6 @@ __all__ = [
     "ImportExampleImagesUseCase",
     "ImportExampleImagesValidationError",
     "FilenameTemplateUseCase",
+    "SidecarMigrationProgressReporter",
+    "SidecarMigrationUseCase",
 ]

py/services/use_cases/filename_template_use_case.py

@@ -12,6 +12,7 @@ import os
 from typing import Any, Awaitable, Callable, Dict, List, Optional, Sequence
 
 from ...utils.constants import AUTO_ORGANIZE_BATCH_SIZE
+from ...utils.sidecar_paths import get_metadata_path
 from ...utils.utils import calculate_filename_for_model
 from ..model_file_service import AutoOrganizeResult, ProgressCallback
 from ..model_lifecycle_service import ModelLifecycleService, load_local_metadata
@@ -200,7 +201,7 @@ class FilenameTemplateUseCase:
         sidecar or no ``original_file_name`` entry exists (models never
         renamed, or renamed before the recording shipped).
         """
-        metadata_path = f"{os.path.splitext(file_path)[0]}.metadata.json"
+        metadata_path = get_metadata_path(file_path)
         metadata = await self._metadata_loader(metadata_path)
         original = metadata.get("original_file_name")
         if not isinstance(original, str):

py/services/use_cases/sidecar_migration_use_case.py

@@ -0,0 +1,685 @@
+"""Use case migrating sidecar metadata and previews between storage layouts.
+
+Two storage layouts exist (see :mod:`py.utils.sidecar_paths`):
+
+- ``alongside``: ``<model_dir>/<name>.metadata.json`` and preview files live
+  next to the model file.
+- ``centralized``: the same files live under the configured sidecar root,
+  mirroring the library-relative directory structure.
+
+This use case moves the ``.metadata.json`` sidecar and preview files for every
+known model from one layout to the other. Model files themselves NEVER move.
+Paths inside the moved sidecar (``file_path``, ``file_name``, ``preview_url``)
+are rewritten the same way :meth:`ModelScanner._update_metadata_paths` does.
+After the move, scanner caches are reconciled so the list API immediately
+serves the new preview locations instead of stale pre-migration URLs.
+
+Intended flow (settings-first):
+
+1. The user switches ``sidecar_storage_mode`` (and optionally
+   ``sidecar_storage_path``) in settings.
+2. The migration runs in the direction of the NEW mode with ``force=True``.
+   After the switch, files in the OLD layout are the source of truth; the
+   guard below would otherwise refuse to run because the active mode already
+   matches the migration target.
+
+Both orderings work because all path computations are mode-independent: the
+alongside location is derived from the model path directly, and the mirror
+location is resolved via ``get_configured_sidecar_root()``, which ignores the
+active mode.
+
+Guards (pass ``force=True`` to bypass):
+
+- ``migrate_to_centralized`` refuses when centralized storage is already the
+  active, resolvable mode.
+- ``migrate_to_alongside`` refuses when the active mode is ``alongside``.
+"""
+
+from __future__ import annotations
+
+import errno
+import json
+import logging
+import os
+import shutil
+from typing import Any, Awaitable, Callable, Dict, List, Optional, Protocol, Sequence, Tuple
+
+from ..service_registry import ServiceRegistry
+from ..settings_manager import get_settings_manager
+from ...utils.constants import PREVIEW_EXTENSIONS
+from ...utils.file_utils import find_preview_file, get_preview_extension
+from ...utils.metadata_manager import MetadataManager
+from ...utils.sidecar_paths import (
+    METADATA_SUFFIX,
+    STORAGE_MODE_CENTRALIZED,
+    get_configured_sidecar_root,
+    get_sidecar_root,
+    get_storage_mode,
+    resolve_centralized_dir_for_dir,
+)
+
+
+class SidecarMigrationProgressReporter(Protocol):
+    """Protocol for progress reporters used during sidecar migration."""
+
+    async def on_progress(self, payload: Dict[str, Any]) -> None:
+        """Handle a sidecar migration progress update."""
+
+
+ScannerFactory = Callable[[], Awaitable[Any]]
+
+DIRECTION_TO_CENTRALIZED = "to_centralized"
+DIRECTION_TO_ALONGSIDE = "to_alongside"
+DIRECTION_RELOCATE_ROOT = "relocate_root"
+
+# Same candidate set find_preview_file recognizes: every PREVIEW_EXTENSIONS
+# suffix plus the legacy ".example.0.jpeg" (issue #225).
+_PREVIEW_CANDIDATE_EXTENSIONS = tuple(PREVIEW_EXTENSIONS) + (".example.0.jpeg",)
+
+
+def _enumerate_preview_names(directory: str, stem: str) -> List[str]:
+    """Return preview filenames for ``stem`` present in ``directory``.
+
+    Case-insensitive full-name match against the preview candidate set, so
+    files like ``model.WEBP`` or ``model.Png`` placed by external tools are
+    migrated along with the exact-case variants.
+    """
+
+    targets = {f"{stem.lower()}{ext}" for ext in _PREVIEW_CANDIDATE_EXTENSIONS}
+    try:
+        entries = os.listdir(directory)
+    except OSError:
+        return []
+    return [entry for entry in entries if entry.lower() in targets]
+
+
+class SidecarMigrationUseCase:
+    """Move sidecars and previews between alongside and centralized layouts."""
+
+    def __init__(
+        self,
+        *,
+        scanner_factories: Sequence[Tuple[str, ScannerFactory]] | None = None,
+        settings_service=None,
+        logger: Optional[logging.Logger] = None,
+    ) -> None:
+        self._settings = settings_service or get_settings_manager()
+        self._scanner_factories: Tuple[Tuple[str, ScannerFactory], ...] = tuple(
+            scanner_factories
+            or (
+                ("lora", ServiceRegistry.get_lora_scanner),
+                ("checkpoint", ServiceRegistry.get_checkpoint_scanner),
+                ("embedding", ServiceRegistry.get_embedding_scanner),
+                ("other", ServiceRegistry.get_other_scanner),
+            )
+        )
+        self._logger = logger or logging.getLogger(__name__)
+
+    async def migrate_to_centralized(
+        self,
+        progress_cb: Optional[SidecarMigrationProgressReporter] = None,
+        *,
+        force: bool = False,
+    ) -> Dict[str, Any]:
+        """Move sidecars/previews from alongside the models into the mirror root."""
+
+        if (
+            not force
+            and get_storage_mode() == STORAGE_MODE_CENTRALIZED
+            and get_sidecar_root()
+        ):
+            return self._refusal(
+                DIRECTION_TO_CENTRALIZED,
+                "sidecar storage is already centralized; pass force=true to migrate anyway",
+            )
+        return await self._migrate(
+            direction=DIRECTION_TO_CENTRALIZED,
+            to_centralized=True,
+            progress_cb=progress_cb,
+        )
+
+    async def migrate_to_alongside(
+        self,
+        progress_cb: Optional[SidecarMigrationProgressReporter] = None,
+        *,
+        force: bool = False,
+    ) -> Dict[str, Any]:
+        """Move sidecars/previews from the mirror root back next to the models."""
+
+        if not force and get_storage_mode() != STORAGE_MODE_CENTRALIZED:
+            return self._refusal(
+                DIRECTION_TO_ALONGSIDE,
+                "sidecar storage is already alongside; pass force=true to migrate anyway",
+            )
+        return await self._migrate(
+            direction=DIRECTION_TO_ALONGSIDE,
+            to_centralized=False,
+            progress_cb=progress_cb,
+        )
+
+    async def migrate_root(
+        self,
+        old_root: str,
+        progress_cb: Optional[SidecarMigrationProgressReporter] = None,
+        *,
+        force: bool = False,
+    ) -> Dict[str, Any]:
+        """Relocate the whole mirror tree from a previous root to the configured one.
+
+        Used after ``sidecar_storage_path`` changes while centralized storage
+        is active: without it, every asset under the old root would silently
+        disappear from the application. Moves every file keeping the
+        root-relative structure, rewrites the ``preview_url`` prefix inside
+        moved sidecars, reconciles scanner caches, and prunes the emptied old
+        tree. Keep-newer conflict resolution matches :meth:`_transfer`.
+        """
+
+        if not force and get_storage_mode() != STORAGE_MODE_CENTRALIZED:
+            return self._refusal(
+                DIRECTION_RELOCATE_ROOT,
+                "sidecar storage is not centralized; pass force=true to relocate anyway",
+            )
+        new_root = get_configured_sidecar_root()
+        if not new_root:
+            return self._refusal(
+                DIRECTION_RELOCATE_ROOT,
+                "cannot resolve the centralized sidecar root",
+            )
+        old = (
+            os.path.abspath(os.path.expanduser(old_root.strip()))
+            if isinstance(old_root, str) and old_root.strip()
+            else ""
+        )
+        if not old:
+            return self._refusal(DIRECTION_RELOCATE_ROOT, "old_root is required")
+        if os.path.normpath(old) == os.path.normpath(new_root):
+            return self._refusal(
+                DIRECTION_RELOCATE_ROOT,
+                "old_root matches the configured sidecar root",
+            )
+
+        files: List[Tuple[str, str]] = []
+        if os.path.isdir(old):
+            for dirpath, _dirnames, filenames in os.walk(old):
+                rel = os.path.relpath(dirpath, old)
+                target_dir = new_root if rel == os.curdir else os.path.join(new_root, rel)
+                for filename in filenames:
+                    files.append(
+                        (os.path.join(dirpath, filename), os.path.join(target_dir, filename))
+                    )
+
+        errors: List[Dict[str, str]] = []
+        counters: Dict[str, Any] = {"moved": 0, "conflicts": 0}
+        moved_sidecars: List[str] = []
+
+        async def emit(status: str, **extra: Any) -> None:
+            if progress_cb is None:
+                return
+            payload: Dict[str, Any] = {
+                "type": "sidecar_migration_progress",
+                "status": status,
+                "direction": DIRECTION_RELOCATE_ROOT,
+                "total": len(files),
+                "processed": extra.pop("processed", 0),
+                "moved": counters["moved"],
+                "skipped": 0,
+                "conflicts": counters["conflicts"],
+                "errors": len(errors),
+            }
+            payload.update(extra)
+            await progress_cb.on_progress(payload)
+
+        await emit("started")
+
+        for index, (src, dst) in enumerate(files, start=1):
+            try:
+                if self._transfer(src, dst, counters) and src.endswith(METADATA_SUFFIX):
+                    moved_sidecars.append(dst)
+            except Exception as exc:
+                self._logger.error(
+                    "Sidecar root relocation failed for %s: %s", src, exc, exc_info=True
+                )
+                errors.append({"model": os.path.basename(src), "error": str(exc)})
+            await emit("processing", processed=index, current=os.path.basename(src))
+
+        old_prefix = old.replace(os.sep, "/").rstrip("/") + "/"
+        new_prefix = new_root.replace(os.sep, "/").rstrip("/") + "/"
+        for sidecar in moved_sidecars:
+            self._rewrite_root_prefix(sidecar, old_prefix, new_prefix)
+        await self._reconcile_root_prefix(old_prefix, new_prefix)
+
+        # Prune the emptied old tree, best-effort.
+        if os.path.isdir(old):
+            for dirpath, dirnames, filenames in os.walk(old, topdown=False):
+                if filenames:
+                    continue
+                for dirname in dirnames:
+                    try:
+                        os.rmdir(os.path.join(dirpath, dirname))
+                    except OSError:
+                        pass
+                try:
+                    os.rmdir(dirpath)
+                except OSError:
+                    pass
+
+        await emit("completed")
+
+        return {
+            "success": not errors,
+            "direction": DIRECTION_RELOCATE_ROOT,
+            "models_total": len(files),
+            "models_processed": len(files),
+            "models_moved": 0,
+            "moved": counters["moved"],
+            "skipped": 0,
+            "conflicts": counters["conflicts"],
+            "errors": errors,
+            "error_count": len(errors),
+        }
+
+    def _rewrite_root_prefix(
+        self, sidecar_path: str, old_prefix: str, new_prefix: str
+    ) -> None:
+        """Repoint preview_url inside a relocated sidecar from old to new root."""
+
+        try:
+            with open(sidecar_path, "r", encoding="utf-8") as handle:
+                metadata = json.load(handle)
+        except (OSError, json.JSONDecodeError) as exc:
+            self._logger.warning(
+                "Sidecar root relocation: cannot read %s: %s", sidecar_path, exc
+            )
+            return
+
+        preview_url = metadata.get("preview_url")
+        if not isinstance(preview_url, str) or not preview_url.startswith(old_prefix):
+            return
+        metadata["preview_url"] = new_prefix + preview_url[len(old_prefix):]
+        try:
+            with open(sidecar_path, "w", encoding="utf-8") as handle:
+                json.dump(metadata, handle, ensure_ascii=False, indent=2)
+        except OSError as exc:
+            self._logger.warning(
+                "Sidecar root relocation: cannot rewrite %s: %s", sidecar_path, exc
+            )
+
+    async def _reconcile_root_prefix(self, old_prefix: str, new_prefix: str) -> None:
+        """Rewrite old-root preview URLs in every scanner cache after relocation."""
+
+        for model_type, factory in self._active_scanner_factories():
+            try:
+                scanner = await factory()
+                cache = await scanner.get_cached_data()
+                changed = False
+                for item in cache.raw_data:
+                    preview_url = item.get("preview_url")
+                    if (
+                        isinstance(preview_url, str)
+                        and preview_url.startswith(old_prefix)
+                    ):
+                        item["preview_url"] = new_prefix + preview_url[len(old_prefix):]
+                        changed = True
+                if changed and hasattr(scanner, "_persist_current_cache"):
+                    await scanner._persist_current_cache()
+            except Exception as exc:
+                self._logger.error(
+                    "Sidecar root relocation: failed to reconcile %s cache: %s",
+                    model_type,
+                    exc,
+                    exc_info=True,
+                )
+
+    @staticmethod
+    def _refusal(direction: str, message: str) -> Dict[str, Any]:
+        return {
+            "success": False,
+            "error": message,
+            "direction": direction,
+            "models_total": 0,
+            "models_processed": 0,
+            "models_moved": 0,
+            "moved": 0,
+            "skipped": 0,
+            "conflicts": 0,
+            "errors": [],
+            "error_count": 0,
+        }
+
+    def _active_scanner_factories(self) -> Tuple[Tuple[str, ScannerFactory], ...]:
+        """Drop the opt-in other scanner while Other Models is disabled."""
+
+        if self._settings.is_other_models_enabled():
+            return self._scanner_factories
+        return tuple(entry for entry in self._scanner_factories if entry[0] != "other")
+
+    async def _collect_model_paths(
+        self, errors: List[Dict[str, str]]
+    ) -> List[Tuple[Any, List[str]]]:
+        """Enumerate model file paths grouped by the scanner that owns them."""
+
+        groups: List[Tuple[Any, List[str]]] = []
+        for model_type, factory in self._active_scanner_factories():
+            try:
+                scanner = await factory()
+                cache = await scanner.get_cached_data()
+            except Exception as exc:
+                self._logger.error(
+                    "Sidecar migration: failed to enumerate %s models: %s",
+                    model_type,
+                    exc,
+                )
+                errors.append({"model": model_type, "error": f"enumeration failed: {exc}"})
+                continue
+            paths = [
+                entry["file_path"]
+                for entry in cache.raw_data
+                if entry.get("file_path")
+            ]
+            groups.append((scanner, paths))
+        return groups
+
+    @staticmethod
+    def _move_file(src: str, dst: str) -> None:
+        """Move a file, tolerating EXDEV when the layouts span filesystems."""
+
+        os.makedirs(os.path.dirname(dst), exist_ok=True)
+        try:
+            os.rename(src, dst)
+        except OSError as exc:
+            if exc.errno != errno.EXDEV:
+                raise
+            shutil.copy2(src, dst)
+            os.remove(src)
+
+    async def _migrate(
+        self,
+        *,
+        direction: str,
+        to_centralized: bool,
+        progress_cb: Optional[SidecarMigrationProgressReporter],
+    ) -> Dict[str, Any]:
+        root = get_configured_sidecar_root()
+        if not root:
+            return self._refusal(
+                direction,
+                "cannot resolve the centralized sidecar root",
+            )
+
+        errors: List[Dict[str, str]] = []
+        scanner_groups = await self._collect_model_paths(errors)
+
+        total = sum(len(paths) for _, paths in scanner_groups)
+        processed = 0
+        models_moved = 0
+        moved = 0
+        skipped = 0
+        conflicts = 0
+        # (file_path, final preview path at the destination layout), grouped
+        # by scanner so caches can be reconciled after the move.
+        preview_updates: List[Tuple[Any, List[Tuple[str, str]]]] = []
+
+        async def emit(status: str, **extra: Any) -> None:
+            if progress_cb is None:
+                return
+            payload: Dict[str, Any] = {
+                "type": "sidecar_migration_progress",
+                "status": status,
+                "direction": direction,
+                "total": total,
+                "processed": processed,
+                "moved": moved,
+                "skipped": skipped,
+                "conflicts": conflicts,
+                "errors": len(errors),
+            }
+            payload.update(extra)
+            await progress_cb.on_progress(payload)
+
+        await emit("started")
+
+        for scanner, model_paths in scanner_groups:
+            updates: List[Tuple[str, str]] = []
+            for model_path in model_paths:
+                processed += 1
+                current = os.path.basename(model_path)
+                try:
+                    result = await self._migrate_model(
+                        model_path,
+                        root=root,
+                        to_centralized=to_centralized,
+                    )
+                    moved += result["moved"]
+                    conflicts += result["conflicts"]
+                    if result["skipped"]:
+                        skipped += 1
+                    else:
+                        updates.append((model_path, result["preview_url"]))
+                    if result["moved"]:
+                        models_moved += 1
+                except Exception as exc:
+                    self._logger.error(
+                        "Sidecar migration failed for %s: %s", model_path, exc, exc_info=True
+                    )
+                    errors.append({"model": current, "error": str(exc)})
+                await emit("processing", current=current)
+            preview_updates.append((scanner, updates))
+
+        await self._reconcile_scanner_caches(preview_updates)
+
+        await emit("completed")
+
+        return {
+            "success": not errors,
+            "direction": direction,
+            "models_total": total,
+            "models_processed": processed,
+            "models_moved": models_moved,
+            "moved": moved,
+            "skipped": skipped,
+            "conflicts": conflicts,
+            "errors": errors,
+            "error_count": len(errors),
+        }
+
+    async def _migrate_model(
+        self,
+        model_path: str,
+        *,
+        root: str,
+        to_centralized: bool,
+    ) -> Dict[str, Any]:
+        """Migrate one model's sidecar + previews; return per-model counters.
+
+        ``preview_url`` in the result is the model's final preview path in the
+        destination layout ("" when none), used to reconcile scanner caches.
+        """
+
+        result: Dict[str, Any] = {"moved": 0, "conflicts": 0, "skipped": 0, "preview_url": ""}
+
+        model_path = os.path.abspath(model_path)
+        if not os.path.exists(model_path):
+            self._logger.warning(
+                "Sidecar migration: model file missing, skipping: %s", model_path
+            )
+            result["skipped"] = 1
+            return result
+
+        model_dir = os.path.dirname(model_path)
+        mirror_dir = resolve_centralized_dir_for_dir(model_dir, sidecar_root=root)
+        if mirror_dir is None:
+            self._logger.warning(
+                "Sidecar migration: %s is outside configured model roots, skipping",
+                model_path,
+            )
+            result["skipped"] = 1
+            return result
+
+        if to_centralized:
+            src_dir, dst_dir = model_dir, mirror_dir
+        else:
+            src_dir, dst_dir = mirror_dir, model_dir
+
+        if os.path.normpath(src_dir) == os.path.normpath(dst_dir):
+            result["skipped"] = 1
+            return result
+
+        stem = os.path.splitext(os.path.basename(model_path))[0]
+        sidecar_name = stem + METADATA_SUFFIX
+
+        moved_previews: List[str] = []
+        for preview_name in _enumerate_preview_names(src_dir, stem):
+            src = os.path.join(src_dir, preview_name)
+            dst = os.path.join(dst_dir, preview_name)
+            if self._transfer(src, dst, result):
+                moved_previews.append(dst)
+
+        sidecar_src = os.path.join(src_dir, sidecar_name)
+        sidecar_moved = False
+        sidecar_dst = os.path.join(dst_dir, sidecar_name)
+        if os.path.exists(sidecar_src):
+            sidecar_moved = self._transfer(sidecar_src, sidecar_dst, result)
+
+        if sidecar_moved:
+            await self._rewrite_sidecar_paths(sidecar_dst, model_path, moved_previews)
+
+        # Ground truth from the destination directory: covers conflict-keep
+        # and partial moves, not just the previews transferred in this run.
+        final_preview = find_preview_file(stem, dst_dir)
+        if final_preview:
+            result["preview_url"] = final_preview.replace(os.sep, "/")
+
+        return result
+
+    async def _reconcile_scanner_caches(
+        self, preview_updates: List[Tuple[Any, List[Tuple[str, str]]]]
+    ) -> None:
+        """Point scanner cache entries at the post-migration preview locations.
+
+        Without this the list API keeps serving pre-migration ``preview_url``
+        values whose files no longer exist; hitting one triggers the preview
+        route's stale-URL cleanup, which would wipe the reference for good.
+        A failing scanner is logged and skipped — the on-disk migration has
+        already succeeded, and a full rescan repairs the cache.
+        """
+
+        for scanner, updates in preview_updates:
+            if not updates:
+                continue
+            try:
+                cache = await scanner.get_cached_data()
+                changed = False
+                for file_path, preview_url in updates:
+                    entry = next(
+                        (item for item in cache.raw_data if item.get("file_path") == file_path),
+                        None,
+                    )
+                    if entry is None:
+                        continue
+                    if entry.get("preview_url", "") == preview_url:
+                        continue
+                    if hasattr(cache, "update_preview_url"):
+                        await cache.update_preview_url(
+                            file_path,
+                            preview_url,
+                            entry.get("preview_nsfw_level", 0),
+                        )
+                    else:  # pragma: no cover - minimal cache doubles
+                        entry["preview_url"] = preview_url
+                    changed = True
+                if changed and hasattr(scanner, "_persist_current_cache"):
+                    await scanner._persist_current_cache()
+            except Exception as exc:
+                self._logger.error(
+                    "Sidecar migration: failed to reconcile scanner cache: %s",
+                    exc,
+                    exc_info=True,
+                )
+
+    def _transfer(self, src: str, dst: str, result: Dict[str, Any]) -> bool:
+        """Move ``src`` to ``dst`` with keep-newer conflict resolution.
+
+        Returns True when the file was actually moved to the destination. On a
+        conflict the newer file wins: a newer source replaces the destination;
+        a newer (or equal) destination is kept and the source is deleted.
+        """
+
+        if os.path.exists(dst):
+            result["conflicts"] += 1
+            if os.path.getmtime(src) > os.path.getmtime(dst):
+                self._logger.info(
+                    "Sidecar migration: conflict at %s; source is newer, replacing", dst
+                )
+                os.remove(dst)
+            else:
+                self._logger.info(
+                    "Sidecar migration: conflict at %s; destination is newer, keeping it",
+                    dst,
+                )
+                os.remove(src)
+                return False
+        self._move_file(src, dst)
+        result["moved"] += 1
+        return True
+
+    async def _rewrite_sidecar_paths(
+        self,
+        sidecar_path: str,
+        model_path: str,
+        moved_previews: List[str],
+    ) -> None:
+        """Update path fields inside a moved sidecar, mirroring ModelScanner."""
+
+        with open(sidecar_path, "r", encoding="utf-8") as handle:
+            metadata = json.load(handle)
+
+        stem = os.path.splitext(os.path.basename(model_path))[0]
+        metadata["file_path"] = model_path.replace(os.sep, "/")
+        metadata["file_name"] = stem
+
+        if moved_previews and metadata.get("preview_url"):
+            recorded_ext = get_preview_extension(metadata["preview_url"])
+            chosen = next(
+                (
+                    path
+                    for path in moved_previews
+                    if get_preview_extension(path) == recorded_ext
+                ),
+                moved_previews[0],
+            )
+            metadata["preview_url"] = chosen.replace(os.sep, "/")
+
+        await MetadataManager.save_metadata(sidecar_path, metadata)
+
+    async def execute_with_error_handling(
+        self,
+        *,
+        direction: str,
+        progress_cb: Optional[SidecarMigrationProgressReporter] = None,
+        force: bool = False,
+        old_root: Optional[str] = None,
+    ) -> Dict[str, Any]:
+        """Wrapper providing progress notification on unexpected failures."""
+
+        try:
+            if direction == DIRECTION_TO_CENTRALIZED:
+                return await self.migrate_to_centralized(progress_cb, force=force)
+            if direction == DIRECTION_TO_ALONGSIDE:
+                return await self.migrate_to_alongside(progress_cb, force=force)
+            if direction == DIRECTION_RELOCATE_ROOT:
+                return await self.migrate_root(old_root or "", progress_cb, force=force)
+            raise ValueError(
+                f"direction must be {DIRECTION_TO_CENTRALIZED!r}, "
+                f"{DIRECTION_TO_ALONGSIDE!r} or {DIRECTION_RELOCATE_ROOT!r}"
+            )
+        except Exception as exc:
+            if progress_cb is not None:
+                await progress_cb.on_progress(
+                    {
+                        "type": "sidecar_migration_progress",
+                        "status": "error",
+                        "direction": direction,
+                        "error": str(exc),
+                    }
+                )
+            raise

py/utils/metadata_manager.py

@@ -8,6 +8,7 @@ from typing import Any, Dict, Optional, Type, Union, cast
 from .models import BaseModelMetadata, CheckpointMetadata, EmbeddingMetadata, LoraMetadata
 from .file_utils import normalize_path, find_preview_file, calculate_sha256, calculate_autov3
 from .lora_metadata import extract_lora_metadata, extract_checkpoint_metadata
+from .sidecar_paths import get_metadata_path, get_preview_dir, resolve_metadata_path
 
 logger = logging.getLogger(__name__)
 
@@ -32,7 +33,7 @@ class MetadataManager:
             - metadata: BaseModelMetadata instance or None
             - should_skip: True if corrupted metadata file exists and model should be skipped
         """
-        metadata_path = f"{os.path.splitext(file_path)[0]}.metadata.json"
+        metadata_path = get_metadata_path(file_path)
         
         # Check if metadata file exists
         if not os.path.exists(metadata_path):
@@ -98,11 +99,7 @@ class MetadataManager:
                 payload.update(unknown_fields)
         else:
             if not should_skip:
-                metadata_path = (
-                    file_path
-                    if file_path.endswith(".metadata.json")
-                    else f"{os.path.splitext(file_path)[0]}.metadata.json"
-                )
+                metadata_path = resolve_metadata_path(file_path)
                 if os.path.exists(metadata_path):
                     try:
                         with open(metadata_path, "r", encoding="utf-8") as handle:
@@ -150,7 +147,7 @@ class MetadataManager:
             return model_data
 
         folder = model_data.get("folder")
-        metadata_path = f"{os.path.splitext(file_path)[0]}.metadata.json"
+        metadata_path = get_metadata_path(file_path)
         sidecar_exists = os.path.exists(metadata_path)
         cached = model_data.copy()
         payload = await MetadataManager.load_metadata_payload(file_path)
@@ -188,15 +185,14 @@ class MetadataManager:
           bool: Success or failure
         """
         # Determine if the input is a metadata path or a model file path
-        if path.endswith('.metadata.json'):
-            metadata_path = path
-        else:
-            # Use existing logic for model file paths
-            file_path = path
-            metadata_path = f"{os.path.splitext(file_path)[0]}.metadata.json"
+        metadata_path = resolve_metadata_path(path)
         temp_path = f"{metadata_path}.tmp"
         
         try:
+            # Centralized sidecar mirrors may not exist yet (unlike the model's
+            # own directory in alongside mode, which always does).
+            os.makedirs(os.path.dirname(metadata_path), exist_ok=True)
+
             # Convert to dict if needed
             if isinstance(metadata, BaseModelMetadata):
                 metadata_dict = metadata.to_dict()
@@ -259,10 +255,9 @@ class MetadataManager:
             
         try:
             base_name = os.path.splitext(os.path.basename(file_path))[0]
-            dir_path = os.path.dirname(file_path)
-            
+
             # Find preview image
-            preview_url = find_preview_file(base_name, dir_path)
+            preview_url = find_preview_file(base_name, get_preview_dir(file_path))
             
             # Calculate file hash
             start_hash_time = time.perf_counter()
@@ -386,15 +381,16 @@ class MetadataManager:
         # Check if preview exists at the current location
         preview_url = metadata.preview_url
         if preview_url:
-            # Get directory parts of both paths
-            file_dir = os.path.dirname(file_path)
+            # Get directory parts of both paths; the preview directory is the
+            # sidecar/preview dir (the model's own dir in alongside mode, the
+            # centralized mirror otherwise).
+            file_dir = get_preview_dir(file_path)
             preview_dir = os.path.dirname(preview_url)
             
             # Update preview if it doesn't exist OR if model and preview are in different directories
             if not os.path.exists(preview_url) or file_dir != preview_dir:
                 base_name = os.path.splitext(os.path.basename(file_path))[0]
-                dir_path = os.path.dirname(file_path)
-                new_preview_url = find_preview_file(base_name, dir_path)
+                new_preview_url = find_preview_file(base_name, file_dir)
                 if new_preview_url:
                     metadata.preview_url = normalize_path(new_preview_url)
                     need_update = True

py/utils/sidecar_paths.py

@@ -0,0 +1,268 @@
+"""Resolution of sidecar metadata and preview storage paths.
+
+All code that needs the on-disk location of a model's ``.metadata.json``
+sidecar or preview assets MUST go through these helpers instead of deriving
+paths inline (``splitext(model_path)[0] + ".metadata.json"`` and friends).
+
+Two storage modes are supported, selected by the ``sidecar_storage_mode``
+setting:
+
+- ``alongside`` (default): sidecars and previews live next to the model
+  file, the historical layout other tools may rely on.
+- ``centralized``: sidecars and previews live under a configurable root
+  (``sidecar_storage_path`` setting, default ``<settings_dir>/sidecars``),
+  mirroring the library-relative directory structure:
+  ``<root>/<library>/<root_basename-roothash>/<rel_dir>/<name>.metadata.json``.
+
+All helpers are pure path computations: no directory scans and no file I/O
+on the hot path. Settings lookups go through ``SettingsManager.get`` (a dict
+read); config roots come from the already-initialized ``config`` singleton.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import logging
+import os
+import re
+from typing import List, Optional
+
+logger = logging.getLogger(__name__)
+
+METADATA_SUFFIX = ".metadata.json"
+
+STORAGE_MODE_ALONGSIDE = "alongside"
+STORAGE_MODE_CENTRALIZED = "centralized"
+
+_VALID_MODES = frozenset({STORAGE_MODE_ALONGSIDE, STORAGE_MODE_CENTRALIZED})
+
+
+def _get_settings_value(key: str, default=None):
+    """Read a setting defensively; never fail path resolution on settings errors."""
+
+    try:
+        from ..services.settings_manager import get_settings_manager
+
+        value = get_settings_manager().get(key)
+    except Exception as exc:  # pragma: no cover - defensive fallback
+        logger.debug("sidecar_paths: settings lookup for %r failed: %s", key, exc)
+        return default
+    return default if value is None else value
+
+
+def get_storage_mode() -> str:
+    """Return the active sidecar storage mode (``alongside`` unless configured)."""
+
+    mode = _get_settings_value("sidecar_storage_mode", STORAGE_MODE_ALONGSIDE)
+    if mode not in _VALID_MODES:
+        return STORAGE_MODE_ALONGSIDE
+    return mode
+
+
+def is_centralized() -> bool:
+    """Return True when centralized sidecar storage is active and resolvable."""
+
+    return get_storage_mode() == STORAGE_MODE_CENTRALIZED and bool(get_sidecar_root())
+
+
+def _resolve_root_from_settings() -> str:
+    """Resolve the configured/default centralized root, ignoring the active mode."""
+
+    configured = _get_settings_value("sidecar_storage_path", "")
+    if configured and isinstance(configured, str):
+        root = os.path.abspath(os.path.expanduser(configured.strip()))
+        if root:
+            return root
+
+    # Default: <settings_dir>/sidecars
+    try:
+        from .settings_paths import get_settings_dir
+
+        return os.path.join(get_settings_dir(), "sidecars")
+    except Exception as exc:  # pragma: no cover - defensive fallback
+        logger.warning("sidecar_paths: cannot resolve default sidecar root: %s", exc)
+        return ""
+
+
+def get_sidecar_root() -> str:
+    """Return the absolute root directory for centralized sidecar storage.
+
+    Empty string when centralized storage is not usable (mode alongside or an
+    unresolvable configured path).
+    """
+
+    if get_storage_mode() != STORAGE_MODE_CENTRALIZED:
+        return ""
+
+    return _resolve_root_from_settings()
+
+
+def get_configured_sidecar_root() -> str:
+    """Return the centralized sidecar root regardless of the active mode.
+
+    Unlike :func:`get_sidecar_root`, this resolves the configured
+    ``sidecar_storage_path`` (or the ``<settings_dir>/sidecars`` default) even
+    when the storage mode is ``alongside``. Migration tooling needs both
+    layouts at once and must not depend on which mode is currently active.
+    """
+
+    return _resolve_root_from_settings()
+
+
+def sanitize_path_component(name: str) -> str:
+    """Return a filesystem-safe single path component."""
+
+    safe = re.sub(r"[^A-Za-z0-9_.-]", "_", name or "")
+    return safe or "_"
+
+
+def _iter_model_roots() -> List[str]:
+    """Return every configured model root for the active library."""
+
+    try:
+        from ..config import config
+    except Exception as exc:  # pragma: no cover - defensive fallback
+        logger.debug("sidecar_paths: config unavailable: %s", exc)
+        return []
+
+    roots: List[str] = []
+    for attr in (
+        "loras_roots",
+        "base_models_roots",
+        "embeddings_roots",
+        "other_roots",
+        "extra_loras_roots",
+        "extra_checkpoints_roots",
+        "extra_unet_roots",
+        "extra_embeddings_roots",
+    ):
+        value = getattr(config, attr, None)
+        if value:
+            roots.extend(value)
+    return roots
+
+
+def _normalize_for_match(path: str) -> str:
+    return os.path.normpath(os.path.abspath(path))
+
+
+def root_mirror_component(root_path: str) -> str:
+    """Return the mirror path component identifying a model root.
+
+    ``<sanitized basename>-<hash>`` where the hash is a short digest of the
+    normalized absolute root path. Two roots sharing a basename (e.g.
+    ``/mnt/a/loras`` and ``/mnt/b/loras``) would otherwise map to the same
+    mirror directory and overwrite each other's sidecars.
+    """
+
+    normalized = _normalize_for_match(root_path)
+    digest = hashlib.sha256(normalized.encode("utf-8")).hexdigest()[:8]
+    return f"{sanitize_path_component(os.path.basename(normalized))}-{digest}"
+
+
+def resolve_centralized_dir(model_path: str) -> Optional[str]:
+    """Return the centralized mirror directory for ``model_path``.
+
+    The mirror layout is
+    ``<sidecar_root>/<library>/<root_basename-roothash>/<rel_dir>``
+    where ``rel_dir`` is the model's directory relative to the model root that
+    contains it. The longest matching root wins so nested roots resolve to the
+    most specific mirror. Returns ``None`` when centralized storage is inactive
+    or the path is not under any configured model root.
+    """
+
+    return resolve_centralized_dir_for_dir(
+        os.path.dirname(_normalize_for_match(model_path))
+    )
+
+
+def resolve_centralized_dir_for_dir(
+    model_dir: str, *, sidecar_root: Optional[str] = None
+) -> Optional[str]:
+    """Return the centralized mirror directory for a model *directory*.
+
+    Same layout as :func:`resolve_centralized_dir`, but accepts the directory
+    itself. Used by folder-level operations (folder rename, mirror-tree walks)
+    that have no model file path to derive from. Passing a configured model
+    root returns the mirror base for that root.
+
+    ``sidecar_root`` overrides the root lookup; pass
+    :func:`get_configured_sidecar_root` to resolve mirror paths independently
+    of the active storage mode (migration tooling).
+    """
+
+    root = sidecar_root if sidecar_root is not None else get_sidecar_root()
+    if not root:
+        return None
+
+    normalized_dir = _normalize_for_match(model_dir)
+
+    best_root: Optional[str] = None
+    for candidate in _iter_model_roots():
+        if not candidate:
+            continue
+        normalized = _normalize_for_match(candidate)
+        if normalized_dir == normalized or normalized_dir.startswith(normalized + os.sep):
+            if best_root is None or len(normalized) > len(best_root):
+                best_root = normalized
+
+    if best_root is None:
+        return None
+
+    try:
+        from ..services.settings_manager import get_settings_manager
+
+        library = get_settings_manager().get_active_library_name() or "default"
+    except Exception:  # pragma: no cover - defensive fallback
+        library = "default"
+
+    rel_dir = os.path.relpath(normalized_dir, best_root)
+    parts = [root, sanitize_path_component(library), root_mirror_component(best_root)]
+    if rel_dir and rel_dir != os.curdir:
+        parts.extend(sanitize_path_component(part) for part in rel_dir.split(os.sep) if part not in ("", os.curdir))
+    return os.path.join(*parts)
+
+
+def get_sidecar_dir(model_path: str) -> str:
+    """Return the directory holding the model's sidecar/preview assets.
+
+    Centralized mode falls back to the model's own directory (with a warning)
+    when the path lies outside every configured model root.
+    """
+
+    if get_storage_mode() == STORAGE_MODE_CENTRALIZED:
+        mirror = resolve_centralized_dir(model_path)
+        if mirror:
+            return mirror
+        logger.warning(
+            "sidecar_paths: %s is outside configured model roots; storing sidecar alongside",
+            model_path,
+        )
+    return os.path.dirname(os.path.abspath(model_path))
+
+
+def get_metadata_path(model_path: str) -> str:
+    """Return the ``.metadata.json`` sidecar path for a model file."""
+
+    base_name = os.path.splitext(os.path.basename(model_path))[0] + METADATA_SUFFIX
+    return os.path.join(get_sidecar_dir(model_path), base_name)
+
+
+def is_metadata_path(path: str) -> bool:
+    """Return True when ``path`` already points at a metadata sidecar file."""
+
+    return path.endswith(METADATA_SUFFIX)
+
+
+def resolve_metadata_path(path: str) -> str:
+    """Accept either a model path or a sidecar path and return the sidecar path."""
+
+    if is_metadata_path(path):
+        return path
+    return get_metadata_path(path)
+
+
+def get_preview_dir(model_path: str) -> str:
+    """Return the directory holding the model's preview assets."""
+
+    return get_sidecar_dir(model_path)