Compare commits

...
Author SHA1 Message Date
Will Miao b90d60f043 fix(sidecars): carry the identity map through a sidecar root move
A model root that moved earlier is mirrored under a pinned name derived
from its old path. If anything resolved against the new sidecar path before
the relocation ran, the destination got a map naming the mirror after the
*current* path. migrate_root's keep-newer transfer then dropped the source
map, so the moved metadata stayed orphaned under the pinned component while
reads followed the new name and rebuilt defaults — losing favorites, notes
and tags a second time.

The identity map is now relocated by relocate_root_map() instead of the
generic transfer: entries recorded under the old sidecar root win for the
roots they describe, destination-only entries are preserved, and the cache
is dropped so the next resolution reloads the merged map. A merge that
cannot be written is reported as a migration error rather than silently
stranding the moved metadata.

Reported by the Codex review on #1131.
2026-09-28 21:25:48 +08:00
Will Miao c3a9350155 fix(sidecars): keep mirrored sidecars when a model root moves
Centralized sidecars were addressed by a hash of the model root's absolute
path, so moving or renaming a root produced a new mirror directory. The
scanner then found no sidecar there, rebuilt default metadata, and silently
lost favorites, notes, tags and usage tips for every model under that root,
leaving the old metadata orphaned on disk.

Mirrors are now addressed by a root identity pinned in
<sidecar_root>/.lm-sidecar-roots.json. The identity starts as the existing
deterministic <basename>-<path digest> -- so pre-existing mirrors keep
resolving even if the map is lost, and a relocated sidecar root keeps its
names -- and is re-anchored to the root's new path when it moves, matched by
basename and recorded sample directories. Ambiguous matches are never
guessed: the mirror is left untouched and reported.

Also:
- drop the <library> path segment (single-library direction); a legacy
  library prefix is only recognised while adopting an existing mirror, which
  also keeps the unreleased centralized layout usable
- surface stranded mirrors in Doctor and in the log instead of silently
  rebuilding sidecars
- keep the default alongside mode untouched: the identity map is loaded and
  reconciled lazily, only while centralized storage is in use
