fix(cache): prevent corrupted cache rows from breaking model listings (#730)

Cache corruption (NULL model_name/file_name from legacy DB rows or partial
writes) caused format_response to raise KeyError/AttributeError, failing the
entire /loras/list request and showing no models in the UI.

Fix across three layers:
- format_response (lora/checkpoint/embedding): replace direct dict[] access
  with .get() fallbacks; return None for entries missing file_path
- handlers: filter None entries from list/excluded/fetch/duplicate/conflict
  endpoints instead of letting them crash or appear as null in responses
- model_scanner: always use validate_batch repaired copies (previously
  discarded when no invalid entries, leaving None values in raw_data)
- persistent_model_cache: add or-empty-string guards on read and write for
  nullable TEXT columns (model_name, file_name, folder, base_model, etc.)

.gitignore

@@ -7,6 +7,10 @@ py/run_test.py
 .vscode/
 cache/
 civitai/
+stats/
+wildcards/
+backups/
+logs/
 node_modules/
 coverage/
 .coverage

py/routes/handlers/model_handlers.py

@@ -203,11 +203,17 @@ class ModelListingHandler:
             result = await self._service.get_paginated_data(**params)
 
             format_start = time.perf_counter()
+            formatted_raw = [
+                await self._service.format_response(entry)
+                for entry in result["items"]
+            ]
+            # Filter out None entries returned for corrupted cache rows (issue #730).
+            # Note: "total" intentionally remains the pre-filter count to reflect
+            # the true number of models in the cache; corrupted entries are rare
+            # and adjusting total would cause pagination drift on every page.
+            formatted_items = [item for item in formatted_raw if item is not None]
             formatted_result = {
-                "items": [
+                "items": formatted_items,
-                    await self._service.format_response(item)
-                    for item in result["items"]
-                ],
                 "total": result["total"],
                 "page": result["page"],
                 "page_size": result["page_size"],
@@ -238,11 +244,15 @@ class ModelListingHandler:
             result = await self._service.get_excluded_paginated_data(**params)
 
             format_start = time.perf_counter()
+            formatted_raw = [
+                await self._service.format_response(entry)
+                for entry in result["items"]
+            ]
+            # Filter out None entries returned for corrupted cache rows (issue #730).
+            # "total" stays at the pre-filter count; see get_models for rationale.
+            formatted_items = [item for item in formatted_raw if item is not None]
             formatted_result = {
-                "items": [
+                "items": formatted_items,
-                    await self._service.format_response(item)
-                    for item in result["items"]
-                ],
                 "total": result["total"],
                 "page": result["page"],
                 "page_size": result["page_size"],
@@ -533,8 +543,13 @@ class ModelManagementHandler:
             if not success:
                 return web.json_response({"success": False, "error": error})
 
-            formatted_metadata = await self._service.format_response(model_data)
+            formatted = await self._service.format_response(model_data)
-            return web.json_response({"success": True, "metadata": formatted_metadata})
+            if formatted is None:
+                return web.json_response(
+                    {"success": False, "error": "Model entry is corrupted (missing file_path)"},
+                    status=500,
+                )
+            return web.json_response({"success": True, "metadata": formatted})
         except Exception as exc:
             if is_expected_offline_error(str(exc)):
                 return web.json_response(
@@ -1091,10 +1106,12 @@ class ModelQueryHandler:
                 # Sort: originals first, copies last
                 sorted_models = self._sort_duplicate_group(filtered)
 
-                # Format response
+                # Format response, filtering out corrupted entries (issue #730)
                 group = {"hash": sha256, "models": []}
                 for model in sorted_models:
-                    group["models"].append(await self._service.format_response(model))
+                    formatted = await self._service.format_response(model)
+                    if formatted is not None:
+                        group["models"].append(formatted)
 
                 # Only include groups with 2+ models after filtering
                 if len(group["models"]) > 1:
@@ -1211,9 +1228,9 @@ class ModelQueryHandler:
                         (m for m in cache.raw_data if m["file_path"] == path), None
                     )
                     if model:
-                        group["models"].append(
+                        formatted = await self._service.format_response(model)
-                            await self._service.format_response(model)
+                        if formatted is not None:
-                        )
+                            group["models"].append(formatted)
                 hash_val = self._service.scanner.get_hash_by_filename(filename)
                 if hash_val:
                     main_path = self._service.get_path_by_hash(hash_val)
@@ -1223,9 +1240,9 @@ class ModelQueryHandler:
                             None,
                         )
                         if main_model:
-                            group["models"].insert(
+                            formatted = await self._service.format_response(main_model)
-                                0, await self._service.format_response(main_model)
+                            if formatted is not None:
-                            )
+                                group["models"].insert(0, formatted)
                 if group["models"]:
                     result.append(group)
             return web.json_response(

py/routes/update_routes.py

@@ -16,6 +16,27 @@ logger = logging.getLogger(__name__)
 
 NETWORK_EXCEPTIONS = (ClientError, OSError, asyncio.TimeoutError)
 
+# User-managed directories that live inside the plugin folder (portable
+# mode) and must survive a Git-based update. ``git clean -fd`` would
+# otherwise delete them because they are untracked and, in released tags,
+# not listed in ``.gitignore``. ``-e`` excludes a path from cleaning
+# regardless of whether it is ignored.
+_PRESERVE_DIRS = ('settings.json', 'civitai', 'wildcards', 'backups', 'stats', 'logs', 'cache', 'model_cache')
+
+
+def _clean_excludes() -> List[str]:
+    """Build the ``-e`` arguments for ``git clean`` from :data:`_PRESERVE_DIRS`."""
+    excludes: List[str] = []
+    for name in _PRESERVE_DIRS:
+        excludes.append('-e')
+        excludes.append(name)
+        # For directories, also exclude nested matches explicitly
+        # (``-e dir`` alone matches the dir entry; ``-e dir/**`` guards
+        # contents under all git versions as defense-in-depth).
+        excludes.append('-e')
+        excludes.append(f'{name}/**')
+    return excludes
+
 
 class UpdateRoutes:
     """Routes for handling plugin update checks"""
@@ -365,6 +386,8 @@ class UpdateRoutes:
             )
             return False, ""
 
+        clean_excludes = _clean_excludes()
+
         try:
             # Open the Git repository
             repo = git.Repo(plugin_root)
@@ -376,8 +399,9 @@ class UpdateRoutes:
             if nightly:
                 # Reset to discard any local changes
                 repo.git.reset('--hard')
-                # Clean untracked files
+                # Clean untracked files, but preserve user-managed directories
-                repo.git.clean('-fd')
+                # (wildcards, backups, stats, civitai, caches, settings.json).
+                repo.git.clean('-fd', *clean_excludes)
                 
                 # Switch to main branch and pull latest
                 main_branch = 'main'
@@ -394,8 +418,9 @@ class UpdateRoutes:
             else:
                 # Reset to discard any local changes
                 repo.git.reset('--hard')
-                # Clean untracked files
+                # Clean untracked files, but preserve user-managed directories
-                repo.git.clean('-fd')
+                # (wildcards, backups, stats, civitai, caches, settings.json).
+                repo.git.clean('-fd', *clean_excludes)
                 
                 # Get latest release tag
                 tags = sorted(repo.tags, key=lambda t: t.commit.committed_datetime, reverse=True)

py/services/base_model_service.py

@@ -791,8 +791,12 @@ class BaseModelService(ABC):
         }
 
     @abstractmethod
-    async def format_response(self, model_data: Dict) -> Dict:
+    async def format_response(self, model_data: Dict) -> Optional[Dict]:
-        """Format model data for API response - must be implemented by subclasses"""
+        """Format model data for API response - must be implemented by subclasses.
+
+        Subclasses should return None for corrupted entries so the handler
+        layer can filter them out. See issue #730.
+        """
         pass
 
     # Common service methods that delegate to scanner

py/services/checkpoint_service.py

@@ -1,6 +1,6 @@
 import os
 import logging
-from typing import Dict
+from typing import Dict, Optional
 
 from .base_model_service import BaseModelService
 from .auto_tag_service import extract_auto_tags
@@ -21,20 +21,37 @@ class CheckpointService(BaseModelService):
         """
         super().__init__("checkpoint", scanner, CheckpointMetadata, update_service=update_service)
     
-    async def format_response(self, checkpoint_data: Dict) -> Dict:
+    async def format_response(self, checkpoint_data: Dict) -> Optional[Dict]:
-        """Format Checkpoint data for API response"""
+        """Format Checkpoint data for API response.
+
+        Returns None when the entry is missing critical fields (corrupted cache
+        row), so the handler layer can filter it out. See issue #730.
+        """
+        # Guard against corrupted cache entries missing critical fields
+        file_path = checkpoint_data.get("file_path")
+        if not file_path or not isinstance(file_path, str):
+            logger.warning(
+                "Skipping corrupted checkpoint entry (missing file_path): %s",
+                checkpoint_data.get("file_name", "<unknown>"),
+            )
+            return None
+
         # Get sub_type from cache entry (new canonical field)
         sub_type = checkpoint_data.get("sub_type", "checkpoint")
 
+        file_name = checkpoint_data.get("file_name") or ""
+        model_name = checkpoint_data.get("model_name") or file_name
+        folder = checkpoint_data.get("folder") or ""
+
         return {
-            "model_name": checkpoint_data["model_name"],
+            "model_name": model_name,
-            "file_name": checkpoint_data["file_name"],
+            "file_name": file_name,
             "preview_url": config.get_preview_static_url(checkpoint_data.get("preview_url", "")),
             "preview_nsfw_level": checkpoint_data.get("preview_nsfw_level", 0),
             "base_model": checkpoint_data.get("base_model", ""),
-            "folder": checkpoint_data["folder"],
+            "folder": folder,
             "sha256": checkpoint_data.get("sha256", ""),
-            "file_path": checkpoint_data["file_path"].replace(os.sep, "/"),
+            "file_path": file_path.replace(os.sep, "/"),
             "file_size": checkpoint_data.get("size", 0),
             "modified": checkpoint_data.get("modified", ""),
             "tags": checkpoint_data.get("tags", []),

py/services/embedding_service.py

@@ -1,6 +1,6 @@
 import os
 import logging
-from typing import Dict
+from typing import Dict, Optional
 
 from .base_model_service import BaseModelService
 from .auto_tag_service import extract_auto_tags
@@ -21,20 +21,37 @@ class EmbeddingService(BaseModelService):
         """
         super().__init__("embedding", scanner, EmbeddingMetadata, update_service=update_service)
     
-    async def format_response(self, embedding_data: Dict) -> Dict:
+    async def format_response(self, embedding_data: Dict) -> Optional[Dict]:
-        """Format Embedding data for API response"""
+        """Format Embedding data for API response.
+
+        Returns None when the entry is missing critical fields (corrupted cache
+        row), so the handler layer can filter it out. See issue #730.
+        """
+        # Guard against corrupted cache entries missing critical fields
+        file_path = embedding_data.get("file_path")
+        if not file_path or not isinstance(file_path, str):
+            logger.warning(
+                "Skipping corrupted embedding entry (missing file_path): %s",
+                embedding_data.get("file_name", "<unknown>"),
+            )
+            return None
+
         # Get sub_type from cache entry (new canonical field)
         sub_type = embedding_data.get("sub_type", "embedding")
 
+        file_name = embedding_data.get("file_name") or ""
+        model_name = embedding_data.get("model_name") or file_name
+        folder = embedding_data.get("folder") or ""
+
         return {
-            "model_name": embedding_data["model_name"],
+            "model_name": model_name,
-            "file_name": embedding_data["file_name"],
+            "file_name": file_name,
             "preview_url": config.get_preview_static_url(embedding_data.get("preview_url", "")),
             "preview_nsfw_level": embedding_data.get("preview_nsfw_level", 0),
             "base_model": embedding_data.get("base_model", ""),
-            "folder": embedding_data["folder"],
+            "folder": folder,
             "sha256": embedding_data.get("sha256", ""),
-            "file_path": embedding_data["file_path"].replace(os.sep, "/"),
+            "file_path": file_path.replace(os.sep, "/"),
             "file_size": embedding_data.get("size", 0),
             "modified": embedding_data.get("modified", ""),
             "tags": embedding_data.get("tags", []),

py/services/lora_service.py

@@ -24,23 +24,41 @@ class LoraService(BaseModelService):
         """
         super().__init__("lora", scanner, LoraMetadata, update_service=update_service)
 
-    async def format_response(self, lora_data: Dict) -> Dict:
+    async def format_response(self, lora_data: Dict) -> Optional[Dict]:
-        """Format LoRA data for API response"""
+        """Format LoRA data for API response.
+
+        Returns None when the entry is missing critical fields (corrupted cache
+        row), so the handler layer can filter it out instead of crashing the
+        whole listing request. See issue #730.
+        """
+        # Guard against corrupted cache entries missing critical fields
+        file_path = lora_data.get("file_path")
+        if not file_path or not isinstance(file_path, str):
+            logger.warning(
+                "Skipping corrupted LoRA entry (missing file_path): %s",
+                lora_data.get("file_name", "<unknown>"),
+            )
+            return None
+
         # Resolve sub_type using priority: sub_type > model_type > civitai.model.type > default
         # Normalize to lowercase for consistent API responses
         sub_type = resolve_sub_type(lora_data).lower()
 
+        file_name = lora_data.get("file_name") or ""
+        model_name = lora_data.get("model_name") or file_name
+        folder = lora_data.get("folder") or ""
+
         return {
-            "model_name": lora_data["model_name"],
+            "model_name": model_name,
-            "file_name": lora_data["file_name"],
+            "file_name": file_name,
             "preview_url": config.get_preview_static_url(
                 lora_data.get("preview_url", "")
             ),
             "preview_nsfw_level": lora_data.get("preview_nsfw_level", 0),
             "base_model": lora_data.get("base_model", ""),
-            "folder": lora_data["folder"],
+            "folder": folder,
             "sha256": lora_data.get("sha256", ""),
-            "file_path": lora_data["file_path"].replace(os.sep, "/"),
+            "file_path": file_path.replace(os.sep, "/"),
             "file_size": lora_data.get("size", 0),
             "modified": lora_data.get("modified", ""),
             "tags": lora_data.get("tags", []),

py/services/model_scanner.py

@@ -476,11 +476,20 @@ class ModelScanner:
             for tag in adjusted_item.get('tags') or []:
                 tags_count[tag] = tags_count.get(tag, 0) + 1
 
-        # Validate cache entries and check health
+        # Validate cache entries and check health.
+        # Always use the validated/repaired entries — even when there are no
+        # invalid entries, auto_repair may have filled in missing optional
+        # fields (model_name, file_name, folder) with safe defaults on a copied
+        # working_entry.  Without this unconditional replacement the repaired
+        # copies are discarded and None values propagate to format_response.
+        # See issue #730.
         valid_entries, invalid_entries = CacheEntryValidator.validate_batch(
             adjusted_raw_data, auto_repair=True
         )
 
+        # Always use the validated entries (repaired copies)
+        adjusted_raw_data = valid_entries
+
         if invalid_entries:
             monitor = CacheHealthMonitor()
             report = monitor.check_health(adjusted_raw_data, auto_repair=True)

py/services/persistent_model_cache.py

@@ -165,8 +165,8 @@ class PersistentModelCache:
 
             item = {
                 "file_path": file_path,
-                "file_name": row["file_name"],
+                "file_name": row["file_name"] or "",
-                "model_name": row["model_name"],
+                "model_name": row["model_name"] or "",
                 "folder": row["folder"] or "",
                 "size": row["size"] or 0,
                 "modified": row["modified"] or 0.0,
@@ -548,19 +548,19 @@ class PersistentModelCache:
         return (
             model_type,
             item.get("file_path"),
-            item.get("file_name"),
+            item.get("file_name") or "",
-            item.get("model_name"),
+            item.get("model_name") or "",
-            item.get("folder"),
+            item.get("folder") or "",
             int(item.get("size") or 0),
             float(item.get("modified") or 0.0),
             (item.get("sha256") or "").lower() or None,
-            item.get("base_model"),
+            item.get("base_model") or "",
-            item.get("preview_url"),
+            item.get("preview_url") or "",
             int(item.get("preview_nsfw_level") or 0),
             1 if item.get("from_civitai", True) else 0,
             1 if item.get("favorite") else 0,
-            item.get("notes"),
+            item.get("notes") or "",
-            item.get("usage_tips"),
+            item.get("usage_tips") or "",
             metadata_source,
             civitai.get("id"),
             civitai.get("modelId"),

py/services/settings_manager.py

@@ -1568,7 +1568,7 @@ class SettingsManager:
         previous_dir = os.path.dirname(previous_path) or target_dir
 
         if os.path.abspath(previous_path) != os.path.abspath(target_path):
-            self._copy_model_cache_directory(previous_dir, target_dir)
+            self._migrate_settings_directory_content(previous_dir, target_dir)
             logger.info("Switching settings file to: %s", target_path)
 
         self._pending_portable_switch = {"other_path": other_path}
@@ -1603,46 +1603,52 @@ class SettingsManager:
         finally:
             self._pending_portable_switch = None
 
-    def _copy_model_cache_directory(self, source_dir: str, target_dir: str) -> None:
+    def _migrate_settings_directory_content(
-        """Copy model_cache artifacts when switching storage locations."""
+        self, source_dir: str, target_dir: str
+    ) -> None:
+        """Migrate settings directory subdirectories when switching storage locations.
+
+        Copies the canonical subdirectories (cache, backups, logs, stats, wildcards)
+        from the old settings directory to the new one. Legacy cache artifacts
+        (model_cache, recipe_cache, etc.) are migrated lazily by
+        ``resolve_cache_path_with_migration`` on first access.
+
+        Args:
+            source_dir: The previous settings directory path.
+            target_dir: The new settings directory path.
+        """
 
         if not source_dir or not target_dir:
             return
 
-        source_cache_dir = os.path.join(source_dir, "model_cache")
+        def _copy_dir(name: str) -> None:
-        target_cache_dir = os.path.join(target_dir, "model_cache")
+            source = os.path.join(source_dir, name)
-        if os.path.isdir(source_cache_dir) and os.path.abspath(
+            target = os.path.join(target_dir, name)
-            source_cache_dir
+            if os.path.isdir(source) and os.path.abspath(source) != os.path.abspath(
-        ) != os.path.abspath(target_cache_dir):
+                target
-            try:
+            ):
-                shutil.copytree(
+                try:
-                    source_cache_dir,
+                    shutil.copytree(
-                    target_cache_dir,
+                        source,
-                    dirs_exist_ok=True,
+                        target,
-                    ignore=shutil.ignore_patterns("*.sqlite-shm", "*.sqlite-wal"),
+                        dirs_exist_ok=True,
-                )
+                        ignore=shutil.ignore_patterns("*.sqlite-shm", "*.sqlite-wal"),
-            except Exception as exc:
+                    )
-                logger.warning(
+                except Exception as exc:
-                    "Failed to copy model_cache directory from %s to %s: %s",
+                    logger.warning(
-                    source_cache_dir,
+                        "Failed to copy directory %s from %s to %s: %s",
-                    target_cache_dir,
+                        name,
-                    exc,
+                        source,
-                )
+                        target,
+                        exc,
+                    )
 
-        source_cache_file = os.path.join(source_dir, "model_cache.sqlite")
+        # Managed subdirectories under settings_dir
-        target_cache_file = os.path.join(target_dir, "model_cache.sqlite")
+        _copy_dir("cache")
-        if os.path.isfile(source_cache_file) and os.path.abspath(
+        _copy_dir("backups")
-            source_cache_file
+        _copy_dir("logs")
-        ) != os.path.abspath(target_cache_file):
+        _copy_dir("stats")
-            try:
+        _copy_dir("wildcards")
-                shutil.copy2(source_cache_file, target_cache_file)
-            except Exception as exc:
-                logger.warning(
-                    "Failed to copy model_cache.sqlite from %s to %s: %s",
-                    source_cache_file,
-                    target_cache_file,
-                    exc,
-                )
 
     def _get_user_config_directory(self) -> str:
         """Return the user configuration directory, falling back to ~/.config."""

tests/routes/test_base_model_routes_smoke.py

@@ -201,6 +201,45 @@ def test_list_models_returns_formatted_items(mock_service, mock_scanner):
     asyncio.run(scenario())
 
 
+def test_list_models_filters_out_corrupted_entries(mock_service, mock_scanner):
+    """Corrupted cache entries (format_response returns None) must not appear
+    in the response items nor cause a 500. See issue #730.
+    """
+    mock_service.paginated_items = [
+        {"file_path": "/tmp/good.safetensors", "name": "Good"},
+        {"file_path": None, "name": "Corrupted"},  # triggers None from format_response
+        {"file_path": "/tmp/also_good.safetensors", "name": "AlsoGood"},
+    ]
+
+    # Override format_response to return None for corrupted entries
+    original_format = mock_service.format_response
+
+    async def conditional_format(item):
+        if item.get("file_path") is None:
+            return None
+        return await original_format(item)
+
+    mock_service.format_response = conditional_format
+
+    async def scenario():
+        client = await create_test_client(mock_service)
+        try:
+            response = await client.get("/api/lm/test-models/list")
+            payload = await response.json()
+
+            assert response.status == 200
+            # Only the 2 non-corrupted entries should appear
+            assert len(payload["items"]) == 2
+            assert payload["items"][0]["name"] == "Good"
+            assert payload["items"][1]["name"] == "AlsoGood"
+            # None should never appear in the items list
+            assert None not in payload["items"]
+        finally:
+            await client.close()
+
+    asyncio.run(scenario())
+
+
 def test_model_types_endpoint_returns_counts(mock_service, mock_scanner):
     mock_service.model_types = [
         {"type": "LoRa", "count": 3},

tests/routes/test_update_routes.py

@@ -59,3 +59,180 @@ async def test_get_nightly_version_network_error_logs_warning(monkeypatch, caplo
     assert changelog == []
     assert "Unable to reach GitHub for nightly version" in caplog.text
     assert "Traceback" not in caplog.text
+
+
+def test_clean_excludes_covers_user_data_dirs():
+    """git clean must receive -e excludes for every user-managed dir."""
+    excludes = update_routes._clean_excludes()
+    assert "-e" in excludes  # at least one exclude flag present
+    for name in update_routes._PRESERVE_DIRS:
+        assert name in excludes
+        assert f"{name}/**" in excludes
+
+
+@pytest.mark.asyncio
+async def test_perform_git_update_preserves_user_dirs(monkeypatch, tmp_path):
+    """``git clean`` must be called with -e excludes for user data dirs.
+
+    Regression test for portable-mode updates wiping wildcards/, stats/,
+    backups/, etc. because ``git clean -fd`` removed untracked, non-ignored
+    directories.
+    """
+    calls = []
+
+    class FakeGit:
+        def reset(self, *args, **kwargs):
+            calls.append(("reset", args))
+
+        def clean(self, *args, **kwargs):
+            calls.append(("clean", args))
+
+        def checkout(self, *args, **kwargs):
+            calls.append(("checkout", args))
+
+    class FakeRemote:
+        def fetch(self):
+            calls.append(("fetch", ()))
+
+        def pull(self, *args, **kwargs):
+            calls.append(("pull", args))
+
+    class FakeRemotes:
+        origin = FakeRemote()
+
+    class FakeCommit:
+        hexsha = "abcdef123456"
+
+    class FakeHeads:
+        def __getitem__(self, name):
+            class Head:
+                def checkout(self_inner):
+                    calls.append(("head-checkout", (name,)))
+            return Head()
+
+    class FakeBranches:
+        names = ["main"]
+
+        def __iter__(self):
+            class B:
+                name = "main"
+            return iter([B()])
+
+    class FakeRepo:
+        def __init__(self, path):
+            calls.append(("repo", (path,)))
+
+        git = FakeGit()
+        remotes = FakeRemotes()
+        head = type("H", (), {"commit": FakeCommit()})()
+        branches = FakeBranches()
+        heads = FakeHeads()
+
+        def create_head(self, name, ref):
+            calls.append(("create_head", (name, ref)))
+
+    class FakeGitModule:
+        class Repo:
+            def __new__(cls, path):
+                return FakeRepo(path)
+
+        class exc:
+            class GitError(Exception):
+                pass
+
+    import builtins
+
+    real_import = builtins.__import__
+
+    def fake_import(name, *args, **kwargs):
+        if name == "git":
+            return FakeGitModule
+        return real_import(name, *args, **kwargs)
+
+    monkeypatch.setattr(builtins, "__import__", fake_import)
+
+    success, version = await update_routes.UpdateRoutes._perform_git_update(
+        str(tmp_path), nightly=True
+    )
+
+    assert success is True
+    clean_calls = [c for c in calls if c[0] == "clean"]
+    assert len(clean_calls) == 1
+    clean_args = clean_calls[0][1]
+    # Every preserved dir must be excluded via -e
+    for name in update_routes._PRESERVE_DIRS:
+        assert name in clean_args, f"{name} missing from git clean excludes"
+        assert f"{name}/**" in clean_args, f"{name}/** missing from git clean excludes"
+        # Ensure there's an -e before each name occurrence
+        idx = clean_args.index(name)
+        assert clean_args[idx - 1] == "-e"
+
+
+@pytest.mark.asyncio
+async def test_perform_git_update_stable_preserves_user_dirs(monkeypatch, tmp_path):
+    """Stable (tag) update path must also pass -e excludes to git clean."""
+    calls = []
+
+    class FakeGit:
+        def reset(self, *args, **kwargs):
+            calls.append(("reset", args))
+
+        def clean(self, *args, **kwargs):
+            calls.append(("clean", args))
+
+        def checkout(self, *args, **kwargs):
+            calls.append(("checkout", args))
+
+    class FakeRemote:
+        def fetch(self):
+            calls.append(("fetch", ()))
+
+    class FakeRemotes:
+        origin = FakeRemote()
+
+    class FakeCommit:
+        committed_datetime = "2026-01-01"
+
+    class FakeTag:
+        name = "v9.9.9"
+        commit = FakeCommit()
+
+    class FakeRepo:
+        def __init__(self, path):
+            calls.append(("repo", (path,)))
+
+        git = FakeGit()
+        remotes = FakeRemotes()
+        tags = [FakeTag()]
+
+    class FakeGitModule:
+        class Repo:
+            def __new__(cls, path):
+                return FakeRepo(path)
+
+        class exc:
+            class GitError(Exception):
+                pass
+
+    import builtins
+
+    real_import = builtins.__import__
+
+    def fake_import(name, *args, **kwargs):
+        if name == "git":
+            return FakeGitModule
+        return real_import(name, *args, **kwargs)
+
+    monkeypatch.setattr(builtins, "__import__", fake_import)
+
+    success, version = await update_routes.UpdateRoutes._perform_git_update(
+        str(tmp_path), nightly=False
+    )
+
+    assert success is True
+    assert version == "v9.9.9"
+    clean_calls = [c for c in calls if c[0] == "clean"]
+    assert len(clean_calls) == 1
+    clean_args = clean_calls[0][1]
+    for name in update_routes._PRESERVE_DIRS:
+        assert name in clean_args, f"{name} missing from git clean excludes (stable)"

tests/services/test_service_format_response_sub_type.py

@@ -204,3 +204,102 @@ class TestEmbeddingServiceFormatResponse:
 
         assert result["sub_type"] == "embedding"
         assert "model_type" not in result  # Removed in refactoring
+
+
+class TestFormatResponseCorruptedEntries:
+    """Test format_response handles corrupted cache entries gracefully (issue #730).
+
+    When cache rows have None/missing critical fields (e.g. from a partially
+    written or legacy DB), format_response must NOT raise KeyError/AttributeError.
+    Instead it returns None so the handler layer can filter the bad entry out
+    instead of failing the entire listing request.
+    """
+
+    @pytest.fixture
+    def mock_scanner(self):
+        scanner = MagicMock()
+        scanner._hash_index = MagicMock()
+        return scanner
+
+    @pytest.fixture
+    def lora_service(self, mock_scanner):
+        return LoraService(mock_scanner)
+
+    @pytest.fixture
+    def checkpoint_service(self, mock_scanner):
+        return CheckpointService(mock_scanner)
+
+    @pytest.fixture
+    def embedding_service(self, mock_scanner):
+        return EmbeddingService(mock_scanner)
+
+    @pytest.mark.asyncio
+    async def test_lora_returns_none_on_missing_file_path(self, lora_service):
+        """format_response returns None when file_path is missing (corrupted row)."""
+        lora_data = {
+            "model_name": "Test LoRA",
+            "file_name": "test_lora",
+            "file_path": None,  # corrupted: missing file_path
+            "folder": "",
+            "sha256": "abc123",
+            "tags": [],
+            "from_civitai": True,
+            "civitai": {},
+        }
+        result = await lora_service.format_response(lora_data)
+        assert result is None
+
+    @pytest.mark.asyncio
+    async def test_lora_handles_none_model_name_gracefully(self, lora_service):
+        """format_response should not crash when model_name is None (legacy DB row)."""
+        lora_data = {
+            "model_name": None,          # NULL from old DB row
+            "file_name": "test_lora",
+            "file_path": "/models/test_lora.safetensors",
+            "folder": "",
+            "sha256": "abc123",
+            "tags": [],
+            "from_civitai": True,
+            "civitai": {},
+        }
+        result = await lora_service.format_response(lora_data)
+        # Should not raise; model_name falls back to file_name
+        assert result is not None
+        assert result["model_name"] == "test_lora"
+
+    @pytest.mark.asyncio
+    async def test_checkpoint_returns_none_on_missing_file_path(self, checkpoint_service):
+        """format_response returns None when file_path is missing (corrupted row)."""
+        checkpoint_data = {
+            "model_name": "Test",
+            "file_name": "test",
+            "file_path": "",  # empty string == corrupted
+            "folder": "",
+            "sha256": "abc",
+            "tags": [],
+            "from_civitai": True,
+            "civitai": {},
+            "sub_type": "checkpoint",
+        }
+        result = await checkpoint_service.format_response(checkpoint_data)
+        assert result is None
+
+    @pytest.mark.asyncio
+    async def test_embedding_handles_none_fields_gracefully(self, embedding_service):
+        """format_response should not crash when optional fields are None."""
+        embedding_data = {
+            "model_name": None,
+            "file_name": None,
+            "file_path": "/models/test.pt",
+            "folder": None,
+            "sha256": "abc",
+            "tags": [],
+            "from_civitai": True,
+            "civitai": {},
+            "sub_type": "embedding",
+        }
+        result = await embedding_service.format_response(embedding_data)
+        assert result is not None
+        assert result["file_path"] == "/models/test.pt"
+        # model_name falls back to file_name which falls back to ""
+        assert result["model_name"] == ""

tests/services/test_settings_manager.py

@@ -200,52 +200,97 @@ def _setup_storage_paths(tmp_path, monkeypatch):
     return project_root, user_dir, user_settings_path
 
 
-def _populate_cache(root_dir, marker_name, db_text):
+def _populate_settings_dir(root_dir):
-    cache_dir = root_dir / "model_cache"
+    """Create test data for all managed subdirectories under a settings directory."""
-    cache_dir.mkdir(exist_ok=True)
+    (root_dir / "cache" / "symlink").mkdir(parents=True, exist_ok=True)
-    marker_file = cache_dir / marker_name
+    (root_dir / "cache" / "symlink" / "symlink_map.json").write_text(
-    marker_file.write_text(marker_name, encoding="utf-8")
+        '{"migrated": true}', encoding="utf-8"
-    (root_dir / "model_cache.sqlite").write_text(db_text, encoding="utf-8")
+    )
+    (root_dir / "backups").mkdir(parents=True, exist_ok=True)
+    (root_dir / "backups" / "backup_test.zip").write_text(
+        "backup", encoding="utf-8"
+    )
+    (root_dir / "logs").mkdir(parents=True, exist_ok=True)
+    (root_dir / "logs" / "session.log").write_text("log", encoding="utf-8")
+    (root_dir / "stats").mkdir(parents=True, exist_ok=True)
+    (root_dir / "stats" / "stats.json").write_text(
+        '{"stats": true}', encoding="utf-8"
+    )
+    (root_dir / "wildcards").mkdir(parents=True, exist_ok=True)
+    (root_dir / "wildcards" / "test.txt").write_text("wildcard", encoding="utf-8")
 
 
-def test_switch_to_portable_mode_copies_cache(tmp_path, monkeypatch):
+def test_switch_to_portable_mode_copies_subdirectories(tmp_path, monkeypatch):
     project_root, user_dir, user_settings = _setup_storage_paths(tmp_path, monkeypatch)
-    _populate_cache(user_dir, "user_marker.txt", "user_db")
+    _populate_settings_dir(user_dir)
 
     manager = SettingsManager()
 
     manager.set("use_portable_settings", True)
 
     assert manager.settings_file == str(project_root / "settings.json")
-    marker_copy = project_root / "model_cache" / "user_marker.txt"
+    # Managed subdirectories should all be migrated
-    assert marker_copy.read_text(encoding="utf-8") == "user_marker.txt"
+    assert (
-    assert (project_root / "model_cache.sqlite").read_text(
+        project_root / "cache" / "symlink" / "symlink_map.json"
+    ).read_text(encoding="utf-8") == '{"migrated": true}'
+    assert (
+        project_root / "backups" / "backup_test.zip"
+    ).read_text(encoding="utf-8") == "backup"
+    assert (project_root / "logs" / "session.log").read_text(
         encoding="utf-8"
-    ) == "user_db"
+    ) == "log"
+    assert (project_root / "stats" / "stats.json").read_text(
+        encoding="utf-8"
+    ) == '{"stats": true}'
+    assert (project_root / "wildcards" / "test.txt").read_text(
+        encoding="utf-8"
+    ) == "wildcard"
     assert user_settings.exists()
 
 
-def test_switching_back_to_user_config_moves_cache(tmp_path, monkeypatch):
+def test_switching_back_to_user_config_moves_subdirectories(tmp_path, monkeypatch):
     project_root, user_dir, user_settings = _setup_storage_paths(tmp_path, monkeypatch)
-    _populate_cache(user_dir, "user_marker.txt", "user_db")
+    _populate_settings_dir(user_dir)
 
     manager = SettingsManager()
     manager.set("use_portable_settings", True)
 
-    project_cache_dir = project_root / "model_cache"
+    # Populate project-root managed subdirectories
-    project_cache_dir.mkdir(exist_ok=True)
+    (project_root / "cache" / "model").mkdir(parents=True, exist_ok=True)
-    (project_cache_dir / "project_marker.txt").write_text(
+    (project_root / "cache" / "model" / "default.sqlite").write_text(
-        "project_marker", encoding="utf-8"
+        "project_db", encoding="utf-8"
+    )
+    (project_root / "backups" / "project_backup.zip").write_text(
+        "project_backup", encoding="utf-8"
+    )
+    (project_root / "logs" / "project.log").write_text(
+        "project_log", encoding="utf-8"
+    )
+    (project_root / "stats" / "project_stats.json").write_text(
+        '{"project": true}', encoding="utf-8"
+    )
+    (project_root / "wildcards" / "project.txt").write_text(
+        "project_wildcard", encoding="utf-8"
     )
-    (project_root / "model_cache.sqlite").write_text("project_db", encoding="utf-8")
 
     manager.set("use_portable_settings", False)
 
     assert manager.settings_file == str(user_settings)
-    assert (user_dir / "model_cache" / "project_marker.txt").read_text(
+    assert (user_dir / "cache" / "model" / "default.sqlite").read_text(
         encoding="utf-8"
-    ) == "project_marker"
+    ) == "project_db"
-    assert (user_dir / "model_cache.sqlite").read_text(encoding="utf-8") == "project_db"
+    assert (user_dir / "backups" / "project_backup.zip").read_text(
+        encoding="utf-8"
+    ) == "project_backup"
+    assert (user_dir / "logs" / "project.log").read_text(
+        encoding="utf-8"
+    ) == "project_log"
+    assert (user_dir / "stats" / "project_stats.json").read_text(
+        encoding="utf-8"
+    ) == '{"project": true}'
+    assert (user_dir / "wildcards" / "project.txt").read_text(
+        encoding="utf-8"
+    ) == "project_wildcard"
 
 
 def test_download_path_template_parses_json_string(manager):