Compare commits

..
Author SHA1 Message Date
Will Miao 0dd8d74032 fix(ui): stop body data-theme from shadowing theme preset tokens
applyTheme() mirrors the active mode onto <body> as data-theme="dark", but
the theme preset is only ever written to <html>. The palette token blocks in
tokens/colors.css and base.css used the bare attribute selector
[data-theme="dark"], so <body> matched them on its own and re-declared the
default dark palette (#1a1a1a / #2d2d2d / ...) directly on the body, where it
shadowed the preset values inherited from <html>. Every non-default preset
therefore painted the selected accent over the default preset's background,
surface, text and border tokens, and flipped into that state ~200ms after
load, when initTheme() first touched <body> — the accent-tinted background
flash seen on reload and nav-tab switches. Reached only in dark mode, since
light mode has no [data-theme="light"] token block.

Scope the palette token blocks to :root so a data-theme attribute on any
descendant (only <body> has one) can no longer re-declare them; descendant
rules such as [data-theme="dark"] .foo still match through <html>. Add a
regression guard that fails on bare attribute token blocks.
2026-09-27 21:48:58 +08:00
22 changed files with 146 additions and 1463 deletions
+5 -8
View File
@@ -20,21 +20,18 @@ By default, `.metadata.json` sidecars and preview images live **alongside** thei
| `sidecar_storage_mode` | `"alongside"` \| `"centralized"` | `"alongside"` |
| `sidecar_storage_path` | Absolute path string; empty = `<settings dir>/sidecars` | `""` |
In centralized mode, sidecars and previews mirror each model root's directory structure:
In centralized mode, sidecars and previews mirror the library-relative directory structure:
```
<sidecar_root>/<root_component>/<rel_dir>/<name>.metadata.json
<sidecar_root>/<library>/<root_basename-roothash>/<rel_dir>/<name>.metadata.json
```
- `<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.
- `<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.
- `.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 `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.
- 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 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. 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.
- All sidecar/preview path derivation goes through the helpers in `py/utils/sidecar_paths.py`; never construct paths inline.
---
-3
View File
@@ -2769,9 +2769,6 @@
"filename_conflicts": {
"title": "Konflikte durch doppelte Dateinamen"
},
"sidecar_mirror_orphans": {
"title": "[TODO: Translate] Centralized Sidecars"
},
"ui_version": {
"title": "UI-Version"
}
-3
View File
@@ -2769,9 +2769,6 @@
"filename_conflicts": {
"title": "Duplicate Filename Conflicts"
},
"sidecar_mirror_orphans": {
"title": "Centralized Sidecars"
},
"ui_version": {
"title": "UI Version"
}
-3
View File
@@ -2769,9 +2769,6 @@
"filename_conflicts": {
"title": "Conflictos de nombres de archivo duplicados"
},
"sidecar_mirror_orphans": {
"title": "[TODO: Translate] Centralized Sidecars"
},
"ui_version": {
"title": "Versión de la interfaz"
}
-3
View File
@@ -2769,9 +2769,6 @@
"filename_conflicts": {
"title": "Conflits de noms de fichiers en double"
},
"sidecar_mirror_orphans": {
"title": "[TODO: Translate] Centralized Sidecars"
},
"ui_version": {
"title": "Version de l'interface"
}
-3
View File
@@ -2769,9 +2769,6 @@
"filename_conflicts": {
"title": "התנגשויות שמות קבצים כפולים"
},
"sidecar_mirror_orphans": {
"title": "[TODO: Translate] Centralized Sidecars"
},
"ui_version": {
"title": "גרסת הממשק"
}
-3
View File
@@ -2769,9 +2769,6 @@
"filename_conflicts": {
"title": "ファイル名重複競合"
},
"sidecar_mirror_orphans": {
"title": "[TODO: Translate] Centralized Sidecars"
},
"ui_version": {
"title": "UI バージョン"
}
-3
View File
@@ -2769,9 +2769,6 @@
"filename_conflicts": {
"title": "파일명 중복 충돌"
},
"sidecar_mirror_orphans": {
"title": "[TODO: Translate] Centralized Sidecars"
},
"ui_version": {
"title": "UI 버전"
}
-3
View File
@@ -2769,9 +2769,6 @@
"filename_conflicts": {
"title": "Конфликты дублирующихся имён файлов"
},
"sidecar_mirror_orphans": {
"title": "[TODO: Translate] Centralized Sidecars"
},
"ui_version": {
"title": "Версия интерфейса"
}
-3
View File
@@ -2769,9 +2769,6 @@
"filename_conflicts": {
"title": "文件名重复冲突"
},
"sidecar_mirror_orphans": {
"title": "[TODO: Translate] Centralized Sidecars"
},
"ui_version": {
"title": "UI 版本"
}
-3
View File
@@ -2769,9 +2769,6 @@
"filename_conflicts": {
"title": "檔案名稱重複衝突"
},
"sidecar_mirror_orphans": {
"title": "[TODO: Translate] Centralized Sidecars"
},
"ui_version": {
"title": "UI 版本"
}
-68
View File
@@ -75,8 +75,6 @@ from ...utils.sidecar_paths import (
get_configured_sidecar_root,
get_metadata_path,
get_preview_dir,
get_storage_mode,
get_unmatched_sidecar_components,
)
from ...utils.usage_stats import UsageStats
from .base_model_handlers import BaseModelHandlerSet
@@ -808,7 +806,6 @@ class DoctorHandler:
await self._check_civitai_api_key(),
await self._check_cache_health(),
await self._check_filename_conflicts(),
self._check_sidecar_mirror_orphans(),
self._check_ui_version(client_version, app_version),
]
@@ -1045,71 +1042,6 @@ class DoctorHandler:
logger.error("Error exporting doctor bundle: %s", exc, exc_info=True)
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]:
api_key = (self._settings.get("civitai_api_key", "") or "").strip()
if not api_key:
@@ -5,8 +5,7 @@ Two storage layouts exist (see :mod:`py.utils.sidecar_paths`):
- ``alongside``: ``<model_dir>/<name>.metadata.json`` and preview files live
next to the model file.
- ``centralized``: the same files live under the configured sidecar root,
mirroring each model root's directory structure under a per-root identity
component (see :func:`py.utils.sidecar_paths.root_mirror_component`).
mirroring the library-relative directory structure.
This use case moves the ``.metadata.json`` sidecar and preview files for every
known model from one layout to the other. Model files themselves NEVER move.
@@ -52,12 +51,10 @@ from ...utils.file_utils import find_preview_file, get_preview_extension
from ...utils.metadata_manager import MetadataManager
from ...utils.sidecar_paths import (
METADATA_SUFFIX,
ROOT_MAP_FILENAME,
STORAGE_MODE_CENTRALIZED,
get_configured_sidecar_root,
get_sidecar_root,
get_storage_mode,
relocate_root_map,
resolve_centralized_dir_for_dir,
)
@@ -202,19 +199,14 @@ class SidecarMigrationUseCase:
)
files: List[Tuple[str, str]] = []
source_map_path = os.path.join(old, ROOT_MAP_FILENAME)
if os.path.isdir(old):
for dirpath, _dirnames, filenames in os.walk(old):
rel = os.path.relpath(dirpath, old)
target_dir = new_root if rel == os.curdir else os.path.join(new_root, rel)
for filename in filenames:
source = os.path.join(dirpath, 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)))
files.append(
(os.path.join(dirpath, filename), os.path.join(target_dir, filename))
)
errors: List[Dict[str, str]] = []
counters: Dict[str, Any] = {"moved": 0, "conflicts": 0}
@@ -250,23 +242,6 @@ class SidecarMigrationUseCase:
errors.append({"model": os.path.basename(src), "error": str(exc)})
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("/") + "/"
new_prefix = new_root.replace(os.sep, "/").rstrip("/") + "/"
for sidecar in moved_sidecars:
+34 -767
View File
@@ -11,39 +11,21 @@ setting:
file, the historical layout other tools may rely on.
- ``centralized``: sidecars and previews live under a configurable root
(``sidecar_storage_path`` setting, default ``<settings_dir>/sidecars``),
mirroring each model root's directory structure::
mirroring the library-relative directory structure:
``<root>/<library>/<root_basename-roothash>/<rel_dir>/<name>.metadata.json``.
<sidecar_root>/<root_component>/<rel_dir>/<name>.metadata.json
``<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.
All helpers are pure path computations: no directory scans and no file I/O
on the hot path. Settings lookups go through ``SettingsManager.get`` (a dict
read); config roots come from the already-initialized ``config`` singleton.
"""
from __future__ import annotations
import hashlib
import json
import logging
import os
import re
import threading
import time
import uuid
from typing import Dict, Iterable, List, Optional, Set, Tuple
from typing import List, Optional
logger = logging.getLogger(__name__)
@@ -54,21 +36,6 @@ STORAGE_MODE_CENTRALIZED = "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):
"""Read a setting defensively; never fail path resolution on settings errors."""
@@ -217,262 +184,16 @@ def _iter_model_roots() -> List[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))
def _normalized_roots(raw_roots: Iterable[str]) -> Tuple[str, ...]:
"""Return the deduplicated, normalized, order-stable model roots."""
def root_mirror_component(root_path: str) -> str:
"""Return the mirror path component identifying a model root.
seen: Dict[str, None] = {}
for candidate in raw_roots:
if not isinstance(candidate, str) or not candidate.strip():
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.
``<sanitized basename>-<hash>`` where the hash is a short digest of the
normalized absolute root path. Two roots sharing a basename (e.g.
``/mnt/a/loras`` and ``/mnt/b/loras``) would otherwise map to the same
mirror directory and overwrite each other's sidecars.
"""
normalized = _normalize_for_match(root_path)
@@ -480,444 +201,12 @@ def _legacy_component(root_path: str) -> str:
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]:
"""Return the centralized mirror directory for ``model_path``.
The mirror layout is ``<sidecar_root>/<root_component>/<rel_dir>`` where
``rel_dir`` is the model's directory relative to the model root that
The mirror layout is
``<sidecar_root>/<library>/<root_basename-roothash>/<rel_dir>``
where ``rel_dir`` is the model's directory relative to the model root that
contains it. The longest matching root wins so nested roots resolve to the
most specific mirror. Returns ``None`` when centralized storage is inactive
or the path is not under any configured model root.
@@ -948,55 +237,33 @@ def resolve_centralized_dir_for_dir(
return None
normalized_dir = _normalize_for_match(model_dir)
normalized_roots = _normalized_roots(_iter_model_roots())
best_root = _best_root_for(normalized_dir, normalized_roots)
best_root: Optional[str] = None
for candidate in _iter_model_roots():
if not candidate:
continue
normalized = _normalize_for_match(candidate)
if normalized_dir == normalized or normalized_dir.startswith(normalized + os.sep):
if best_root is None or len(normalized) > len(best_root):
best_root = normalized
if best_root is None:
return None
try:
from ..services.settings_manager import get_settings_manager
library = get_settings_manager().get_active_library_name() or "default"
except Exception: # pragma: no cover - defensive fallback
library = "default"
rel_dir = os.path.relpath(normalized_dir, best_root)
component = _component_for(best_root, root, normalized_roots, rel_dir)
if not component:
return None
parts = [root, component]
parts = [root, sanitize_path_component(library), root_mirror_component(best_root)]
if rel_dir and rel_dir != os.curdir:
parts.extend(
sanitize_path_component(part)
for part in rel_dir.split(os.sep)
if part not in ("", os.curdir)
)
parts.extend(sanitize_path_component(part) for part in rel_dir.split(os.sep) if part not in ("", os.curdir))
return os.path.join(*parts)
def get_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:
"""Return the directory holding the model's sidecar/preview assets.
+1 -1
View File
@@ -81,7 +81,7 @@ body {
--badge-skip-refresh-glow: var(--color-skip-refresh-glow);
}
[data-theme="dark"] {
:root[data-theme="dark"] {
--bg-color: var(--bg-base);
--text-color: var(--text-primary);
--text-muted: var(--text-secondary);
+1 -1
View File
@@ -572,7 +572,7 @@
}
/* Dark mode: use each preset's dark-mode accent lightness for visibility.
These match the --color-accent-l values from [data-theme="dark"][data-theme-preset="..."]
These match the --color-accent-l values from :root[data-theme="dark"][data-theme-preset="..."]
in tokens/colors.css so the swatch accurately previews what the theme looks like. */
[data-theme="dark"] .preset-swatch-default {
+11 -11
View File
@@ -78,7 +78,7 @@
--favorite-glow: oklch(65% 0.15 85 / 0.5);
}
[data-theme="dark"] {
:root[data-theme="dark"] {
--bg-base: #1a1a1a;
--bg-elevated: oklch(25% 0.02 256 / 0.98);
--bg-overlay: oklch(0% 0 0 / 0.75);
@@ -118,7 +118,7 @@
/* ── Preset: Nord ──────────────────────────────────────────── */
[data-theme-preset="nord"] {
:root[data-theme-preset="nord"] {
--color-accent-h: 213;
--color-accent-c: 0.18;
--color-accent-l: 62%;
@@ -152,7 +152,7 @@
--favorite-glow: oklch(72% 0.14 85 / 0.5);
}
[data-theme="dark"][data-theme-preset="nord"] {
:root[data-theme="dark"][data-theme-preset="nord"] {
--color-accent-h: 213;
--color-accent-c: 0.18;
--color-accent-l: 68%;
@@ -188,7 +188,7 @@
/* ── Preset: Midnight ───────────────────────────────────────── */
[data-theme-preset="midnight"] {
:root[data-theme-preset="midnight"] {
--color-accent-h: 300;
--color-accent-c: 0.15;
--color-accent-l: 52%;
@@ -222,7 +222,7 @@
--favorite-glow: oklch(72% 0.16 85 / 0.5);
}
[data-theme="dark"][data-theme-preset="midnight"] {
:root[data-theme="dark"][data-theme-preset="midnight"] {
--color-accent-h: 300;
--color-accent-c: 0.14;
--color-accent-l: 68%;
@@ -258,7 +258,7 @@
/* ── Preset: Monokai ───────────────────────────────────────── */
[data-theme-preset="monokai"] {
:root[data-theme-preset="monokai"] {
--color-accent-h: 190;
--color-accent-c: 0.24;
--color-accent-l: 72%;
@@ -291,7 +291,7 @@
--favorite-glow: oklch(72% 0.16 85 / 0.5);
}
[data-theme="dark"][data-theme-preset="monokai"] {
:root[data-theme="dark"][data-theme-preset="monokai"] {
--color-accent-h: 190;
--color-accent-c: 0.24;
--color-accent-l: 72%;
@@ -326,7 +326,7 @@
/* ── Preset: Dracula ───────────────────────────────────────── */
[data-theme-preset="dracula"] {
:root[data-theme-preset="dracula"] {
--color-accent-h: 265;
--color-accent-c: 0.24;
--color-accent-l: 68%;
@@ -359,7 +359,7 @@
--favorite-glow: oklch(72% 0.16 85 / 0.5);
}
[data-theme="dark"][data-theme-preset="dracula"] {
:root[data-theme="dark"][data-theme-preset="dracula"] {
--color-accent-h: 265;
--color-accent-c: 0.24;
--color-accent-l: 72%;
@@ -394,7 +394,7 @@
/* ── Preset: Solarized ─────────────────────────────────────── */
[data-theme-preset="solarized"] {
:root[data-theme-preset="solarized"] {
--color-accent-h: 175;
--color-accent-c: 0.18;
--color-accent-l: 55%;
@@ -429,7 +429,7 @@
--favorite-glow: oklch(68% 0.16 75 / 0.5);
}
[data-theme="dark"][data-theme-preset="solarized"] {
:root[data-theme="dark"][data-theme-preset="solarized"] {
--color-accent-h: 175;
--color-accent-c: 0.18;
--color-accent-l: 60%;
@@ -0,0 +1,68 @@
import { describe, it, expect } from 'vitest';
import { readFileSync } from 'fs';
import path from 'path';
// Regression guard: theme palette tokens may only be declared on the root element.
//
// `applyTheme()` (static/js/utils/uiHelpers.js, and Header.setThemeMode) mirrors
// the active mode onto <body> as `data-theme="dark"`, but the theme *preset* is
// only ever written to <html> (`data-theme-preset`). While the token blocks in
// tokens/colors.css used the bare attribute selector `[data-theme="dark"]`, the
// body matched that block on its own and re-declared the DEFAULT dark palette
// (#1a1a1a / #2d2d2d), shadowing the preset palette it inherited from <html>.
// The page painted the selected preset's accent but the default preset's
// backgrounds/surfaces/text, and flipped into that state ~200ms after load when
// initTheme() first touched <body> — the accent-tinted background flash on
// reload and nav-tab switches, visible under every preset except "default".
// Root-scoping the token blocks keeps <body>'s data-theme inert.
describe('Theme token scope', () => {
const repoRoot = path.resolve(__dirname, '../../..');
const read = (rel) => readFileSync(path.join(repoRoot, rel), 'utf-8');
const stripComments = (css) => css.replace(/\/\*[\s\S]*?\*\//g, '');
const COLOR_TOKENS = read('static/css/tokens/colors.css');
const BASE_CSS = read('static/css/base.css');
// Selectors that declare a palette token, e.g. `--bg-base:` / `--lora-surface:`.
const tokenDeclaringSelectors = (css) => {
const selectors = [];
const ruleRe = /([^{}]+)\{([^{}]*)\}/g;
let match;
while ((match = ruleRe.exec(stripComments(css)))) {
const selector = match[1].trim();
const declaresPaletteToken = /(^|[;\s])--(?:bg|surface|text|border|color|favorite|lora|badge|card)-[\w-]+\s*:/.test(
match[2]
);
if (declaresPaletteToken) selectors.push(selector);
}
return selectors;
};
const isRootScoped = (selector) =>
selector.split(',').every((part) => /^\s*(:root|html)\b/.test(part));
it('declares every colors.css palette token on :root', () => {
const selectors = tokenDeclaringSelectors(COLOR_TOKENS);
expect(selectors.length).toBeGreaterThan(0);
expect(selectors.filter((selector) => !isRootScoped(selector))).toEqual([]);
});
it('declares every base.css palette alias on :root', () => {
const selectors = tokenDeclaringSelectors(BASE_CSS);
expect(selectors.length).toBeGreaterThan(0);
expect(selectors.filter((selector) => !isRootScoped(selector))).toEqual([]);
});
it('keeps dark/preset token blocks anchored to the root element', () => {
const css = stripComments(COLOR_TOKENS);
expect(css).toContain(':root[data-theme="dark"] {');
for (const preset of ['nord', 'midnight', 'monokai', 'dracula', 'solarized']) {
expect(css).toContain(`:root[data-theme-preset="${preset}"] {`);
expect(css).toContain(`:root[data-theme="dark"][data-theme-preset="${preset}"] {`);
}
// No bare attribute selector may open a rule: <body> carries data-theme
// without the preset, so such a block would re-declare the default palette.
expect(css).not.toMatch(/(^|\n)\s*\[data-theme/);
expect(stripComments(BASE_CSS)).not.toMatch(/(^|\n)\s*\[data-theme/);
});
});
-103
View File
@@ -12,7 +12,6 @@ import pytest
from aiohttp import web
from py.services.model_hash_index import ModelHashIndex
from py.routes.handlers import misc_handlers
from py.routes.handlers.misc_handlers import (
BackupHandler,
DoctorHandler,
@@ -210,108 +209,6 @@ async def test_doctor_handler_reports_key_cache_and_ui_issues():
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
async def test_doctor_handler_can_repair_cache():
scanner = DummyDoctorScanner(model_type="lora", raw_data=[])
@@ -19,22 +19,13 @@ from py.services.model_lifecycle_service import (
from py.services.pending_delete_service import PendingDeleteService
from py.services.settings_manager import get_settings_manager
from py.utils.metadata_manager import MetadataManager
from py.utils.sidecar_paths import reset_root_map_cache, root_mirror_component
from py.utils.sidecar_paths import root_mirror_component
def _normalize(path) -> str:
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
def library_root(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
"""Single checkpoint root under tmp_path; every other root emptied."""
@@ -68,10 +59,11 @@ def centralized(library_root: Path, tmp_path: Path) -> Path:
def _mirror_dir(library_root: Path, sidecar_root: Path, *rel: str) -> Path:
"""Expected mirror directory for a root-relative path."""
"""Expected mirror directory for a library-relative path."""
component = root_mirror_component(str(library_root), sidecar_root=str(sidecar_root))
return sidecar_root.joinpath(component, *rel)
library = get_settings_manager().get_active_library_name()
component = root_mirror_component(str(library_root))
return sidecar_root.joinpath(library, component, *rel)
def _write_sidecar(
@@ -4,7 +4,6 @@ from __future__ import annotations
import json
import os
import shutil
from pathlib import Path
from typing import Any, Dict, List
@@ -13,26 +12,13 @@ import pytest
from py.config import config
from py.services.settings_manager import get_settings_manager
from py.services.use_cases.sidecar_migration_use_case import SidecarMigrationUseCase
from py.utils.sidecar_paths import (
get_metadata_path,
reset_root_map_cache,
root_mirror_component,
)
from py.utils.sidecar_paths import root_mirror_component
def _normalize(path) -> str:
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
def library_root(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
"""Single lora root under tmp_path; every other root emptied."""
@@ -68,10 +54,11 @@ def _set_mode(mode: str) -> None:
def _mirror_dir(library_root: Path, sidecar_root: Path, *rel: str) -> Path:
"""Expected mirror directory for a root-relative path."""
"""Expected mirror directory for a library-relative path."""
component = root_mirror_component(str(library_root), sidecar_root=str(sidecar_root))
return sidecar_root.joinpath(component, *rel)
library = get_settings_manager().get_active_library_name()
component = root_mirror_component(str(library_root))
return sidecar_root.joinpath(library, component, *rel)
def _write_model(directory: Path, stem: str) -> Path:
@@ -495,61 +482,6 @@ async def test_migrate_root_relocates_tree_and_reconciles(
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
async def test_migrate_root_guards(
library_root: Path, sidecar_root: Path, tmp_path: Path
+12 -359
View File
@@ -2,10 +2,7 @@
from __future__ import annotations
import hashlib
import json
import os
import shutil
from pathlib import Path
import pytest
@@ -14,17 +11,14 @@ from py.services.settings_manager import get_settings_manager
from py.utils import sidecar_paths
from py.utils.sidecar_paths import (
METADATA_SUFFIX,
ROOT_MAP_FILENAME,
get_configured_sidecar_root,
get_metadata_path,
get_preview_dir,
get_sidecar_dir,
get_sidecar_root,
get_storage_mode,
get_unmatched_sidecar_components,
is_centralized,
is_metadata_path,
relocate_root_map,
resolve_centralized_dir,
resolve_centralized_dir_for_dir,
resolve_metadata_path,
@@ -33,36 +27,10 @@ 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:
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
def model_roots(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> dict:
"""Point every config model root the sidecar module reads at tmp_path."""
@@ -100,17 +68,6 @@ def centralized(model_roots: dict, tmp_path: Path) -> Path:
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:
def test_default_mode_is_alongside(self):
assert get_storage_mode() == "alongside"
@@ -129,26 +86,6 @@ class TestAlongsideMode:
assert get_sidecar_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:
def test_is_metadata_path(self):
@@ -180,12 +117,13 @@ class TestSanitizePathComponent:
class TestCentralizedMode:
def test_mirror_layout(self, model_roots: dict, centralized: Path):
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"]))
metadata_path = get_metadata_path(str(model))
expected = os.path.join(
str(centralized), root_component, "styles", "anime", "model" + METADATA_SUFFIX
str(centralized), library, root_component, "styles", "anime", "model" + METADATA_SUFFIX
)
assert metadata_path == expected
assert get_preview_dir(str(model)) == os.path.dirname(expected)
@@ -217,7 +155,7 @@ class TestCentralizedMode:
assert root_mirror_component(str(model_roots["loras"])) != root_mirror_component(
str(other_root)
)
# Same root always maps to the same component (pinned identity).
# Same root always maps to the same component (stable hash).
assert root_mirror_component(str(model_roots["loras"])) == root_mirror_component(
str(model_roots["loras"]) + os.sep
)
@@ -233,10 +171,11 @@ class TestCentralizedMode:
[str(model_roots["loras"]), str(nested)],
raising=False,
)
library = get_settings_manager().get_active_library_name()
model = nested / "model.safetensors"
assert get_metadata_path(str(model)) == os.path.join(
str(centralized), root_mirror_component(str(nested)), "model" + METADATA_SUFFIX
str(centralized), library, root_mirror_component(str(nested)), "model" + METADATA_SUFFIX
)
def test_outside_roots_falls_back_to_alongside(
@@ -265,8 +204,10 @@ class TestCentralizedMode:
def test_resolve_centralized_dir_for_dir_root_maps_to_mirror_base(
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(
str(centralized), root_mirror_component(str(model_roots["loras"]))
str(centralized), library, root_mirror_component(str(model_roots["loras"]))
)
def test_empty_path_uses_default_sidecar_root(self, model_roots: dict, tmp_path: Path):
@@ -280,293 +221,6 @@ class TestCentralizedMode:
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:
"""Migration tooling resolves the mirror layout regardless of active mode."""
@@ -594,18 +248,17 @@ class TestModeIndependentResolution:
sidecar_root = tmp_path / "sidecars"
settings = get_settings_manager()
settings.set("sidecar_storage_mode", "alongside")
settings.set("sidecar_storage_path", str(sidecar_root))
library = settings.get_active_library_name()
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.
assert resolve_centralized_dir_for_dir(str(model_dir)) is None
assert resolve_centralized_dir_for_dir(
str(model_dir), sidecar_root=str(sidecar_root)
) == os.path.join(str(sidecar_root), component, "sub")
) == os.path.join(
str(sidecar_root), library, root_mirror_component(str(model_roots["loras"])), "sub"
)
class TestSettingsValidation: