mirror of
https://github.com/willmiao/ComfyUI-Lora-Manager.git
synced 2026-09-28 14:34:09 -03:00
Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
b90d60f043 | ||
|
|
c3a9350155 |
@@ -20,18 +20,21 @@ 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 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.
|
||||
- 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/`).
|
||||
- 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.
|
||||
|
||||
---
|
||||
|
||||
|
||||
@@ -2769,6 +2769,9 @@
|
||||
"filename_conflicts": {
|
||||
"title": "Konflikte durch doppelte Dateinamen"
|
||||
},
|
||||
"sidecar_mirror_orphans": {
|
||||
"title": "[TODO: Translate] Centralized Sidecars"
|
||||
},
|
||||
"ui_version": {
|
||||
"title": "UI-Version"
|
||||
}
|
||||
|
||||
@@ -2769,6 +2769,9 @@
|
||||
"filename_conflicts": {
|
||||
"title": "Duplicate Filename Conflicts"
|
||||
},
|
||||
"sidecar_mirror_orphans": {
|
||||
"title": "Centralized Sidecars"
|
||||
},
|
||||
"ui_version": {
|
||||
"title": "UI Version"
|
||||
}
|
||||
|
||||
@@ -2769,6 +2769,9 @@
|
||||
"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"
|
||||
}
|
||||
|
||||
@@ -2769,6 +2769,9 @@
|
||||
"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"
|
||||
}
|
||||
|
||||
@@ -2769,6 +2769,9 @@
|
||||
"filename_conflicts": {
|
||||
"title": "התנגשויות שמות קבצים כפולים"
|
||||
},
|
||||
"sidecar_mirror_orphans": {
|
||||
"title": "[TODO: Translate] Centralized Sidecars"
|
||||
},
|
||||
"ui_version": {
|
||||
"title": "גרסת הממשק"
|
||||
}
|
||||
|
||||
@@ -2769,6 +2769,9 @@
|
||||
"filename_conflicts": {
|
||||
"title": "ファイル名重複競合"
|
||||
},
|
||||
"sidecar_mirror_orphans": {
|
||||
"title": "[TODO: Translate] Centralized Sidecars"
|
||||
},
|
||||
"ui_version": {
|
||||
"title": "UI バージョン"
|
||||
}
|
||||
|
||||
@@ -2769,6 +2769,9 @@
|
||||
"filename_conflicts": {
|
||||
"title": "파일명 중복 충돌"
|
||||
},
|
||||
"sidecar_mirror_orphans": {
|
||||
"title": "[TODO: Translate] Centralized Sidecars"
|
||||
},
|
||||
"ui_version": {
|
||||
"title": "UI 버전"
|
||||
}
|
||||
|
||||
@@ -2769,6 +2769,9 @@
|
||||
"filename_conflicts": {
|
||||
"title": "Конфликты дублирующихся имён файлов"
|
||||
},
|
||||
"sidecar_mirror_orphans": {
|
||||
"title": "[TODO: Translate] Centralized Sidecars"
|
||||
},
|
||||
"ui_version": {
|
||||
"title": "Версия интерфейса"
|
||||
}
|
||||
|
||||
@@ -2769,6 +2769,9 @@
|
||||
"filename_conflicts": {
|
||||
"title": "文件名重复冲突"
|
||||
},
|
||||
"sidecar_mirror_orphans": {
|
||||
"title": "[TODO: Translate] Centralized Sidecars"
|
||||
},
|
||||
"ui_version": {
|
||||
"title": "UI 版本"
|
||||
}
|
||||
|
||||
@@ -2769,6 +2769,9 @@
|
||||
"filename_conflicts": {
|
||||
"title": "檔案名稱重複衝突"
|
||||
},
|
||||
"sidecar_mirror_orphans": {
|
||||
"title": "[TODO: Translate] Centralized Sidecars"
|
||||
},
|
||||
"ui_version": {
|
||||
"title": "UI 版本"
|
||||
}
|
||||
|
||||
@@ -75,6 +75,8 @@ 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
|
||||
@@ -806,6 +808,7 @@ 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),
|
||||
]
|
||||
|
||||
@@ -1042,6 +1045,71 @@ 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,7 +5,8 @@ Two storage layouts exist (see :mod:`py.utils.sidecar_paths`):
|
||||
- ``alongside``: ``<model_dir>/<name>.metadata.json`` and preview files live
|
||||
next to the model file.
|
||||
- ``centralized``: the same files live under the configured sidecar root,
|
||||
mirroring the library-relative directory structure.
|
||||
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
|
||||
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.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,
|
||||
)
|
||||
|
||||
@@ -199,14 +202,19 @@ 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:
|
||||
files.append(
|
||||
(os.path.join(dirpath, filename), os.path.join(target_dir, filename))
|
||||
)
|
||||
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)))
|
||||
|
||||
errors: List[Dict[str, str]] = []
|
||||
counters: Dict[str, Any] = {"moved": 0, "conflicts": 0}
|
||||
@@ -242,6 +250,23 @@ 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:
|
||||
|
||||
+767
-34
@@ -11,21 +11,39 @@ 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 the library-relative directory structure:
|
||||
``<root>/<library>/<root_basename-roothash>/<rel_dir>/<name>.metadata.json``.
|
||||
mirroring each model root's directory structure::
|
||||
|
||||
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.
|
||||
<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.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
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__)
|
||||
|
||||
@@ -36,6 +54,21 @@ 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."""
|
||||
@@ -184,16 +217,262 @@ 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 root_mirror_component(root_path: str) -> str:
|
||||
"""Return the mirror path component identifying a model root.
|
||||
def _normalized_roots(raw_roots: Iterable[str]) -> Tuple[str, ...]:
|
||||
"""Return the deduplicated, normalized, order-stable model roots."""
|
||||
|
||||
``<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.
|
||||
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.
|
||||
"""
|
||||
|
||||
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}"
|
||||
|
||||
|
||||
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>/<library>/<root_basename-roothash>/<rel_dir>``
|
||||
where ``rel_dir`` is the model's directory relative to the model root that
|
||||
The mirror layout is ``<sidecar_root>/<root_component>/<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.
|
||||
@@ -237,33 +948,55 @@ def resolve_centralized_dir_for_dir(
|
||||
return None
|
||||
|
||||
normalized_dir = _normalize_for_match(model_dir)
|
||||
|
||||
best_root: Optional[str] = None
|
||||
for candidate in _iter_model_roots():
|
||||
if not candidate:
|
||||
continue
|
||||
normalized = _normalize_for_match(candidate)
|
||||
if normalized_dir == normalized or normalized_dir.startswith(normalized + os.sep):
|
||||
if best_root is None or len(normalized) > len(best_root):
|
||||
best_root = normalized
|
||||
|
||||
normalized_roots = _normalized_roots(_iter_model_roots())
|
||||
best_root = _best_root_for(normalized_dir, normalized_roots)
|
||||
if best_root is None:
|
||||
return None
|
||||
|
||||
try:
|
||||
from ..services.settings_manager import get_settings_manager
|
||||
|
||||
library = get_settings_manager().get_active_library_name() or "default"
|
||||
except Exception: # pragma: no cover - defensive fallback
|
||||
library = "default"
|
||||
|
||||
rel_dir = os.path.relpath(normalized_dir, best_root)
|
||||
parts = [root, sanitize_path_component(library), root_mirror_component(best_root)]
|
||||
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:
|
||||
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.
|
||||
|
||||
|
||||
@@ -12,6 +12,7 @@ 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,
|
||||
@@ -209,6 +210,108 @@ 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,13 +19,22 @@ 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 root_mirror_component
|
||||
from py.utils.sidecar_paths import reset_root_map_cache, 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."""
|
||||
@@ -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:
|
||||
"""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))
|
||||
return sidecar_root.joinpath(library, component, *rel)
|
||||
component = root_mirror_component(str(library_root), sidecar_root=str(sidecar_root))
|
||||
return sidecar_root.joinpath(component, *rel)
|
||||
|
||||
|
||||
def _write_sidecar(
|
||||
|
||||
@@ -4,6 +4,7 @@ from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List
|
||||
|
||||
@@ -12,13 +13,26 @@ 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 root_mirror_component
|
||||
from py.utils.sidecar_paths import (
|
||||
get_metadata_path,
|
||||
reset_root_map_cache,
|
||||
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."""
|
||||
@@ -54,11 +68,10 @@ def _set_mode(mode: str) -> None:
|
||||
|
||||
|
||||
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))
|
||||
return sidecar_root.joinpath(library, component, *rel)
|
||||
component = root_mirror_component(str(library_root), sidecar_root=str(sidecar_root))
|
||||
return sidecar_root.joinpath(component, *rel)
|
||||
|
||||
|
||||
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()
|
||||
|
||||
|
||||
@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
|
||||
|
||||
@@ -2,7 +2,10 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
@@ -11,14 +14,17 @@ 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,
|
||||
@@ -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:
|
||||
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."""
|
||||
@@ -68,6 +100,17 @@ 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"
|
||||
@@ -86,6 +129,26 @@ 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):
|
||||
@@ -117,13 +180,12 @@ 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), library, root_component, "styles", "anime", "model" + METADATA_SUFFIX
|
||||
str(centralized), root_component, "styles", "anime", "model" + METADATA_SUFFIX
|
||||
)
|
||||
assert metadata_path == 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(
|
||||
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(
|
||||
str(model_roots["loras"]) + os.sep
|
||||
)
|
||||
@@ -171,11 +233,10 @@ 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), 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(
|
||||
@@ -204,10 +265,8 @@ 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), 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):
|
||||
@@ -221,6 +280,293 @@ 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."""
|
||||
|
||||
@@ -248,17 +594,18 @@ class TestModeIndependentResolution:
|
||||
sidecar_root = tmp_path / "sidecars"
|
||||
settings = get_settings_manager()
|
||||
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"
|
||||
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), library, root_mirror_component(str(model_roots["loras"])), "sub"
|
||||
)
|
||||
) == os.path.join(str(sidecar_root), component, "sub")
|
||||
|
||||
|
||||
class TestSettingsValidation:
|
||||
|
||||
Reference in New Issue
Block a user