2026-09-28 21:11:38 +08:00
18 changed files with 1450 additions and 65 deletions
+8 -5
View File
@@ -20,18 +20,21 @@ By default, `.metadata.json` sidecars and preview images live **alongside** thei
| `sidecar_storage_mode` | `"alongside"` \| `"centralized"` | `"alongside"` | | `sidecar_storage_mode` | `"alongside"` \| `"centralized"` | `"alongside"` |
| `sidecar_storage_path` | Absolute path string; empty = `<settings dir>/sidecars` | `""` | | `sidecar_storage_path` | Absolute path string; empty = `<settings dir>/sidecars` | `""` |
In centralized mode, sidecars and previews mirror the library-relative directory structure: In centralized mode, sidecars and previews mirror each model root's directory structure:
``` ```
<sidecar_root>/<library>/<root_basename-roothash>/<rel_dir>/<name>.metadata.json <sidecar_root>/<root_component>/<rel_dir>/<name>.metadata.json
``` ```
- `<library>` is the active library name and `<rel_dir>` the model's directory relative to the model root containing the file. `<root_basename-roothash>` combines the root's basename with a short hash of its full path so two roots sharing a basename (e.g. `/mnt/a/loras` and `/mnt/b/loras`) never collide. Each component is sanitized to filesystem-safe characters. - `<rel_dir>` is the model's directory relative to the model root containing the file; the longest matching root wins, so nested roots mirror under the most specific root.
- `<root_component>` identifies the model root and **survives the root being moved or renamed**. It starts as the deterministic `<sanitized basename>-<path digest>` — so mirrors created by older builds, and mirrors left behind by a relocated sidecar root, still resolve — and is then pinned in `<sidecar_root>/.lm-sidecar-roots.json` alongside the root's last known path and a few sample subdirectories. Two roots sharing a basename (e.g. `/mnt/a/loras` and `/mnt/b/loras`) always get distinct components and never collide. Each path component is sanitized to filesystem-safe characters.
- **Moving or renaming a model root does not strand its sidecars.** On the next run the mirror identity is re-anchored to the root's new path (matched by basename and recorded sample directories), so favorites, notes, tags and usage tips keep resolving. An existing hash-named mirror from an older build is adopted as-is on first use.
- If an identity cannot be re-anchored unambiguously (e.g. two same-named candidate roots), nothing is guessed: the mirror stays on disk untouched and surfaces as an orphan in **Doctor → Centralized Sidecars** (and in the log). Restoring the original root path re-links it automatically.
- `.civitai.info` files always stay next to the model file, in both modes. - `.civitai.info` files always stay next to the model file, in both modes.
- Changing the mode does **not** move existing files automatically — run the migration (`POST /api/lm/sidecars/migrate` with `{"direction": "to_centralized" | "to_alongside"}`, or the "Migrate Sidecars Now" button in settings). The migration covers excluded (hidden) models too, so un-excluding one later never strands its sidecar in the old layout. The result payload includes a `sidecar_root` field with the resolved centralized root, and the settings UI shows the outcome counters plus an "Open Folder" shortcut. - Changing the mode does **not** move existing files automatically — run the migration (`POST /api/lm/sidecars/migrate` with `{"direction": "to_centralized" | "to_alongside"}`, or the "Migrate Sidecars Now" button in settings). The migration covers excluded (hidden) models too, so un-excluding one later never strands its sidecar in the old layout. The result payload includes a `sidecar_root` field with the resolved centralized root, and the settings UI shows the outcome counters plus an "Open Folder" shortcut.
- Changing `sidecar_storage_path` while centralized likewise needs a root relocation: `{"direction": "relocate_root", "old_root": "<previous path>"}` moves the whole mirror tree to the new root (the settings UI offers this automatically). - Changing `sidecar_storage_path` while centralized likewise needs a root relocation: `{"direction": "relocate_root", "old_root": "<previous path>"}` moves the whole mirror tree to the new root (the settings UI offers this automatically). The identity map travels with the tree, and its entries win over any map the destination acquired beforehand — so a mirror that was re-anchored earlier keeps its name even if something resolved against the new path before the relocation ran.
- The settings UI always shows the resolved effective storage root (via the `sidecar_storage_root*` fields in `GET /api/lm/settings`), with `POST /api/lm/sidecars/open-location` opening it in the file manager. When the resolved root lies inside the plugin installation folder (portable settings mode), the UI warns: reinstalling or clean-updating the plugin would delete the sidecars, so an explicit path outside the installation folder is recommended. The repo `.gitignore` excludes the portable-mode default (`/sidecars/`). - The settings UI always shows the resolved effective storage root (via the `sidecar_storage_root*` fields in `GET /api/lm/settings`), with `POST /api/lm/sidecars/open-location` opening it in the file manager. When the resolved root lies inside the plugin installation folder (portable settings mode), the UI warns: reinstalling or clean-updating the plugin would delete the sidecars, so an explicit path outside the installation folder is recommended. The repo `.gitignore` excludes the portable-mode default (`/sidecars/`).
- All sidecar/preview path derivation goes through the helpers in `py/utils/sidecar_paths.py`; never construct paths inline. - All sidecar/preview path derivation goes through the helpers in `py/utils/sidecar_paths.py`; never construct paths inline. In the default `alongside` mode these helpers do no extra I/O at all — the identity map is only loaded and reconciled when centralized storage is actually in use.
--- ---
+3
View File
@@ -2769,6 +2769,9 @@
"filename_conflicts": { "filename_conflicts": {
"title": "Konflikte durch doppelte Dateinamen" "title": "Konflikte durch doppelte Dateinamen"
}, },
"sidecar_mirror_orphans": {
"title": "[TODO: Translate] Centralized Sidecars"
},
"ui_version": { "ui_version": {
"title": "UI-Version" "title": "UI-Version"
} }
+3
View File
@@ -2769,6 +2769,9 @@
"filename_conflicts": { "filename_conflicts": {
"title": "Duplicate Filename Conflicts" "title": "Duplicate Filename Conflicts"
}, },
"sidecar_mirror_orphans": {
"title": "Centralized Sidecars"
},
"ui_version": { "ui_version": {
"title": "UI Version" "title": "UI Version"
} }
+3
View File
@@ -2769,6 +2769,9 @@
"filename_conflicts": { "filename_conflicts": {
"title": "Conflictos de nombres de archivo duplicados" "title": "Conflictos de nombres de archivo duplicados"
}, },
"sidecar_mirror_orphans": {
"title": "[TODO: Translate] Centralized Sidecars"
},
"ui_version": { "ui_version": {
"title": "Versión de la interfaz" "title": "Versión de la interfaz"
} }
+3
View File
@@ -2769,6 +2769,9 @@
"filename_conflicts": { "filename_conflicts": {
"title": "Conflits de noms de fichiers en double" "title": "Conflits de noms de fichiers en double"
}, },
"sidecar_mirror_orphans": {
"title": "[TODO: Translate] Centralized Sidecars"
},
"ui_version": { "ui_version": {
"title": "Version de l'interface" "title": "Version de l'interface"
} }
+3
View File
@@ -2769,6 +2769,9 @@
"filename_conflicts": { "filename_conflicts": {
"title": "התנגשויות שמות קבצים כפולים" "title": "התנגשויות שמות קבצים כפולים"
}, },
"sidecar_mirror_orphans": {
"title": "[TODO: Translate] Centralized Sidecars"
},
"ui_version": { "ui_version": {
"title": "גרסת הממשק" "title": "גרסת הממשק"
} }
+3
View File
@@ -2769,6 +2769,9 @@
"filename_conflicts": { "filename_conflicts": {
"title": "ファイル名重複競合" "title": "ファイル名重複競合"
}, },
"sidecar_mirror_orphans": {
"title": "[TODO: Translate] Centralized Sidecars"
},
"ui_version": { "ui_version": {
"title": "UI バージョン" "title": "UI バージョン"
} }
+3
View File
@@ -2769,6 +2769,9 @@
"filename_conflicts": { "filename_conflicts": {
"title": "파일명 중복 충돌" "title": "파일명 중복 충돌"
}, },
"sidecar_mirror_orphans": {
"title": "[TODO: Translate] Centralized Sidecars"
},
"ui_version": { "ui_version": {
"title": "UI 버전" "title": "UI 버전"
} }
+3
View File
@@ -2769,6 +2769,9 @@
"filename_conflicts": { "filename_conflicts": {
"title": "Конфликты дублирующихся имён файлов" "title": "Конфликты дублирующихся имён файлов"
}, },
"sidecar_mirror_orphans": {
"title": "[TODO: Translate] Centralized Sidecars"
},
"ui_version": { "ui_version": {
"title": "Версия интерфейса" "title": "Версия интерфейса"
} }
+3
View File
@@ -2769,6 +2769,9 @@
"filename_conflicts": { "filename_conflicts": {
"title": "文件名重复冲突" "title": "文件名重复冲突"
}, },
"sidecar_mirror_orphans": {
"title": "[TODO: Translate] Centralized Sidecars"
},
"ui_version": { "ui_version": {
"title": "UI 版本" "title": "UI 版本"
} }
+3
View File
@@ -2769,6 +2769,9 @@
"filename_conflicts": { "filename_conflicts": {
"title": "檔案名稱重複衝突" "title": "檔案名稱重複衝突"
}, },
"sidecar_mirror_orphans": {
"title": "[TODO: Translate] Centralized Sidecars"
},
"ui_version": { "ui_version": {
"title": "UI 版本" "title": "UI 版本"
} }
+68
View File
@@ -75,6 +75,8 @@ from ...utils.sidecar_paths import (
get_configured_sidecar_root, get_configured_sidecar_root,
get_metadata_path, get_metadata_path,
get_preview_dir, get_preview_dir,
get_storage_mode,
get_unmatched_sidecar_components,
) )
from ...utils.usage_stats import UsageStats from ...utils.usage_stats import UsageStats
from .base_model_handlers import BaseModelHandlerSet from .base_model_handlers import BaseModelHandlerSet
@@ -806,6 +808,7 @@ class DoctorHandler:
await self._check_civitai_api_key(), await self._check_civitai_api_key(),
await self._check_cache_health(), await self._check_cache_health(),
await self._check_filename_conflicts(), await self._check_filename_conflicts(),
self._check_sidecar_mirror_orphans(),
self._check_ui_version(client_version, app_version), self._check_ui_version(client_version, app_version),
] ]
@@ -1042,6 +1045,71 @@ class DoctorHandler:
logger.error("Error exporting doctor bundle: %s", exc, exc_info=True) logger.error("Error exporting doctor bundle: %s", exc, exc_info=True)
return web.json_response({"success": False, "error": str(exc)}, status=500) return web.json_response({"success": False, "error": str(exc)}, status=500)
def _check_sidecar_mirror_orphans(self) -> dict[str, Any]:
"""Flag centralized sidecars stranded by a moved/removed model root.
Centralized sidecars live under a per-root mirror directory. A root
that was moved, renamed, or dropped from the configuration leaves its
mirror behind; without this check the loss is silent, because the
scanner simply rebuilds default metadata at the new location.
"""
actions = [{"id": "open-settings", "label": "Open Settings"}]
try:
mode = get_storage_mode()
except Exception as exc: # pragma: no cover - defensive fallback
logger.debug("Doctor: sidecar mode lookup failed: %s", exc)
mode = "alongside"
if mode != "centralized":
return {
"id": "sidecar_mirror_orphans",
"title": "Centralized Sidecars",
"status": "ok",
"summary": "Sidecar metadata is stored alongside the models.",
"details": [],
"actions": actions,
}
try:
orphans = get_unmatched_sidecar_components()
except Exception as exc: # pragma: no cover - defensive fallback
logger.warning("Doctor: sidecar orphan check failed: %s", exc)
orphans = []
if not orphans:
return {
"id": "sidecar_mirror_orphans",
"title": "Centralized Sidecars",
"status": "ok",
"summary": "Every mirrored sidecar directory is linked to a model root.",
"details": [f"Root: {describe_sidecar_root().get('root', '')}"],
"actions": actions,
}
details = [
"Metadata (favorites, notes, tags, usage tips) for these models is on disk but is not being read.",
"This usually means a model root was moved, renamed, or removed. Restore the original root path in Settings; the mirror is re-linked automatically.",
]
details.extend(
f"{item['component']} — last known root: {item['last_path'] or 'unknown'}"
for item in orphans[:5]
)
if len(orphans) > 5:
details.append(f"… and {len(orphans) - 5} more")
return {
"id": "sidecar_mirror_orphans",
"title": "Centralized Sidecars",
"status": "warning",
"summary": (
f"{len(orphans)} sidecar "
f"director{'y' if len(orphans) == 1 else 'ies'} could not be "
"linked to a configured model root."
),
"details": details,
"actions": actions,
}
async def _check_civitai_api_key(self) -> dict[str, Any]: async def _check_civitai_api_key(self) -> dict[str, Any]:
api_key = (self._settings.get("civitai_api_key", "") or "").strip() api_key = (self._settings.get("civitai_api_key", "") or "").strip()
if not api_key: if not api_key:
@@ -5,7 +5,8 @@ Two storage layouts exist (see :mod:`py.utils.sidecar_paths`):
- ``alongside``: ``<model_dir>/<name>.metadata.json`` and preview files live - ``alongside``: ``<model_dir>/<name>.metadata.json`` and preview files live
next to the model file. next to the model file.
- ``centralized``: the same files live under the configured sidecar root, - ``centralized``: the same files live under the configured sidecar root,
mirroring the library-relative directory structure. mirroring each model root's directory structure under a per-root identity
component (see :func:`py.utils.sidecar_paths.root_mirror_component`).
This use case moves the ``.metadata.json`` sidecar and preview files for every 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. known model from one layout to the other. Model files themselves NEVER move.
@@ -51,10 +52,12 @@ from ...utils.file_utils import find_preview_file, get_preview_extension
from ...utils.metadata_manager import MetadataManager from ...utils.metadata_manager import MetadataManager
from ...utils.sidecar_paths import ( from ...utils.sidecar_paths import (
METADATA_SUFFIX, METADATA_SUFFIX,
ROOT_MAP_FILENAME,
STORAGE_MODE_CENTRALIZED, STORAGE_MODE_CENTRALIZED,
get_configured_sidecar_root, get_configured_sidecar_root,
get_sidecar_root, get_sidecar_root,
get_storage_mode, get_storage_mode,
relocate_root_map,
resolve_centralized_dir_for_dir, resolve_centralized_dir_for_dir,
) )
@@ -199,14 +202,19 @@ class SidecarMigrationUseCase:
) )
files: List[Tuple[str, str]] = [] files: List[Tuple[str, str]] = []
source_map_path = os.path.join(old, ROOT_MAP_FILENAME)
if os.path.isdir(old): if os.path.isdir(old):
for dirpath, _dirnames, filenames in os.walk(old): for dirpath, _dirnames, filenames in os.walk(old):
rel = os.path.relpath(dirpath, old) rel = os.path.relpath(dirpath, old)
target_dir = new_root if rel == os.curdir else os.path.join(new_root, rel) target_dir = new_root if rel == os.curdir else os.path.join(new_root, rel)
for filename in filenames: for filename in filenames:
files.append( source = os.path.join(dirpath, filename)
(os.path.join(dirpath, filename), os.path.join(target_dir, filename)) # The identity map is handled by relocate_root_map below:
) # _transfer's keep-newer rule would let a destination map
# written before the relocation displace it.
if source == source_map_path:
continue
files.append((source, os.path.join(target_dir, filename)))
errors: List[Dict[str, str]] = [] errors: List[Dict[str, str]] = []
counters: Dict[str, Any] = {"moved": 0, "conflicts": 0} counters: Dict[str, Any] = {"moved": 0, "conflicts": 0}
@@ -242,6 +250,23 @@ class SidecarMigrationUseCase:
errors.append({"model": os.path.basename(src), "error": str(exc)}) errors.append({"model": os.path.basename(src), "error": str(exc)})
await emit("processing", processed=index, current=os.path.basename(src)) await emit("processing", processed=index, current=os.path.basename(src))
# The identity map names the directories just moved, so it travels with
# them and wins over any map the destination acquired beforehand.
# A failure here strands the moved metadata, so it is a real error.
try:
if not relocate_root_map(old, new_root):
errors.append(
{
"model": ROOT_MAP_FILENAME,
"error": "sidecar root map could not be written to the new root",
}
)
except Exception as exc:
self._logger.error(
"Sidecar root relocation failed for the root map: %s", exc, exc_info=True
)
errors.append({"model": ROOT_MAP_FILENAME, "error": str(exc)})
old_prefix = old.replace(os.sep, "/").rstrip("/") + "/" old_prefix = old.replace(os.sep, "/").rstrip("/") + "/"
new_prefix = new_root.replace(os.sep, "/").rstrip("/") + "/" new_prefix = new_root.replace(os.sep, "/").rstrip("/") + "/"
for sidecar in moved_sidecars: for sidecar in moved_sidecars:
+767 -34
View File
@@ -11,21 +11,39 @@ setting:
file, the historical layout other tools may rely on. file, the historical layout other tools may rely on.
- ``centralized``: sidecars and previews live under a configurable root - ``centralized``: sidecars and previews live under a configurable root
(``sidecar_storage_path`` setting, default ``<settings_dir>/sidecars``), (``sidecar_storage_path`` setting, default ``<settings_dir>/sidecars``),
mirroring the library-relative directory structure: mirroring each model root's 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 <sidecar_root>/<root_component>/<rel_dir>/<name>.metadata.json
on the hot path. Settings lookups go through ``SettingsManager.get`` (a dict
read); config roots come from the already-initialized ``config`` singleton. ``<root_component>`` identifies the model root and **survives the root being
moved or renamed**. It starts as the deterministic ``<sanitized
basename>-<path digest>`` (so pre-existing mirrors, and mirrors left behind
by a relocated sidecar root, still resolve) and is then pinned in
``<sidecar_root>/.lm-sidecar-roots.json`` together with the root's last
known path and a few sample subdirectories. Re-anchoring a remembered
identity to a new path keeps every sidecar under it usable; computing a new
name instead would strand them and silently rebuild default metadata
(losing favorites, notes, tags).
Hot-path behaviour: path resolution is a settings read plus a dict lookup.
The persistent map is loaded and reconciled lazily, at most once per change
of the configured model-root set, and only when centralized storage is
actually in use (``get_sidecar_dir`` short-circuits in alongside mode), so
default installs never read, write, or scan anything new. The root map file
and its directory are only created when there is something to remember.
""" """
from __future__ import annotations from __future__ import annotations
import hashlib import hashlib
import json
import logging import logging
import os import os
import re import re
from typing import List, Optional import threading
import time
import uuid
from typing import Dict, Iterable, List, Optional, Set, Tuple
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -36,6 +54,21 @@ STORAGE_MODE_CENTRALIZED = "centralized"
_VALID_MODES = frozenset({STORAGE_MODE_ALONGSIDE, STORAGE_MODE_CENTRALIZED}) _VALID_MODES = frozenset({STORAGE_MODE_ALONGSIDE, STORAGE_MODE_CENTRALIZED})
# Persistent root-identity map, stored inside the sidecar root so it travels
# with the mirror tree it describes.
ROOT_MAP_FILENAME = ".lm-sidecar-roots.json"
_ROOT_MAP_VERSION = 1
# Bounds for the re-anchor heuristic and the deferred sample persistence.
_MAX_SAMPLE_REL_DIRS = 16
_MAX_MATCH_PROBES = 64
_SAMPLE_SAVE_INTERVAL_SECONDS = 30.0
# Mirror directory names are ``<sanitized basename>-<8 hex>``; the suffix is a
# root id (or the legacy path digest) and is what distinguishes a component
# directory from a pre-identity-map library prefix directory.
_COMPONENT_PATTERN = re.compile(r"^.+-(?:[0-9a-f]{8})$")
def _get_settings_value(key: str, default=None): def _get_settings_value(key: str, default=None):
"""Read a setting defensively; never fail path resolution on settings errors.""" """Read a setting defensively; never fail path resolution on settings errors."""
@@ -184,16 +217,262 @@ def _iter_model_roots() -> List[str]:
def _normalize_for_match(path: str) -> str: def _normalize_for_match(path: str) -> str:
"""Normalize a path for identity comparisons.
Business paths are preserved (no ``realpath``): symlinks are not resolved,
matching the rest of the codebase.
"""
return os.path.normpath(os.path.abspath(path)) return os.path.normpath(os.path.abspath(path))
def root_mirror_component(root_path: str) -> str: def _normalized_roots(raw_roots: Iterable[str]) -> Tuple[str, ...]:
"""Return the mirror path component identifying a model root. """Return the deduplicated, normalized, order-stable model roots."""
``<sanitized basename>-<hash>`` where the hash is a short digest of the seen: Dict[str, None] = {}
normalized absolute root path. Two roots sharing a basename (e.g. for candidate in raw_roots:
``/mnt/a/loras`` and ``/mnt/b/loras``) would otherwise map to the same if not isinstance(candidate, str) or not candidate.strip():
mirror directory and overwrite each other's sidecars. continue
seen.setdefault(_normalize_for_match(candidate), None)
return tuple(seen)
def _best_root_for(
normalized_dir: str, normalized_roots: Tuple[str, ...]
) -> Optional[str]:
"""Return the most specific configured root containing ``normalized_dir``."""
best: Optional[str] = None
for normalized in normalized_roots:
if normalized_dir == normalized or normalized_dir.startswith(
normalized + os.sep
):
if best is None or len(normalized) > len(best):
best = normalized
return best
class _RootMapState:
"""In-memory view of one sidecar root's identity map."""
__slots__ = (
"signature",
"components",
"root_ids",
"entries",
"unmatched",
"loaded",
"dirty_samples",
"last_save",
"persist_disabled",
)
def __init__(self) -> None:
self.signature: Tuple[str, ...] = ()
self.components: Dict[str, str] = {}
self.root_ids: Dict[str, str] = {}
self.entries: Dict[str, Dict[str, object]] = {}
self.unmatched: List[str] = []
self.loaded = False
self.dirty_samples = False
self.last_save = 0.0
self.persist_disabled = False
_ROOT_MAPS: Dict[str, _RootMapState] = {}
_ROOT_MAPS_LOCK = threading.RLock()
def reset_root_map_cache() -> None:
"""Forget every cached root-identity map (tests, storage relocation)."""
with _ROOT_MAPS_LOCK:
_ROOT_MAPS.clear()
def _root_map_path(sidecar_root: str) -> str:
return os.path.join(sidecar_root, ROOT_MAP_FILENAME)
def _load_root_map(sidecar_root: str) -> Dict[str, Dict[str, object]]:
"""Read the persistent map; unreadable or malformed files degrade to empty."""
path = _root_map_path(sidecar_root)
try:
with open(path, "r", encoding="utf-8") as handle:
payload = json.load(handle)
except FileNotFoundError:
return {}
except (OSError, ValueError) as exc:
logger.warning("sidecar_paths: ignoring unreadable root map %s: %s", path, exc)
return {}
if not isinstance(payload, dict):
return {}
raw_roots = payload.get("roots")
if not isinstance(raw_roots, dict):
return {}
entries: Dict[str, Dict[str, object]] = {}
for root_id, raw_entry in raw_roots.items():
if not isinstance(root_id, str) or not root_id:
continue
if not isinstance(raw_entry, dict):
continue
component = raw_entry.get("component")
if not isinstance(component, str) or not component:
continue
samples = raw_entry.get("sample_rel_dirs")
entries[root_id] = {
"component": component,
"basename": raw_entry.get("basename") or "",
"last_path": raw_entry.get("last_path") or "",
"sample_rel_dirs": (
[item for item in samples if isinstance(item, str)][
:_MAX_SAMPLE_REL_DIRS
]
if isinstance(samples, list)
else []
),
}
return entries
def _persistable(sidecar_root: str) -> bool:
"""Return True when the root map could realistically be written.
Walks up to the nearest existing ancestor (the sidecar root itself is
created lazily, alongside the first sidecar) and checks write access. Used
to decide whether identities can be remembered across restarts: when they
cannot, resolution falls back to the deterministic path-derived component
rather than handing out identifiers that would be forgotten.
"""
probe = os.path.abspath(sidecar_root)
while probe and not os.path.exists(probe):
parent = os.path.dirname(probe)
if parent == probe:
return False
probe = parent
return bool(probe) and os.access(probe, os.W_OK)
def _write_root_map(sidecar_root: str, entries: Dict[str, Dict[str, object]]) -> bool:
"""Atomically persist ``entries`` as the root map for ``sidecar_root``."""
path = _root_map_path(sidecar_root)
# Snapshot before serializing: sample directories are appended from the
# resolution path, which is not serialized with the map lock.
payload = {
"version": _ROOT_MAP_VERSION,
"roots": {
root_id: {
"component": entry.get("component", ""),
"basename": entry.get("basename", ""),
"last_path": entry.get("last_path", ""),
"sample_rel_dirs": list(entry.get("sample_rel_dirs") or []),
}
for root_id, entry in entries.items()
},
}
temp_path = f"{path}.tmp"
try:
os.makedirs(sidecar_root, exist_ok=True)
with open(temp_path, "w", encoding="utf-8") as handle:
json.dump(payload, handle, indent=2, ensure_ascii=False)
os.replace(temp_path, path)
except OSError as exc:
logger.warning("sidecar_paths: cannot persist the root map %s: %s", path, exc)
return False
return True
def _save_root_map(sidecar_root: str, state: _RootMapState) -> bool:
"""Persist a reconciled state; disables persistence when it cannot write."""
if state.persist_disabled:
return False
if not _write_root_map(sidecar_root, state.entries):
state.persist_disabled = True
logger.warning(
"sidecar_paths: mirror directory names fall back to path-derived "
"components for %s",
sidecar_root,
)
return False
state.dirty_samples = False
state.last_save = time.monotonic()
return True
def relocate_root_map(source_root: str, destination_root: str) -> bool:
"""Carry the root map from a relocated sidecar root to its destination.
Call this after the mirror tree itself has been moved. Entries recorded
under ``source_root`` win over any identity the destination picked up on
its own: resolving against the new sidecar path *before* the relocation
writes a map that names mirrors after the current model-root path, while
the directories actually being moved are still named after the pinned
identity. Destination-only entries are preserved, and the source file is
always removed so the emptied tree can be pruned.
Returns False only when a source map existed but could not be written to
the destination — the caller must surface that, since the moved metadata
would otherwise be unreachable. Cached state is dropped either way so the
next resolution reloads the merged map.
"""
source_path = _root_map_path(source_root)
source_entries = _load_root_map(source_root)
with _ROOT_MAPS_LOCK:
if source_entries:
destination_path = _root_map_path(destination_root)
merged: Dict[str, Dict[str, object]] = {}
if os.path.exists(destination_path):
merged.update(_load_root_map(destination_root))
source_paths = {
_normalize_for_match(str(entry["last_path"]))
for entry in source_entries.values()
if entry.get("last_path")
}
preserved = {
root_id: entry
for root_id, entry in merged.items()
if not entry.get("last_path")
or _normalize_for_match(str(entry.get("last_path"))) not in source_paths
}
preserved.update(source_entries)
if not _write_root_map(destination_root, preserved):
return False
if os.path.exists(source_path):
try:
os.remove(source_path)
except OSError as exc: # pragma: no cover - defensive cleanup
logger.debug(
"sidecar_paths: cannot remove relocated root map %s: %s",
source_path,
exc,
)
reset_root_map_cache()
return True
def _new_root_id() -> str:
return uuid.uuid4().hex[:8]
def _legacy_component(root_path: str) -> str:
"""Return the deterministic path-derived component for a root.
``<sanitized basename>-<digest of the normalized absolute path>``. This is
both the name mirrors created by older builds already use and the initial
identity for a newly seen root, so a mirror is found again even when the
root map is missing. Once an identity is recorded, the map keeps the name
pinned across root moves.
""" """
normalized = _normalize_for_match(root_path) normalized = _normalize_for_match(root_path)
@@ -201,12 +480,444 @@ def root_mirror_component(root_path: str) -> str:
return f"{sanitize_path_component(os.path.basename(normalized))}-{digest}" return f"{sanitize_path_component(os.path.basename(normalized))}-{digest}"
def _list_component_dirs(sidecar_root: str) -> Set[str]:
"""Return the mirror component directories present under ``sidecar_root``.
Components are normally immediate children. Pre-identity-map builds nested
them one level deeper under a library name
(``<sidecar_root>/<library>/<component>``); those are reported as
``<library>/<component>`` so the legacy layout keeps resolving — the legacy
prefix is treated purely as part of the component name and is never
interpreted as a library.
"""
try:
names = os.listdir(sidecar_root)
except OSError:
return set()
components: Set[str] = set()
unqualified: List[str] = []
for name in names:
if name == ROOT_MAP_FILENAME or name.endswith(".tmp"):
continue
if not os.path.isdir(os.path.join(sidecar_root, name)):
continue
if _looks_like_component(name):
components.add(name)
else:
unqualified.append(name)
for prefix in unqualified:
try:
nested = os.listdir(os.path.join(sidecar_root, prefix))
except OSError:
continue
for name in nested:
if not _looks_like_component(name):
continue
if os.path.isdir(os.path.join(sidecar_root, prefix, name)):
components.add(f"{prefix}/{name}")
return components
def _looks_like_component(name: str) -> bool:
"""True for ``<basename>-<8 hex>`` mirror directory names."""
return bool(_COMPONENT_PATTERN.match(name))
def _subdir_names(directory: str) -> Set[str]:
"""Sanitized names of ``directory``'s immediate subdirectories (bounded)."""
names: Set[str] = set()
try:
with os.scandir(directory) as iterator:
for index, dir_entry in enumerate(iterator):
if index >= _MAX_MATCH_PROBES:
break
try:
if dir_entry.is_dir():
names.add(sanitize_path_component(dir_entry.name))
except OSError:
continue
except OSError:
return set()
return names
def _mirror_dir_overlap(sidecar_root: str, component: str, root: str) -> int:
"""Count sibling directories shared by a mirror component and a root.
Last-resort signal for re-anchoring when the entry carries no recorded
sample directories (e.g. the map was lost): both sides are sanitized the
same way, so renamed-but-similar layouts still overlap.
"""
if not component:
return 0
mirror_names = _subdir_names(os.path.join(sidecar_root, component))
if not mirror_names:
return 0
root_names = _subdir_names(root)
if not root_names:
return 0
return len(mirror_names & root_names)
def _match_score(sidecar_root: str, entry: Dict[str, object], root: str) -> int:
"""Score how likely ``entry`` describes the mirror of ``root``."""
score = 0
if str(entry.get("basename") or "") == os.path.basename(root):
score += 2
samples = entry.get("sample_rel_dirs") or []
if isinstance(samples, list):
for rel in samples[:_MAX_MATCH_PROBES]:
if not isinstance(rel, str) or not rel:
continue
parts = [part for part in rel.split("/") if part not in ("", ".")]
if parts and os.path.isdir(os.path.join(root, *parts)):
score += 1
if not samples:
score += _mirror_dir_overlap(
sidecar_root, str(entry.get("component") or ""), root
)
return score
def _rank_scores(
assignments: List[Tuple[int, str, str]],
) -> Tuple[Dict[str, int], Dict[str, int], Dict[str, int], Dict[str, int]]:
"""Return (best/runner-up score per entry id, best/runner-up per root)."""
best_id: Dict[str, int] = {}
runner_id: Dict[str, int] = {}
best_root: Dict[str, int] = {}
runner_root: Dict[str, int] = {}
for score, root_id, root in assignments:
if score > best_id.get(root_id, 0):
runner_id[root_id] = best_id.get(root_id, 0)
best_id[root_id] = score
elif score > runner_id.get(root_id, 0):
runner_id[root_id] = score
if score > best_root.get(root, 0):
runner_root[root] = best_root.get(root, 0)
best_root[root] = score
elif score > runner_root.get(root, 0):
runner_root[root] = score
return best_id, runner_id, best_root, runner_root
def _due_for_sample_save(state: _RootMapState, now: float) -> bool:
return now - state.last_save >= _SAMPLE_SAVE_INTERVAL_SECONDS
def _reconcile_root_map(
sidecar_root: str,
normalized_roots: Tuple[str, ...],
state: _RootMapState,
) -> None:
"""Match remembered root identities against the configured roots.
Order of preference per root: the identity previously recorded for the
same path, then a deterministic path-derived component already on disk
(pre-identity-map installs), then an orphaned identity re-anchored by
basename/sample-directory scoring, then a unique unclaimed component whose
name matches, and finally a brand-new identity.
"""
entries = (
state.entries
if state.loaded and state.entries
else _load_root_map(sidecar_root)
)
# Copy so partial mutation cannot leak between reconciliation attempts.
entries = {root_id: dict(entry) for root_id, entry in entries.items()}
on_disk = _list_component_dirs(sidecar_root)
persistable = _persistable(sidecar_root)
if not persistable:
state.persist_disabled = True
components: Dict[str, str] = {}
root_ids: Dict[str, str] = {}
claimed: Set[str] = set()
changed = False
by_path: Dict[str, str] = {}
for root_id, entry in entries.items():
last_path = entry.get("last_path")
if isinstance(last_path, str) and last_path.strip():
by_path.setdefault(_normalize_for_match(last_path), root_id)
# 1. Same path as last time: reuse the remembered identity verbatim.
pending: List[str] = []
for root in normalized_roots:
root_id = by_path.get(root)
if root_id is not None:
component = str(entries[root_id]["component"])
if component not in claimed:
components[root] = component
root_ids[root] = root_id
claimed.add(component)
continue
pending.append(root)
if not persistable:
# Identities created here could never be remembered; keep the layout
# deterministic instead of stranding sidecars on the next restart.
for root in normalized_roots:
components.setdefault(root, _legacy_component(root))
state.entries = {}
state.root_ids = {}
state.components = components
state.unmatched = sorted(
component for component in on_disk if component not in set(components.values())
)
state.signature = normalized_roots
state.loaded = True
_warn_unmatched(sidecar_root, state.unmatched)
return
# 2. Adopt a deterministic path-derived component already on disk. Older
# builds nested it under the library name, hence the last-segment match.
still_pending: List[str] = []
for root in pending:
legacy = _legacy_component(root)
matches = [
component
for component in on_disk
if component == legacy
or component.rsplit("/", 1)[-1] == legacy
]
component = matches[0] if len(matches) == 1 and matches[0] not in claimed else None
if component is not None:
root_id = _new_root_id()
entries[root_id] = {
"component": component,
"basename": os.path.basename(root),
"last_path": root,
"sample_rel_dirs": [],
}
components[root] = component
root_ids[root] = root_id
claimed.add(component)
changed = True
else:
still_pending.append(root)
# 3. Re-anchor identities whose recorded path no longer exists.
configured = set(normalized_roots)
assignments: List[Tuple[int, str, str]] = []
for root_id, entry in entries.items():
last_path = str(entry.get("last_path") or "")
component = str(entry["component"])
if not last_path.strip() or component in claimed:
continue
if _normalize_for_match(last_path) in configured:
continue
for root in still_pending:
score = _match_score(sidecar_root, entry, root)
if score > 0:
assignments.append((score, root_id, root))
if assignments:
best_id, runner_id, best_root, runner_root = _rank_scores(assignments)
assignments.sort(key=lambda item: (-item[0], item[1], item[2]))
used_ids: Set[str] = set()
used_roots: Set[str] = set()
for score, root_id, root in assignments:
if root_id in used_ids or root in used_roots:
continue
# Never guess: the pair must be the unique best on both sides.
if score < best_id[root_id] or score < best_root[root]:
continue
if runner_id.get(root_id, 0) == score or runner_root.get(root, 0) == score:
continue
entries[root_id]["last_path"] = root
components[root] = str(entries[root_id]["component"])
root_ids[root] = root_id
claimed.add(str(entries[root_id]["component"]))
used_ids.add(root_id)
used_roots.add(root)
changed = True
still_pending = [root for root in still_pending if root not in used_roots]
# 4/5. Remaining roots: adopt a uniquely matching unclaimed component,
# otherwise mint a fresh identity.
referenced = {str(entry["component"]) for entry in entries.values()}
free_components = [
component
for component in on_disk
if component not in claimed and component not in referenced
]
for root in still_pending:
prefix = sanitize_path_component(os.path.basename(root)) + "-"
candidates = [
component
for component in free_components
if component.rsplit("/", 1)[-1].startswith(prefix)
]
# Default to the deterministic path-derived name. It is stable across
# restarts even if the map is lost, and it matches what a sidecar-root
# relocation or a mirror created by an older build already used; the
# map only overrides it later, to keep the name across a root move.
component = (
candidates[0] if len(candidates) == 1 else _legacy_component(root)
)
if len(candidates) == 1:
free_components.remove(component)
root_id = _new_root_id()
entries[root_id] = {
"component": component,
"basename": os.path.basename(root),
"last_path": root,
"sample_rel_dirs": [],
}
components[root] = component
root_ids[root] = root_id
claimed.add(component)
changed = True
state.entries = entries
state.components = components
state.root_ids = root_ids
state.unmatched = sorted(
component for component in on_disk if component not in claimed
)
state.signature = normalized_roots
state.loaded = True
_warn_unmatched(sidecar_root, state.unmatched)
if changed and not _save_root_map(sidecar_root, state):
_degrade_to_legacy(state, normalized_roots)
def _warn_unmatched(sidecar_root: str, unmatched: List[str]) -> None:
"""Surface mirror directories no configured root claims (never silent)."""
if not unmatched:
return
preview = ", ".join(unmatched[:5])
logger.warning(
"sidecar_paths: %d mirrored sidecar dir(s) under %s are not linked to any "
"configured model root (%s%s). Their metadata stays on disk but is not "
"used until the model root is restored or re-anchored.",
len(unmatched),
sidecar_root,
preview,
", ..." if len(unmatched) > 5 else "",
)
def _degrade_to_legacy(state: _RootMapState, normalized_roots: Tuple[str, ...]) -> None:
"""Fall back to deterministic components after a failed persist."""
state.components = {root: _legacy_component(root) for root in normalized_roots}
state.root_ids = {}
state.entries = {}
state.unmatched = []
state.dirty_samples = False
state.persist_disabled = True
def _ensure_root_map(
sidecar_root: str, normalized_roots: Tuple[str, ...]
) -> _RootMapState:
"""Return the reconciled identity map, reusing the cached one when current."""
key = _normalize_for_match(sidecar_root)
state = _ROOT_MAPS.get(key)
if state is not None and state.loaded and state.signature == normalized_roots:
return state
with _ROOT_MAPS_LOCK:
state = _ROOT_MAPS.get(key)
if state is not None and state.loaded and state.signature == normalized_roots:
return state
if state is None:
state = _RootMapState()
_ROOT_MAPS[key] = state
_reconcile_root_map(sidecar_root, normalized_roots, state)
return state
def _component_for(
best_root: str,
sidecar_root: str,
normalized_roots: Tuple[str, ...],
rel_dir: str,
) -> Optional[str]:
"""Return the pinned mirror component for ``best_root``."""
state = _ensure_root_map(sidecar_root, normalized_roots)
if not state.components.get(best_root):
return _legacy_component(best_root)
_record_sample(sidecar_root, state, best_root, rel_dir)
# Re-read: a failed sample persist degrades the state to deterministic
# names, and this call must agree with every later one.
return state.components.get(best_root) or _legacy_component(best_root)
def _record_sample(
sidecar_root: str, state: _RootMapState, root: str, rel_dir: str
) -> None:
"""Remember a root-relative directory to make future re-anchoring precise."""
root_id = state.root_ids.get(root)
if not root_id:
return
entry = state.entries.get(root_id)
if entry is None:
return
normalized = (rel_dir or "").replace(os.sep, "/").strip("/")
if not normalized or normalized == ".":
return
samples = entry.get("sample_rel_dirs")
if not isinstance(samples, list):
samples = []
entry["sample_rel_dirs"] = samples
if normalized in samples or len(samples) >= _MAX_SAMPLE_REL_DIRS:
return
samples.append(normalized)
state.dirty_samples = True
if state.persist_disabled or not _due_for_sample_save(state, time.monotonic()):
return
with _ROOT_MAPS_LOCK:
if state.persist_disabled or not _due_for_sample_save(state, time.monotonic()):
return
if not _save_root_map(sidecar_root, state):
_degrade_to_legacy(state, state.signature)
def root_mirror_component(root_path: str, *, sidecar_root: Optional[str] = None) -> str:
"""Return the mirror component identifying a model root.
The deterministic ``<sanitized basename>-<path digest>`` until the root map
pins a different name for it — which happens only when a remembered root
was moved or renamed and had to be re-anchored. Two roots sharing a
basename always get distinct components, so their mirrors cannot collide.
"""
resolved_root = sidecar_root if sidecar_root is not None else get_configured_sidecar_root()
normalized = _normalize_for_match(root_path)
if not resolved_root:
return _legacy_component(normalized)
state = _ensure_root_map(resolved_root, _normalized_roots(_iter_model_roots()))
return state.components.get(normalized) or _legacy_component(normalized)
def resolve_centralized_dir(model_path: str) -> Optional[str]: def resolve_centralized_dir(model_path: str) -> Optional[str]:
"""Return the centralized mirror directory for ``model_path``. """Return the centralized mirror directory for ``model_path``.
The mirror layout is The mirror layout is ``<sidecar_root>/<root_component>/<rel_dir>`` where
``<sidecar_root>/<library>/<root_basename-roothash>/<rel_dir>`` ``rel_dir`` is the model's directory relative to the model root that
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 contains it. The longest matching root wins so nested roots resolve to the
most specific mirror. Returns ``None`` when centralized storage is inactive most specific mirror. Returns ``None`` when centralized storage is inactive
or the path is not under any configured model root. or the path is not under any configured model root.
@@ -237,33 +948,55 @@ def resolve_centralized_dir_for_dir(
return None return None
normalized_dir = _normalize_for_match(model_dir) normalized_dir = _normalize_for_match(model_dir)
normalized_roots = _normalized_roots(_iter_model_roots())
best_root: Optional[str] = None best_root = _best_root_for(normalized_dir, normalized_roots)
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: if best_root is None:
return 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) rel_dir = os.path.relpath(normalized_dir, best_root)
parts = [root, sanitize_path_component(library), root_mirror_component(best_root)] component = _component_for(best_root, root, normalized_roots, rel_dir)
if not component:
return None
parts = [root, component]
if rel_dir and rel_dir != os.curdir: 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)) 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) return os.path.join(*parts)
def get_unmatched_sidecar_components() -> List[Dict[str, str]]:
"""Describe mirrored sidecar directories no configured root claims.
Empty unless centralized storage is active. Used by diagnostics to surface
metadata stranded by a moved model root (see :func:`_warn_unmatched`).
"""
sidecar_root = get_sidecar_root()
if not sidecar_root:
return []
state = _ensure_root_map(sidecar_root, _normalized_roots(_iter_model_roots()))
by_component: Dict[str, Dict[str, object]] = {}
for entry in state.entries.values():
by_component.setdefault(str(entry.get("component") or ""), entry)
unmatched: List[Dict[str, str]] = []
for component in state.unmatched:
entry = by_component.get(component) or {}
unmatched.append(
{
"component": component,
"basename": str(entry.get("basename") or ""),
"last_path": str(entry.get("last_path") or ""),
}
)
return unmatched
def get_sidecar_dir(model_path: str) -> str: def get_sidecar_dir(model_path: str) -> str:
"""Return the directory holding the model's sidecar/preview assets. """Return the directory holding the model's sidecar/preview assets.
+103
View File
@@ -12,6 +12,7 @@ import pytest
from aiohttp import web from aiohttp import web
from py.services.model_hash_index import ModelHashIndex from py.services.model_hash_index import ModelHashIndex
from py.routes.handlers import misc_handlers
from py.routes.handlers.misc_handlers import ( from py.routes.handlers.misc_handlers import (
BackupHandler, BackupHandler,
DoctorHandler, DoctorHandler,
@@ -209,6 +210,108 @@ async def test_doctor_handler_reports_key_cache_and_ui_issues():
assert diagnostic_map["ui_version"]["status"] == "warning" assert diagnostic_map["ui_version"]["status"] == "warning"
@pytest.mark.asyncio
async def test_doctor_handler_flags_orphaned_centralized_sidecars(
monkeypatch: pytest.MonkeyPatch,
):
async def civitai_factory():
return DummyCivitaiClient()
handler = DoctorHandler(
settings_service=DummySettings({"civitai_api_key": "token"}),
civitai_client_factory=civitai_factory,
scanner_factories=(),
)
monkeypatch.setattr(misc_handlers, "get_storage_mode", lambda: "centralized")
monkeypatch.setattr(
misc_handlers,
"get_unmatched_sidecar_components",
lambda: [
{
"component": "loras-abc12345",
"basename": "loras",
"last_path": "/mnt/old/loras",
},
{"component": "loras-def67890", "basename": "loras", "last_path": ""},
],
)
response = await handler.get_doctor_diagnostics(
FakeRequest(method="GET") # pyright: ignore[reportArgumentType]
)
payload = _json_payload(response)
item = {entry["id"]: entry for entry in payload["diagnostics"]}[
"sidecar_mirror_orphans"
]
assert item["status"] == "warning"
assert "2 sidecar directories" in item["summary"]
assert any("/mnt/old/loras" in line for line in item["details"])
assert any("unknown" in line for line in item["details"])
@pytest.mark.asyncio
async def test_doctor_handler_sidecar_check_ok_when_all_linked(
monkeypatch: pytest.MonkeyPatch,
):
async def civitai_factory():
return DummyCivitaiClient()
handler = DoctorHandler(
settings_service=DummySettings({"civitai_api_key": "token"}),
civitai_client_factory=civitai_factory,
scanner_factories=(),
)
monkeypatch.setattr(misc_handlers, "get_storage_mode", lambda: "centralized")
monkeypatch.setattr(misc_handlers, "get_unmatched_sidecar_components", lambda: [])
monkeypatch.setattr(
misc_handlers, "describe_sidecar_root", lambda: {"root": "/sidecars"}
)
response = await handler.get_doctor_diagnostics(
FakeRequest(method="GET") # pyright: ignore[reportArgumentType]
)
payload = _json_payload(response)
item = {entry["id"]: entry for entry in payload["diagnostics"]}[
"sidecar_mirror_orphans"
]
assert item["status"] == "ok"
@pytest.mark.asyncio
async def test_doctor_handler_sidecar_check_skipped_alongside(
monkeypatch: pytest.MonkeyPatch,
):
async def civitai_factory():
return DummyCivitaiClient()
handler = DoctorHandler(
settings_service=DummySettings({"civitai_api_key": "token"}),
civitai_client_factory=civitai_factory,
scanner_factories=(),
)
monkeypatch.setattr(misc_handlers, "get_storage_mode", lambda: "alongside")
def _unexpected():
raise AssertionError("orphan lookup must not run in alongside mode")
monkeypatch.setattr(
misc_handlers, "get_unmatched_sidecar_components", _unexpected
)
response = await handler.get_doctor_diagnostics(
FakeRequest(method="GET") # pyright: ignore[reportArgumentType]
)
payload = _json_payload(response)
item = {entry["id"]: entry for entry in payload["diagnostics"]}[
"sidecar_mirror_orphans"
]
assert item["status"] == "ok"
assert "alongside" in item["summary"]
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_doctor_handler_can_repair_cache(): async def test_doctor_handler_can_repair_cache():
scanner = DummyDoctorScanner(model_type="lora", raw_data=[]) scanner = DummyDoctorScanner(model_type="lora", raw_data=[])
@@ -19,13 +19,22 @@ from py.services.model_lifecycle_service import (
from py.services.pending_delete_service import PendingDeleteService from py.services.pending_delete_service import PendingDeleteService
from py.services.settings_manager import get_settings_manager from py.services.settings_manager import get_settings_manager
from py.utils.metadata_manager import MetadataManager from py.utils.metadata_manager import MetadataManager
from py.utils.sidecar_paths import root_mirror_component from py.utils.sidecar_paths import reset_root_map_cache, root_mirror_component
def _normalize(path) -> str: def _normalize(path) -> str:
return str(path).replace(os.sep, "/") return str(path).replace(os.sep, "/")
@pytest.fixture(autouse=True)
def _isolate_root_map():
"""Root identities are cached per sidecar root; keep tests independent."""
reset_root_map_cache()
yield
reset_root_map_cache()
@pytest.fixture @pytest.fixture
def library_root(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: def library_root(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
"""Single checkpoint root under tmp_path; every other root emptied.""" """Single checkpoint root under tmp_path; every other root emptied."""
@@ -59,11 +68,10 @@ def centralized(library_root: Path, tmp_path: Path) -> Path:
def _mirror_dir(library_root: Path, sidecar_root: Path, *rel: str) -> Path: def _mirror_dir(library_root: Path, sidecar_root: Path, *rel: str) -> Path:
"""Expected mirror directory for a library-relative path.""" """Expected mirror directory for a root-relative path."""
library = get_settings_manager().get_active_library_name() component = root_mirror_component(str(library_root), sidecar_root=str(sidecar_root))
component = root_mirror_component(str(library_root)) return sidecar_root.joinpath(component, *rel)
return sidecar_root.joinpath(library, component, *rel)
def _write_sidecar( def _write_sidecar(
@@ -4,6 +4,7 @@ from __future__ import annotations
import json import json
import os import os
import shutil
from pathlib import Path from pathlib import Path
from typing import Any, Dict, List from typing import Any, Dict, List
@@ -12,13 +13,26 @@ import pytest
from py.config import config from py.config import config
from py.services.settings_manager import get_settings_manager from py.services.settings_manager import get_settings_manager
from py.services.use_cases.sidecar_migration_use_case import SidecarMigrationUseCase from py.services.use_cases.sidecar_migration_use_case import SidecarMigrationUseCase
from py.utils.sidecar_paths import root_mirror_component from py.utils.sidecar_paths import (
get_metadata_path,
reset_root_map_cache,
root_mirror_component,
)
def _normalize(path) -> str: def _normalize(path) -> str:
return str(path).replace(os.sep, "/") return str(path).replace(os.sep, "/")
@pytest.fixture(autouse=True)
def _isolate_root_map():
"""Root identities are cached per sidecar root; keep tests independent."""
reset_root_map_cache()
yield
reset_root_map_cache()
@pytest.fixture @pytest.fixture
def library_root(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: def library_root(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
"""Single lora root under tmp_path; every other root emptied.""" """Single lora root under tmp_path; every other root emptied."""
@@ -54,11 +68,10 @@ def _set_mode(mode: str) -> None:
def _mirror_dir(library_root: Path, sidecar_root: Path, *rel: str) -> Path: def _mirror_dir(library_root: Path, sidecar_root: Path, *rel: str) -> Path:
"""Expected mirror directory for a library-relative path.""" """Expected mirror directory for a root-relative path."""
library = get_settings_manager().get_active_library_name() component = root_mirror_component(str(library_root), sidecar_root=str(sidecar_root))
component = root_mirror_component(str(library_root)) return sidecar_root.joinpath(component, *rel)
return sidecar_root.joinpath(library, component, *rel)
def _write_model(directory: Path, stem: str) -> Path: def _write_model(directory: Path, stem: str) -> Path:
@@ -482,6 +495,61 @@ async def test_migrate_root_relocates_tree_and_reconciles(
assert not old_root.exists() assert not old_root.exists()
@pytest.mark.asyncio
async def test_migrate_root_preserves_a_reanchored_identity(
library_root: Path, sidecar_root: Path, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
):
"""Relocating the sidecar root must not displace a pinned mirror name.
A root that moved earlier is mirrored under a name derived from its *old*
path. If anything resolves against the destination sidecar root before the
relocation runs, it writes a map naming the mirror after the *current*
path; the relocation must keep the pinned entry, because the directories
actually being moved are named after it.
"""
_set_mode("centralized")
first_sidecars = tmp_path / "first_sidecars"
get_settings_manager().set("sidecar_storage_path", str(first_sidecars))
reset_root_map_cache()
model = _write_model(library_root / "sub", "model")
pinned = root_mirror_component(str(library_root))
original = Path(get_metadata_path(str(model)))
original.parent.mkdir(parents=True, exist_ok=True)
original.write_text(json.dumps({"favorite": True}), encoding="utf-8")
# The model root moves; the recorded identity is re-anchored to it.
moved_root = tmp_path / "relocated" / "loras"
shutil.copytree(library_root, moved_root)
monkeypatch.setattr(config, "loras_roots", [str(moved_root)], raising=False)
reset_root_map_cache()
moved_model = moved_root / "sub" / "model.safetensors"
assert Path(get_metadata_path(str(moved_model))) == original
assert root_mirror_component(str(moved_root)) == pinned
# Settings now point at the destination and something resolves first: the
# destination map is written naming the mirror after the *current* path.
get_settings_manager().set("sidecar_storage_path", str(sidecar_root))
reset_root_map_cache()
before_move = Path(get_metadata_path(str(moved_model)))
assert before_move.parent.parent.name != pinned
assert not before_move.exists()
use_case = _make_use_case([str(moved_model)])
summary = await use_case.migrate_root(str(first_sidecars), force=True)
assert summary["success"] is True
# The metadata travelled with the tree and is still found under the
# pinned identity that names it.
relocated = sidecar_root / original.relative_to(first_sidecars)
assert relocated.exists()
assert json.loads(relocated.read_text(encoding="utf-8")) == {"favorite": True}
reset_root_map_cache()
assert Path(get_metadata_path(str(moved_model))) == relocated
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_migrate_root_guards( async def test_migrate_root_guards(
library_root: Path, sidecar_root: Path, tmp_path: Path library_root: Path, sidecar_root: Path, tmp_path: Path
+359 -12
View File
@@ -2,7 +2,10 @@
from __future__ import annotations from __future__ import annotations
import hashlib
import json
import os import os
import shutil
from pathlib import Path from pathlib import Path
import pytest import pytest
@@ -11,14 +14,17 @@ from py.services.settings_manager import get_settings_manager
from py.utils import sidecar_paths from py.utils import sidecar_paths
from py.utils.sidecar_paths import ( from py.utils.sidecar_paths import (
METADATA_SUFFIX, METADATA_SUFFIX,
ROOT_MAP_FILENAME,
get_configured_sidecar_root, get_configured_sidecar_root,
get_metadata_path, get_metadata_path,
get_preview_dir, get_preview_dir,
get_sidecar_dir, get_sidecar_dir,
get_sidecar_root, get_sidecar_root,
get_storage_mode, get_storage_mode,
get_unmatched_sidecar_components,
is_centralized, is_centralized,
is_metadata_path, is_metadata_path,
relocate_root_map,
resolve_centralized_dir, resolve_centralized_dir,
resolve_centralized_dir_for_dir, resolve_centralized_dir_for_dir,
resolve_metadata_path, resolve_metadata_path,
@@ -27,10 +33,36 @@ from py.utils.sidecar_paths import (
) )
def _write_map(root: Path, entries: dict) -> Path:
root.mkdir(parents=True, exist_ok=True)
path = root / ROOT_MAP_FILENAME
path.write_text(
json.dumps({"version": 1, "roots": entries}), encoding="utf-8"
)
return path
def _normalize(path: Path) -> str: def _normalize(path: Path) -> str:
return str(path).replace(os.sep, "/") return str(path).replace(os.sep, "/")
def _legacy_component(root) -> str:
"""Independent reimplementation of the pre-identity-map component name."""
normalized = os.path.normpath(os.path.abspath(str(root)))
digest = hashlib.sha256(normalized.encode("utf-8")).hexdigest()[:8]
return f"{sanitize_path_component(os.path.basename(normalized))}-{digest}"
@pytest.fixture(autouse=True)
def _reset_root_map_cache():
"""Identities are cached per sidecar root; keep tests order-independent."""
sidecar_paths.reset_root_map_cache()
yield
sidecar_paths.reset_root_map_cache()
@pytest.fixture @pytest.fixture
def model_roots(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> dict: def model_roots(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> dict:
"""Point every config model root the sidecar module reads at tmp_path.""" """Point every config model root the sidecar module reads at tmp_path."""
@@ -68,6 +100,17 @@ def centralized(model_roots: dict, tmp_path: Path) -> Path:
return sidecar_root return sidecar_root
def _write_sidecar(model_path: Path, payload: dict | None = None) -> Path:
"""Create the model file and its sidecar at the module-resolved path."""
model_path.parent.mkdir(parents=True, exist_ok=True)
model_path.write_bytes(b"weights")
sidecar = Path(get_metadata_path(str(model_path)))
sidecar.parent.mkdir(parents=True, exist_ok=True)
sidecar.write_text(json.dumps(payload or {"favorite": True}), encoding="utf-8")
return sidecar
class TestAlongsideMode: class TestAlongsideMode:
def test_default_mode_is_alongside(self): def test_default_mode_is_alongside(self):
assert get_storage_mode() == "alongside" assert get_storage_mode() == "alongside"
@@ -86,6 +129,26 @@ class TestAlongsideMode:
assert get_sidecar_dir(str(model)) == expected assert get_sidecar_dir(str(model)) == expected
assert get_preview_dir(str(model)) == expected assert get_preview_dir(str(model)) == expected
def test_alongside_resolution_touches_nothing(
self, model_roots: dict, tmp_path: Path
):
"""The default branch must not load, create, or scan the mirror tree."""
sidecar_root = tmp_path / "sidecars"
settings = get_settings_manager()
settings.set("sidecar_storage_mode", "alongside")
settings.set("sidecar_storage_path", str(sidecar_root))
model = model_roots["loras"] / "sub" / "model.safetensors"
assert get_metadata_path(str(model)) == os.path.join(
str(model_roots["loras"]), "sub", "model" + METADATA_SUFFIX
)
assert get_preview_dir(str(model)) == os.path.join(
str(model_roots["loras"]), "sub"
)
assert not sidecar_root.exists()
assert sidecar_paths._ROOT_MAPS == {}
class TestPathPredicates: class TestPathPredicates:
def test_is_metadata_path(self): def test_is_metadata_path(self):
@@ -117,13 +180,12 @@ class TestSanitizePathComponent:
class TestCentralizedMode: class TestCentralizedMode:
def test_mirror_layout(self, model_roots: dict, centralized: Path): def test_mirror_layout(self, model_roots: dict, centralized: Path):
model = model_roots["loras"] / "styles" / "anime" / "model.safetensors" model = model_roots["loras"] / "styles" / "anime" / "model.safetensors"
library = get_settings_manager().get_active_library_name()
root_component = root_mirror_component(str(model_roots["loras"])) root_component = root_mirror_component(str(model_roots["loras"]))
metadata_path = get_metadata_path(str(model)) metadata_path = get_metadata_path(str(model))
expected = os.path.join( expected = os.path.join(
str(centralized), library, root_component, "styles", "anime", "model" + METADATA_SUFFIX str(centralized), root_component, "styles", "anime", "model" + METADATA_SUFFIX
) )
assert metadata_path == expected assert metadata_path == expected
assert get_preview_dir(str(model)) == os.path.dirname(expected) assert get_preview_dir(str(model)) == os.path.dirname(expected)
@@ -155,7 +217,7 @@ class TestCentralizedMode:
assert root_mirror_component(str(model_roots["loras"])) != root_mirror_component( assert root_mirror_component(str(model_roots["loras"])) != root_mirror_component(
str(other_root) str(other_root)
) )
# Same root always maps to the same component (stable hash). # Same root always maps to the same component (pinned identity).
assert root_mirror_component(str(model_roots["loras"])) == root_mirror_component( assert root_mirror_component(str(model_roots["loras"])) == root_mirror_component(
str(model_roots["loras"]) + os.sep str(model_roots["loras"]) + os.sep
) )
@@ -171,11 +233,10 @@ class TestCentralizedMode:
[str(model_roots["loras"]), str(nested)], [str(model_roots["loras"]), str(nested)],
raising=False, raising=False,
) )
library = get_settings_manager().get_active_library_name()
model = nested / "model.safetensors" model = nested / "model.safetensors"
assert get_metadata_path(str(model)) == os.path.join( assert get_metadata_path(str(model)) == os.path.join(
str(centralized), library, root_mirror_component(str(nested)), "model" + METADATA_SUFFIX str(centralized), root_mirror_component(str(nested)), "model" + METADATA_SUFFIX
) )
def test_outside_roots_falls_back_to_alongside( def test_outside_roots_falls_back_to_alongside(
@@ -204,10 +265,8 @@ class TestCentralizedMode:
def test_resolve_centralized_dir_for_dir_root_maps_to_mirror_base( def test_resolve_centralized_dir_for_dir_root_maps_to_mirror_base(
self, model_roots: dict, centralized: Path self, model_roots: dict, centralized: Path
): ):
library = get_settings_manager().get_active_library_name()
assert resolve_centralized_dir_for_dir(str(model_roots["loras"])) == os.path.join( assert resolve_centralized_dir_for_dir(str(model_roots["loras"])) == os.path.join(
str(centralized), library, root_mirror_component(str(model_roots["loras"])) str(centralized), root_mirror_component(str(model_roots["loras"]))
) )
def test_empty_path_uses_default_sidecar_root(self, model_roots: dict, tmp_path: Path): def test_empty_path_uses_default_sidecar_root(self, model_roots: dict, tmp_path: Path):
@@ -221,6 +280,293 @@ class TestCentralizedMode:
assert is_centralized() assert is_centralized()
class TestRootIdentityMap:
"""The root component is a remembered identity, not a path hash."""
def test_component_is_pinned_in_the_map_file(
self, model_roots: dict, centralized: Path
):
model = model_roots["loras"] / "sub" / "model.safetensors"
first = get_metadata_path(str(model))
map_path = centralized / ROOT_MAP_FILENAME
assert map_path.exists()
payload = json.loads(map_path.read_text(encoding="utf-8"))
persisted = {entry["component"] for entry in payload["roots"].values()}
assert root_mirror_component(str(model_roots["loras"])) in persisted
# Restarting the process must resolve the exact same location.
sidecar_paths.reset_root_map_cache()
assert get_metadata_path(str(model)) == first
def test_component_is_deterministic_and_survives_a_lost_map(
self, model_roots: dict, centralized: Path
):
"""Deleting the map must not strand a mirror whose root has not moved."""
model = model_roots["loras"] / "sub" / "model.safetensors"
first = _write_sidecar(model)
assert Path(get_metadata_path(str(model))) == first
(centralized / ROOT_MAP_FILENAME).unlink()
sidecar_paths.reset_root_map_cache()
assert Path(get_metadata_path(str(model))) == first
assert root_mirror_component(str(model_roots["loras"])) == _legacy_component(
model_roots["loras"]
)
def test_legacy_path_derived_component_is_adopted(
self, model_roots: dict, centralized: Path
):
"""Upgrading from the hash-named layout keeps existing mirrors usable."""
legacy = _legacy_component(model_roots["loras"])
sidecar = centralized / legacy / "sub" / ("model" + METADATA_SUFFIX)
sidecar.parent.mkdir(parents=True)
sidecar.write_text(json.dumps({"favorite": True}), encoding="utf-8")
model = model_roots["loras"] / "sub" / "model.safetensors"
assert get_metadata_path(str(model)) == str(sidecar)
assert root_mirror_component(str(model_roots["loras"])) == legacy
def test_pre_identity_map_library_nested_component_is_adopted(
self, model_roots: dict, centralized: Path
):
"""Mirrors created by the pre-identity-map build nested under a library."""
legacy = _legacy_component(model_roots["loras"])
sidecar = (
centralized / "comfyui" / legacy / "sub" / ("model" + METADATA_SUFFIX)
)
sidecar.parent.mkdir(parents=True)
sidecar.write_text(json.dumps({"favorite": True}), encoding="utf-8")
model = model_roots["loras"] / "sub" / "model.safetensors"
assert get_metadata_path(str(model)) == str(sidecar)
assert root_mirror_component(str(model_roots["loras"])) == f"comfyui/{legacy}"
# The adopted tree counts as linked, not orphaned.
assert get_unmatched_sidecar_components() == []
def test_moved_root_reuses_its_mirror(
self, model_roots: dict, centralized: Path, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
):
"""A relocated (and reconfigured) root keeps its remembered identity."""
from py.config import config
model = model_roots["loras"] / "sub" / "model.safetensors"
before = _write_sidecar(model)
moved_root = tmp_path / "relocated" / "loras"
shutil.copytree(model_roots["loras"], moved_root)
monkeypatch.setattr(config, "loras_roots", [str(moved_root)], raising=False)
sidecar_paths.reset_root_map_cache()
moved_model = moved_root / "sub" / "model.safetensors"
after = Path(get_metadata_path(str(moved_model)))
assert after == before
assert json.loads(after.read_text(encoding="utf-8")) == {"favorite": True}
def test_renamed_root_reanchors_via_directory_overlap(
self, model_roots: dict, centralized: Path, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
):
"""A rename that changes the basename still re-anchors when the tree matches."""
from py.config import config
model = model_roots["loras"] / "sub" / "model.safetensors"
before = _write_sidecar(model)
renamed_root = tmp_path / "brand-new-name"
shutil.copytree(model_roots["loras"], renamed_root)
monkeypatch.setattr(config, "loras_roots", [str(renamed_root)], raising=False)
sidecar_paths.reset_root_map_cache()
renamed_model = renamed_root / "sub" / "model.safetensors"
assert Path(get_metadata_path(str(renamed_model))) == before
def test_ambiguous_reanchor_reports_orphans_instead_of_guessing(
self, model_roots: dict, centralized: Path, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
):
from py.config import config
model = model_roots["loras"] / "sub" / "model.safetensors"
before = _write_sidecar(model)
component = before.parent.parent.name
# Two equally plausible candidates: never silently pick one.
candidate_a = tmp_path / "a" / "loras"
candidate_b = tmp_path / "b" / "loras"
shutil.copytree(model_roots["loras"], candidate_a)
shutil.copytree(model_roots["loras"], candidate_b)
monkeypatch.setattr(
config, "loras_roots", [str(candidate_a), str(candidate_b)], raising=False
)
sidecar_paths.reset_root_map_cache()
for candidate in (candidate_a, candidate_b):
resolved = Path(
get_metadata_path(str(candidate / "sub" / "model.safetensors"))
)
assert resolved != before
assert component not in str(resolved)
orphans = get_unmatched_sidecar_components()
assert [item["component"] for item in orphans] == [component]
def test_removed_root_surfaces_its_component_as_orphan(
self, model_roots: dict, centralized: Path, monkeypatch: pytest.MonkeyPatch
):
from py.config import config
model = model_roots["loras"] / "sub" / "model.safetensors"
sidecar = _write_sidecar(model)
component = sidecar.parent.parent.name
monkeypatch.setattr(config, "loras_roots", [], raising=False)
sidecar_paths.reset_root_map_cache()
assert [item["component"] for item in get_unmatched_sidecar_components()] == [
component
]
def test_unmatched_helper_is_empty_in_alongside_mode(
self, model_roots: dict, tmp_path: Path
):
settings = get_settings_manager()
settings.set("sidecar_storage_mode", "alongside")
settings.set("sidecar_storage_path", str(tmp_path / "sidecars"))
assert get_unmatched_sidecar_components() == []
assert not (tmp_path / "sidecars").exists()
def test_relocated_sidecar_root_resolves_after_map_written_first(
self, model_roots: dict, tmp_path: Path
):
"""A resolve before the mirror tree is relocated must not rename it."""
settings = get_settings_manager()
old_root = tmp_path / "old-sidecars"
new_root = tmp_path / "new-sidecars"
settings.set("sidecar_storage_mode", "centralized")
settings.set("sidecar_storage_path", str(old_root))
model = model_roots["loras"] / "sub" / "model.safetensors"
sidecar = _write_sidecar(model)
# The path setting changes first, and something resolves against the
# new root before the user runs the relocation.
settings.set("sidecar_storage_path", str(new_root))
sidecar_paths.reset_root_map_cache()
expected = new_root / sidecar.relative_to(old_root)
assert Path(get_metadata_path(str(model))) == expected
assert not expected.exists()
# Now the mirror tree moves, exactly as migrate_root does.
for dirpath, _dirnames, filenames in os.walk(old_root):
rel = os.path.relpath(dirpath, old_root)
target_dir = new_root if rel == os.curdir else new_root / rel
target_dir.mkdir(parents=True, exist_ok=True)
for filename in filenames:
shutil.move(str(Path(dirpath) / filename), str(target_dir / filename))
sidecar_paths.reset_root_map_cache()
assert Path(get_metadata_path(str(model))) == expected
assert json.loads(expected.read_text(encoding="utf-8")) == {"favorite": True}
def test_relocate_root_map_prefers_the_source_entries(self, tmp_path: Path):
"""A map written at the destination before a relocation must not win."""
source = tmp_path / "old-sidecars"
destination = tmp_path / "new-sidecars"
_write_map(
source,
{
"aaaa1111": {
"component": "loras-pinned",
"basename": "loras",
"last_path": "/models/loras",
"sample_rel_dirs": [],
}
},
)
_write_map(
destination,
{
"bbbb2222": {
"component": "loras-recomputed",
"basename": "loras",
"last_path": "/models/loras",
"sample_rel_dirs": [],
},
"cccc3333": {
"component": "vae-other",
"basename": "vae",
"last_path": "/models/vae",
"sample_rel_dirs": [],
},
},
)
assert relocate_root_map(str(source), str(destination)) is True
merged = json.loads(
(destination / ROOT_MAP_FILENAME).read_text(encoding="utf-8")
)["roots"]
assert merged["aaaa1111"]["component"] == "loras-pinned"
assert "bbbb2222" not in merged # superseded for the same root path
assert merged["cccc3333"]["component"] == "vae-other"
assert not (source / ROOT_MAP_FILENAME).exists()
def test_relocate_root_map_without_a_source_map_is_a_noop(self, tmp_path: Path):
source = tmp_path / "old-sidecars"
destination = tmp_path / "new-sidecars"
_write_map(
destination,
{
"bbbb2222": {
"component": "loras-keep",
"basename": "loras",
"last_path": "/models/loras",
"sample_rel_dirs": [],
}
},
)
assert relocate_root_map(str(source), str(destination)) is True
kept = json.loads(
(destination / ROOT_MAP_FILENAME).read_text(encoding="utf-8")
)["roots"]
assert kept["bbbb2222"]["component"] == "loras-keep"
def test_corrupt_map_file_is_ignored_and_rebuilt(
self, model_roots: dict, centralized: Path
):
model = model_roots["loras"] / "model.safetensors"
map_path = centralized / ROOT_MAP_FILENAME
get_metadata_path(str(model))
map_path.write_text("{not json", encoding="utf-8")
sidecar_paths.reset_root_map_cache()
path_after = get_metadata_path(str(model))
assert json.loads(map_path.read_text(encoding="utf-8"))["roots"]
assert os.path.basename(os.path.dirname(path_after)).startswith("loras-")
def test_root_map_file_is_not_mistaken_for_a_component(
self, model_roots: dict, centralized: Path
):
model = model_roots["loras"] / "model.safetensors"
get_metadata_path(str(model))
assert (centralized / ROOT_MAP_FILENAME).is_file()
assert get_unmatched_sidecar_components() == []
class TestModeIndependentResolution: class TestModeIndependentResolution:
"""Migration tooling resolves the mirror layout regardless of active mode.""" """Migration tooling resolves the mirror layout regardless of active mode."""
@@ -248,17 +594,18 @@ class TestModeIndependentResolution:
sidecar_root = tmp_path / "sidecars" sidecar_root = tmp_path / "sidecars"
settings = get_settings_manager() settings = get_settings_manager()
settings.set("sidecar_storage_mode", "alongside") settings.set("sidecar_storage_mode", "alongside")
library = settings.get_active_library_name() settings.set("sidecar_storage_path", str(sidecar_root))
model_dir = model_roots["loras"] / "sub" model_dir = model_roots["loras"] / "sub"
component = root_mirror_component(
str(model_roots["loras"]), sidecar_root=str(sidecar_root)
)
# Alongside mode: no root resolves without the override. # Alongside mode: no root resolves without the override.
assert resolve_centralized_dir_for_dir(str(model_dir)) is None assert resolve_centralized_dir_for_dir(str(model_dir)) is None
assert resolve_centralized_dir_for_dir( assert resolve_centralized_dir_for_dir(
str(model_dir), sidecar_root=str(sidecar_root) str(model_dir), sidecar_root=str(sidecar_root)
) == os.path.join( ) == os.path.join(str(sidecar_root), component, "sub")
str(sidecar_root), library, root_mirror_component(str(model_roots["loras"])), "sub"
)
class TestSettingsValidation: class TestSettingsValidation: