test(memory): let the musl test fake os.confstr on Windows (#523)

The portability run on main fails since #523: Windows has no
os.confstr, so patching it without create=True raises before the test
runs.

__init__.py

@@ -92,10 +92,20 @@ try:
     routes = config.routes
     api = PromptManagerAPI()
     api.add_routes(routes)
-    # Temp files of preview writes a crash interrupted (#317), off the
-    # import path: the walk may cover a large thumbnails folder.
+
+    # The thumbnail location's startup check (#320), then the temp files of
+    # preview writes a crash interrupted (#317), off the import path: the
+    # walk may cover a large thumbnails folder.
+    # #320: the configured thumbnail location is in force before any request
+    # (no I/O); the thread below checks it.
+    api.use_preview_location()
+
+    def _preview_startup():
+        api.start_preview_location()
+        api.sweep_thumbnail_temps()
+
     threading.Thread(
-        target=api.sweep_thumbnail_temps,
+        target=_preview_startup,
         name="PromptManagerPreviewTempSweep",
         daemon=True,
     ).start()

database/library_move.py

@@ -78,6 +78,10 @@ MESSAGES = {
     ),
     "same-folder": "The new folder is the node folder itself.",
     "target-invalid": "The new folder cannot hold the library.",
+    "data-dir-served": (
+        "The new folder is inside Prompt Manager's thumbnail folder. Nothing was "
+        "moved."
+    ),
     "target-has-library": (
         "The new folder already holds a Prompt Manager library. Nothing was moved."
     ),

py/api/__init__.py

@@ -11,6 +11,7 @@ Split into domain-specific mixins for maintainability:
 - JobRoutesMixin: the session job registry behind Activity (py/api/jobs.py)
 - ImageTagJobsMixin: bulk Add tags / Auto-tag as image-tags jobs
 - ThumbnailJobsMixin: thumbnail plan, thumbnail job and the gallery stream on it
+- ThumbnailLocationMixin: the thumbnail location check and payloads (#320)
 - ThumbnailConvertMixin: Convert outdated thumbnails (#317), a thumbnail job mode
 - AutoThumbnailMixin: previews made on first view, on capture and after a scan
 - ReviewRoutesMixin: the Needs review queue, item detail and decisions
@@ -60,6 +61,7 @@ from .settings_routes import SettingsRoutesMixin
 from .job_routes import JobRoutesMixin
 from .image_tag_jobs import ImageTagJobsMixin
 from .thumbnail_jobs import ThumbnailJobsMixin
+from .thumbnail_location import ThumbnailLocationMixin
 from .thumbnail_convert import ThumbnailConvertMixin
 from .thumbnail_auto import AutoThumbnailMixin
 from .system import SystemRoutesMixin
@@ -338,6 +340,7 @@ class PromptManagerAPI(
     JobRoutesMixin,
     ImageTagJobsMixin,
     ThumbnailJobsMixin,
+    ThumbnailLocationMixin,
     ThumbnailConvertMixin,
     AutoThumbnailMixin,
     SystemRoutesMixin,
@@ -852,20 +855,11 @@ class PromptManagerAPI(
                         f"{url_quote(rel_path.as_posix(), safe='/')}"
                     )
 
-                    # Any preview the file owns, the current format first (#317)
-                    from . import thumbnail_settings
-                    from .prompts import thumbnail_url_for
+                    # Any preview the file owns, the current format first
+                    # (#317); no store I/O here in folder mode (#320).
+                    from .images import listing_thumbnail_url
 
-                    existing = thumbnail_settings.existing_thumbnail(
-                        output_path, rel_path
-                    )
-                    thumbnail_url = None
-                    if existing is not None:
-                        thumb_rel = (
-                            thumbnail_settings.thumbnail_stem_rel(rel_path)
-                            + existing.suffix
-                        )
-                        thumbnail_url = thumbnail_url_for(thumb_rel, existing)
+                    thumbnail_url = listing_thumbnail_url(output_path, rel_path)
                     if thumbnail_url:
                         image["thumbnail_url"] = thumbnail_url
                     break  # Found matching root, stop searching

py/api/autotag_routes.py

@@ -19,8 +19,8 @@ from .jobs import (
 )
 
 from .maintenance import disk_plan
-from .prompts import thumbnail_url_for
-from . import preview_store, thumbnail_settings
+from .images import listing_thumbnail_url
+from . import preview_store
 from .. import model_gate
 
 try:
@@ -1371,24 +1371,21 @@ class AutotagRoutesMixin:
             files = await self._run_in_executor(
                 self._without_removed, output_path, files
             )
-            store = preview_store.store_dir(output_path)
-            has_thumbnails = store is not None and store.is_dir()
+            location = preview_store.current()
+            if location.mode == preview_store.FOLDER_MODE:
+                has_thumbnails = location.available  # no store I/O here (#320)
+            else:
+                store = preview_store.store_dir(output_path)
+                has_thumbnails = store is not None and store.is_dir()
 
             images = []
             for image_path in files:
                 rel_path = image_path.relative_to(output_path)
                 thumbnail_url = None
                 if has_thumbnails:
-                    # Any preview the file owns, the current format first (#317)
-                    existing = thumbnail_settings.existing_thumbnail(
-                        output_path, rel_path
-                    )
-                    if existing is not None:
-                        thumb_rel = (
-                            thumbnail_settings.thumbnail_stem_rel(rel_path)
-                            + existing.suffix
-                        )
-                        thumbnail_url = thumbnail_url_for(thumb_rel, existing)
+                    # Any preview the file owns, the current format first
+                    # (#317); in folder mode without store I/O (#320).
+                    thumbnail_url = listing_thumbnail_url(output_path, rel_path)
                 images.append(
                     {
                         "filename": image_path.name,

py/api/history_current.py

@@ -244,19 +244,36 @@ def resolve_current_candidate(
         return _unavailable("temporary_unavailable", error=True)
 
 
+def _target_authorized(target, namespace, scope, store_root=None) -> bool:
+    """A preview in the thumbnail folder (#320, folder mode) is authorized by
+    the store of its source (``preview_store.owns``: inside the key folder,
+    no link); everything else by the output namespace and the roots."""
+    if (
+        store_root is not None
+        and preview_store.current().mode == preview_store.FOLDER_MODE
+    ):
+        return preview_store.owns(store_root, target)
+    return _authorized(target, namespace, scope)
+
+
 def _open_current_target(
-    target: Path, namespace: Path, scope: CurrentOutputScope, preview: bool = False
+    target: Path,
+    namespace: Path,
+    scope: CurrentOutputScope,
+    preview: bool = False,
+    store_root: Path | None = None,
 ) -> BinaryIO:
     """Open only the canonical target and verify authority against the actual handle.
 
-    A *preview* (a file under ``thumbnails/``) may also be JPEG XL (#317)."""
+    A *preview* (a file under ``thumbnails/``) may also be JPEG XL (#317); in
+    folder mode it is authorized by the store of *store_root* (#320)."""
     if _canonical(target) != target:
         raise CurrentLocationUnavailable("not_authorized", stale=True)
     if in_trash_below_roots(target, [namespace]):
         # Also at the open: the cover chooses its preview after the candidate
         # check, and a preview can be a link into a trash folder (#543).
         raise CurrentLocationUnavailable("missing")
-    if not _authorized(target, namespace, scope):
+    if not _target_authorized(target, namespace, scope, store_root):
         raise CurrentLocationUnavailable("not_authorized")
     suffix = target.suffix.lower()
     if suffix not in SERVABLE_EXTENSIONS and not (
@@ -289,7 +306,7 @@ def _open_current_target(
         if (
             _canonical(target) != target
             or (observed.st_dev, observed.st_ino) != (current.st_dev, current.st_ino)
-            or not _authorized(target, namespace, scope)
+            or not _target_authorized(target, namespace, scope, store_root)
         ):
             raise CurrentLocationUnavailable("not_authorized", stale=True)
         return os.fdopen(fd, "rb")
@@ -379,7 +396,7 @@ def _cover_target(resolved, ensure_preview, placeholder, accepts_jxl):
         return _canonical(preview)
     if preview is not None:
         return _canonical(preview)
-    preview = media_preview_path(resolved.gallery_root, relative)
+    preview = media_preview_path(resolved.gallery_root, relative, create=True)
     if (
         ensure_preview is not None
         and preview is not None
@@ -420,7 +437,13 @@ def _prepare_current_media(
             return chosen
         if chosen is not None:
             target, is_preview = chosen, True
-    handle = _open_current_target(target, resolved.namespace, scope, is_preview)
+    handle = _open_current_target(
+        target,
+        resolved.namespace,
+        scope,
+        is_preview,
+        resolved.gallery_root if is_preview else None,
+    )
     try:
         return _OpenedCurrentMedia(
             handle,

py/api/images.py

@@ -35,6 +35,7 @@ from .gallery_listing import (
     walk_root,
 )
 from . import preview_store, thumbnail_settings
+from .thumbnail_location import location_refusal
 from .thumbnail_encode import (
     PREVIEW_HELD,
     EncoderClosed,
@@ -1076,10 +1077,14 @@ def media_thumbnail_rel(rel_path, settings=None):
     return thumbnail_settings.thumbnail_rel(rel_path, settings.extension)
 
 
-def media_preview_path(output_path, rel_path, settings=None):
-    """Where the preview of *rel_path* is written now (``preview_store``)."""
+def media_preview_path(output_path, rel_path, settings=None, *, create=False):
+    """Where the preview of *rel_path* is written now (``preview_store``);
+    *create* for a writer (a new source gets its key). None when nothing
+    may be written (the thumbnail folder is unavailable)."""
     settings = settings or writer_settings()
-    return preview_store.preview_path(output_path, rel_path, settings.extension)
+    return preview_store.preview_path(
+        output_path, rel_path, settings.extension, create=create
+    )
 
 
 def existing_media_thumbnail(output_path, rel_path):
@@ -1091,6 +1096,27 @@ def existing_media_thumbnail(output_path, rel_path):
     )
 
 
+def listing_thumbnail_url(output_path, rel_path):
+    """The thumbnail URL of an older listing (``_enrich_images``, the
+    auto-tag scan), or None. These run on the event loop: in folder mode
+    (#320) the store is never touched there, so the URL names the preview
+    as it is written now and the serve route finds or makes it off the loop
+    (None while the thumbnail folder is unavailable)."""
+    from .prompts import thumbnail_url_for
+
+    location = preview_store.current()
+    if location.mode == preview_store.FOLDER_MODE:
+        if not location.available:
+            return None
+        quoted = urllib.parse.quote(media_thumbnail_rel(rel_path), safe="/")
+        return f"/prompt_manager/images/serve/{quoted}"
+    existing = thumbnail_settings.existing_thumbnail(output_path, rel_path)
+    if existing is None:
+        return None
+    thumb_rel = thumbnail_settings.thumbnail_stem_rel(rel_path) + existing.suffix
+    return thumbnail_url_for(thumb_rel, existing)
+
+
 def media_thumbnail(output_path, rel_path, root_index, *, root_scope=None):
     """``(thumbnail_url, ready)`` for *rel_path* under *output_path* (a few
     blocking stats). The Images listing, the Needs review queue cards and the
@@ -2813,7 +2839,31 @@ class ImageRoutesMixin:
                 if preview_store.is_preview_text(filepath)
                 else None
             )
+            location = preview_store.current()
+            if location.mode == preview_store.FOLDER_MODE:
+                if preview_parts is None and await self._run_image(
+                    self._in_source_thumbnails, output_dirs, filepath
+                ):
+                    # Previews live in the thumbnail folder: anything else
+                    # under a source's thumbnails/ is a leftover (#320), in
+                    # any case on a case-insensitive disk.
+                    return _forbidden()
+                if preview_parts is not None:
+                    # Store I/O never runs on the event loop (a share may
+                    # stall): the lookup runs on the image pool.
+                    found = await self._run_image(
+                        self._find_preview, output_dirs, preview_parts, location
+                    )
+                    if found == "forbidden":
+                        return _forbidden()
+                    if found is not None:
+                        return self._file_response(found)
             for output_path in output_dirs:
+                if (
+                    preview_parts is not None
+                    and location.mode == preview_store.FOLDER_MODE
+                ):
+                    break  # looked up above
                 if preview_parts is None:
                     image_path = output_path / filepath
                 else:
@@ -2856,6 +2906,48 @@ class ImageRoutesMixin:
                 {"success": False, "error": _safe_error(e)}, status=500
             )
 
+    @staticmethod
+    def _in_source_thumbnails(output_dirs, filepath):
+        """Whether *filepath* lies in a source's ``thumbnails`` folder: its
+        first part is ``thumbnails`` exactly, or another case of it that is
+        the same folder (a case-insensitive disk). ``Thumbnails/`` media on
+        a case-sensitive disk is another folder and stays served."""
+        first = filepath.split("/", 1)[0]
+        if first == "thumbnails":
+            return True
+        if first.casefold() != "thumbnails":
+            return False
+        for output_path in output_dirs:
+            spelled = output_path / first
+            cache = preview_store.source_folder(output_path)
+            try:
+                if spelled.is_dir() and cache.is_dir() and spelled.samefile(cache):
+                    return True
+            except OSError:
+                continue
+        return False
+
+    @staticmethod
+    def _find_preview(output_dirs, parts, location):
+        """The preview file *parts* names in folder mode, ``"forbidden"``
+        (it resolves out of its store or through a link) or None (blocking;
+        the first root whose store holds it, in ``_select_root`` order)."""
+        for output_path in output_dirs:
+            image_path = preview_store.request_path(output_path, parts, location)
+            if image_path is None:
+                continue
+            try:
+                with preview_store.store_io():
+                    if not os.path.lexists(image_path):
+                        continue
+                    if not preview_store.owns(output_path, image_path, location):
+                        return "forbidden"
+                    if image_path.is_file():
+                        return image_path
+            except OSError:
+                return None
+        return None
+
     def _is_recorded_media(self, path) -> bool:
         """Whether the database records ``path`` as a linked image.
 
@@ -2982,6 +3074,9 @@ class ImageRoutesMixin:
         data, error = await _json_body_or_error(request)
         if error is not None:
             return error
+        refused = location_refusal()
+        if refused is not None:
+            return refused
         return await self._run_held(
             THUMBNAILS_JOB_KIND,
             THUMBNAIL_GENERATION_TITLE,
@@ -3071,9 +3166,9 @@ class ImageRoutesMixin:
             if stop is not None and stop():
                 break
             root = Path(root)
-            store = preview_store.store_dir(root)
+            store = preview_store.ensure_store(root)
             if store is None:
-                continue
+                continue  # the thumbnail folder is unavailable
             plan = self._plan_thumbnails(
                 root,
                 store,
@@ -3115,6 +3210,8 @@ class ImageRoutesMixin:
                     "format": writer_settings().format,
                     # Files whose automatic preview failed this session (#249).
                     "failed": self._auto_thumbnail_failures(),
+                    # #320: where previews live (no path here, D6A).
+                    "location": self._thumbnail_location_payload(with_path=False),
                     "collection_id": hashlib.sha256(collection.encode()).hexdigest()[
                         :16
                     ],
@@ -3163,6 +3260,8 @@ class ImageRoutesMixin:
         thumbnails_dir = preview_store.given_store(output_path, thumbnails_dir)
         plan = {key: 0 for key in _PLAN_COUNTS}
         plan.update(targets=[], fresh_files=[])
+        if thumbnails_dir is None:
+            return plan  # the thumbnail folder is unavailable
         folders = {}
         removed = self._removal_check_under(output_path)
         for src in _iter_media_files(output_path, MAX_GALLERY_FILES):
@@ -3265,7 +3364,7 @@ class ImageRoutesMixin:
                 return {**result, "action": "error", "error": "File no longer exists"}
             if not rebuild and _thumbnail_is_fresh(src, dst):
                 return {**result, "action": "skipped", "path": str(dst)}
-            dst.parent.mkdir(parents=True, exist_ok=True)
+            preview_store.make_parent(dst)
             if is_video:
                 made = self._generate_video_thumbnail(
                     src,
@@ -3470,8 +3569,12 @@ class ImageRoutesMixin:
 
         Invalid thumbnail directories return 400; unavailable gallery roots
         return 404, and unexpected filesystem failures return 500. Holds the
-        registry as guarded work, like generation.
+        registry as guarded work, like generation. Refused (409) while the
+        thumbnail folder is unavailable (#320).
         """
+        refused = location_refusal()
+        if refused is not None:
+            return refused
         return await self._run_held(
             THUMBNAILS_JOB_KIND,
             THUMBNAIL_CLEAR_TITLE,
@@ -3536,7 +3639,11 @@ class ImageRoutesMixin:
             return 0
         for output_path in output_dirs:
             folder = preview_store.store_dir(output_path)
-            if folder is None or folder.is_symlink() or not folder.is_dir():
+            if (
+                folder is None
+                or preview_store.store_is_link(folder)
+                or not folder.is_dir()
+            ):
                 continue
             removed += sweep_stray_temps(folder)[0]
         if removed:
@@ -3561,7 +3668,7 @@ class ImageRoutesMixin:
                 continue
             container = preview_store.store_container(output_path)
             if (
-                thumbnails_dir.is_symlink()
+                preview_store.store_is_link(thumbnails_dir)
                 or not _is_within(thumbnails_dir, container)
                 or _canonical(thumbnails_dir) == _canonical(container)
             ):
@@ -3578,7 +3685,15 @@ class ImageRoutesMixin:
             temps, temp_bytes = sweep_stray_temps(thumbnails_dir, min_age=0)
             cleared_count += temps
             cleared_size += temp_bytes
-            for root, _dirs, files in os.walk(thumbnails_dir):
+            folder_mode = preview_store.current().mode == preview_store.FOLDER_MODE
+            for root, dirs, files in os.walk(thumbnails_dir):
+                if folder_mode:
+                    # Never into a junction or reparse point (#320).
+                    dirs[:] = [
+                        d
+                        for d in dirs
+                        if not preview_store.store_is_link(Path(root) / d)
+                    ]
                 for name in files:
                     path = Path(root) / name
                     if (
@@ -3586,6 +3701,7 @@ class ImageRoutesMixin:
                         or path.suffix.lower()
                         not in thumbnail_settings.THUMBNAIL_EXTENSIONS
                         or path.is_symlink()
+                        or not preview_store.may_write(path)
                     ):
                         continue
                     try:

py/api/library_location.py

@@ -50,6 +50,7 @@ _REFUSAL_STATUS = {
     "not-movable": 400,
     "same-folder": 400,
     "target-invalid": 400,
+    "data-dir-served": 400,
     "target-has-library": 409,
     "target-conflict": 409,
     "insufficient-space": 507,
@@ -74,6 +75,16 @@ def _failure(message, status, code):
     )
 
 
+def _inside_thumbnail_folder(path):
+    """Whether *path* lies in the thumbnail folder in use (#320)."""
+    from . import preview_store
+
+    setting = preview_store.setting()
+    if setting.mode != preview_store.FOLDER_MODE or not setting.path:
+        return False
+    return ll._inside_any(path, [setting.path])
+
+
 def _plural(count, one, many=None):
     return f"{count:,} {one if count == 1 else (many or one + 's')}"
 
@@ -272,6 +283,8 @@ class LibraryLocationRoutesMixin:
     def _plan_library_move(self):
         """The checked plan (blocking); raises MoveFailed."""
         plan = lm.plan_move(self._live_location())
+        if _inside_thumbnail_folder(plan.target_dir):
+            raise lm.MoveFailed("data-dir-served")
         if lm.space_problem(plan) is not None:
             raise lm.MoveFailed("insufficient-space")
         return plan

py/api/preview_location.py

@@ -0,0 +1,643 @@
+"""The thumbnail location setting (#320): parsing, validation, source keys
+and the thumbnail folder's marker.
+
+``preview_store`` holds the runtime state and maps paths; this module holds
+the rules it applies. Nothing here keeps state.
+
+Schema (config.json ``thumbnails.location``)::
+
+    {"mode": "folder", "path": "D:\\\\Thumbs", "id": "<uuid4>",
+     "install": "<uuid4>", "cleanup": []}
+
+The default (and a missing key) is ``{"mode": "source"}``: previews inside
+each source. An invalid value never falls back to the sources: it parses as
+a folder location with problem ``invalid``, so nothing is written anywhere.
+
+Layout in folder mode: ``<folder>/<source key>/<rel dir>/<stem>_thumb<ext>``.
+A source key is a readable slug plus a short hash of the source's identity
+(``output-3f9a2c1b``), assigned once and stored in the marker
+``.promptmanager-thumbnails.json`` with the identity it belongs to.
+"""
+
+import errno
+import hashlib
+import json
+import logging
+import ntpath
+import os
+import posixpath
+import re
+import stat as stat_module
+import threading
+import unicodedata
+import uuid
+from dataclasses import dataclass, field
+
+try:
+    from ...utils import library_location
+except ImportError:  # tests import the package top-level
+    from utils import library_location
+
+logger = logging.getLogger("prompt_manager.thumbnails")
+
+SOURCE_MODE = "source"
+FOLDER_MODE = "folder"
+MARKER_NAME = ".promptmanager-thumbnails.json"
+MARKER_VERSION = 1
+# A source key: one folder name, a slug and a 8-16 hex hash (``source_key``).
+KEY_PATTERN = re.compile(r"^[a-z0-9_-]{1,24}-[0-9a-f]{8,16}$")
+# A marker is a few keys; anything larger is not ours.
+MAX_MARKER_BYTES = 4 * 2**20
+# OS metadata files a folder may hold and still count as empty.
+OS_METADATA = frozenset({"thumbs.db", "desktop.ini", ".ds_store"})
+KEY_HASH_LENGTHS = (8, 12, 16)
+SLUG_LENGTH = 24
+# Folder + the longest key (24 + 1 + 16) + this margin must fit (Windows).
+WINDOWS_FOLDER_LIMIT = 247
+LONGEST_KEY = SLUG_LENGTH + 1 + KEY_HASH_LENGTHS[-1]
+PATH_MARGIN = 40
+# A validation stat on a share that does not answer in time is unavailable.
+VALIDATION_TIMEOUT = 5.0
+FILE_ATTRIBUTE_REPARSE_POINT = 0x400
+
+_RESERVED = re.compile(r"^(con|prn|aux|nul|com[1-9]|lpt[1-9])(\..*)?$", re.IGNORECASE)
+_BAD_CHARS = re.compile(r'[<>:"|?*\x00-\x1f]')
+
+# Problem codes, their HTTP status and the text the UI shows (no path).
+MESSAGES = {
+    "location-not-absolute": (
+        400,
+        "Enter a full folder path, such as D:\\Thumbnails or /mnt/fast/thumbnails.",
+    ),
+    "location-device-path": (
+        400,
+        "Enter the folder as a drive letter or \\\\server\\share path, "
+        "without \\\\?\\.",
+    ),
+    "location-invalid-name": (
+        400,
+        "This folder name can't be used on Windows. Choose another name.",
+    ),
+    "location-too-broad": (
+        400,
+        "Choose a folder inside a drive or share, not the drive itself.",
+    ),
+    "location-served": (
+        400,
+        "Choose a folder outside your image sources and ComfyUI's output, input "
+        "and temp folders.",
+    ),
+    "location-node-folder": (
+        400,
+        "Choose a folder outside the Prompt Manager node folder.",
+    ),
+    "location-contains-source": (
+        400,
+        "This folder contains an image source. Choose a folder beside it instead.",
+    ),
+    "location-same": (400, "Thumbnails are already stored here."),
+    "location-parent-missing": (
+        400,
+        "The folder that would hold it doesn't exist. Check the path and that its "
+        "drive is connected.",
+    ),
+    "location-unavailable": (
+        409,
+        "This folder can't be reached right now. Check that its drive or share is "
+        "connected.",
+    ),
+    "location-not-empty": (
+        409,
+        "This folder already holds other files. Choose an empty folder or one "
+        "Prompt Manager used before.",
+    ),
+    "location-not-writable": (
+        409,
+        "Prompt Manager can't write to this folder. Check its permissions.",
+    ),
+    "location-path-too-long": (
+        400,
+        "This path is too long for Windows. Choose a shorter folder path.",
+    ),
+    "location-needs-move": (
+        409,
+        "Use Move thumbnails in System › Sources & scans.",
+    ),
+}
+HOLDS_THUMBNAILS_REASON = "This folder holds Prompt Manager's thumbnails."
+
+
+def message(code):
+    """``(status, text)`` of a problem code."""
+    return MESSAGES[code]
+
+
+# ── parsing ──────────────────────────────────────────────────────────
+
+
+@dataclass(frozen=True)
+class ParsedLocation:
+    """A ``thumbnails.location`` value, checked as data (no filesystem)."""
+
+    mode: str = SOURCE_MODE
+    path: str | None = None
+    id: str | None = None
+    install: str | None = None
+    cleanup: tuple = field(default_factory=tuple)
+    problem: str | None = None
+
+    def to_config(self):
+        """The config.json form (``invalid`` values are never written)."""
+        if self.mode == SOURCE_MODE:
+            value = {"mode": SOURCE_MODE}
+        else:
+            value = {"mode": FOLDER_MODE, "path": self.path, "id": self.id}
+        if self.install:
+            value["install"] = self.install
+        value["cleanup"] = list(self.cleanup)
+        return value
+
+
+def _cleanup(value):
+    if not isinstance(value, list):
+        return ()
+    return tuple(item for item in value if isinstance(item, dict))
+
+
+def parse_location(value, *, flavor=os.path):
+    """The location a ``thumbnails.location`` value names.
+
+    Missing: inside each source. Anything else that is not a well-formed
+    location (an unknown mode, a relative path, a missing id) becomes a
+    folder location with problem ``invalid``: never a silent fall-back.
+    """
+    if value is None:
+        return ParsedLocation()
+    if not isinstance(value, dict):
+        return ParsedLocation(mode=FOLDER_MODE, problem="invalid")
+    mode = value.get("mode", SOURCE_MODE)
+    install = value.get("install") if isinstance(value.get("install"), str) else None
+    cleanup = _cleanup(value.get("cleanup"))
+    if mode == SOURCE_MODE:
+        return ParsedLocation(install=install, cleanup=cleanup)
+    if mode != FOLDER_MODE:
+        return ParsedLocation(mode=FOLDER_MODE, problem="invalid", cleanup=cleanup)
+    path, marker_id = value.get("path"), value.get("id")
+    if (
+        not isinstance(path, str)
+        or form_problem(path, flavor=flavor) is not None
+        or not isinstance(marker_id, str)
+        or not marker_id.strip()
+    ):
+        return ParsedLocation(
+            mode=FOLDER_MODE,
+            path=path if isinstance(path, str) else None,
+            problem="invalid",
+            cleanup=cleanup,
+        )
+    return ParsedLocation(
+        mode=FOLDER_MODE,
+        path=flavor.abspath(path),
+        id=marker_id,
+        install=install,
+        cleanup=cleanup,
+    )
+
+
+# ── form (text only) ─────────────────────────────────────────────────
+
+
+def _is_windows(flavor):
+    return flavor is ntpath or getattr(flavor, "__name__", "") == "ntpath"
+
+
+def form_problem(path, *, flavor=os.path):
+    """The text-only problem of *path*, or None (no filesystem access).
+
+    ``\\\\?\\`` and ``\\\\.\\`` prefixes, relative, drive-relative (``C:x``)
+    and root-relative (``\\x``) spellings, and names Windows cannot hold
+    (checked on every OS: the folder may be moved to Windows).
+    """
+    if not isinstance(path, str) or not path.strip() or path != path.lstrip():
+        return "location-not-absolute"
+    if path.replace("/", "\\").startswith(("\\\\?\\", "\\\\.\\")):
+        return "location-device-path"
+    if _is_windows(flavor):
+        drive, rest = ntpath.splitdrive(path)
+        if not drive or not rest.startswith(("\\", "/")):
+            return "location-not-absolute"
+    elif not posixpath.isabs(path):
+        return "location-not-absolute"
+    drive, rest = flavor.splitdrive(path)
+    for part in re.split(r"[\\/]+", rest):
+        if not part:
+            continue
+        if (
+            _RESERVED.match(part)
+            or part.endswith((".", " "))
+            or _BAD_CHARS.search(part)
+        ):
+            return "location-invalid-name"
+    return None
+
+
+def is_too_broad(path, *, home=None, flavor=os.path):
+    """A filesystem or share root, or the home folder itself."""
+    normalized = flavor.normpath(path)
+    drive, rest = flavor.splitdrive(normalized)
+    if rest in ("", "\\", "/"):
+        return True
+    home = home if home is not None else os.path.expanduser("~")
+    return bool(home) and flavor.normcase(normalized) == flavor.normcase(
+        flavor.normpath(home)
+    )
+
+
+def too_long_for_windows(path, long_paths=None):
+    """Folder + the longest key + a margin over the folder limit (Windows
+    without long paths; *long_paths* injects the registry answer)."""
+    if long_paths is None:
+        long_paths = long_paths_enabled()
+    if long_paths:
+        return False
+    return len(path) + 1 + LONGEST_KEY + PATH_MARGIN > WINDOWS_FOLDER_LIMIT
+
+
+def long_paths_enabled():
+    """``LongPathsEnabled`` on Windows (True on other systems)."""
+    if os.name != "nt":
+        return True
+    try:
+        import winreg
+
+        with winreg.OpenKey(
+            winreg.HKEY_LOCAL_MACHINE,
+            r"SYSTEM\CurrentControlSet\Control\FileSystem",
+        ) as key:
+            return winreg.QueryValueEx(key, "LongPathsEnabled")[0] == 1
+    except OSError:
+        return False
+
+
+# ── links ────────────────────────────────────────────────────────────
+
+
+def is_link_like(path):
+    """A symlink, a junction or any reparse point (one lstat), or missing.
+
+    Raises OSError other than "not found" (an unreachable share).
+    """
+    try:
+        info = os.lstat(path)
+    except FileNotFoundError:
+        return False
+    if stat_module.S_ISLNK(info.st_mode):
+        return True
+    attributes = getattr(info, "st_file_attributes", 0)
+    if attributes & FILE_ATTRIBUTE_REPARSE_POINT:
+        return True
+    is_junction = getattr(os.path, "isjunction", None)
+    return bool(is_junction and is_junction(path))
+
+
+# ── source keys ──────────────────────────────────────────────────────
+
+
+def slug(name):
+    """``NFKD`` to ASCII, lower case, runs outside ``[a-z0-9_]`` as ``-``,
+    trimmed and cut to 24 characters; empty becomes ``source``."""
+    text = unicodedata.normalize("NFKD", str(name))
+    text = text.encode("ascii", "ignore").decode("ascii").lower()
+    text = re.sub(r"[^a-z0-9_]+", "-", text).strip("-")
+    text = text[:SLUG_LENGTH].strip("-")
+    return text or "source"
+
+
+def source_key(spelling, identity, taken):
+    """The key of a source: ``slug(basename) + "-" + sha256(identity)[:n]``.
+
+    *taken* maps existing keys to their identity; a key taken by another
+    identity grows its hash to 12, then 16 characters.
+    """
+    base = slug(ntpath.basename(str(spelling).rstrip("\\/")))
+    digest = hashlib.sha256(str(identity).encode("utf-8")).hexdigest()
+    for length in KEY_HASH_LENGTHS:
+        key = f"{base}-{digest[:length]}"
+        owner = taken.get(key)
+        if owner is None or owner == identity:
+            return key
+    raise ValueError("no free source key")
+
+
+def file_identity(path):
+    """``[st_dev, st_ino]`` of an existing folder, or None (no file ids)."""
+    try:
+        info = os.stat(path)
+    except (OSError, ValueError):
+        return None
+    if not info.st_ino:
+        return None
+    return [info.st_dev, info.st_ino]
+
+
+def find_key(keys, spelling, identity, dev_ino):
+    """The marker key of a source: the same configured spelling, then the
+    same identity text, then the same file id; or None."""
+    keys = {k: v for k, v in keys.items() if valid_key(k) and isinstance(v, dict)}
+    for key, entry in keys.items():
+        if entry.get("path") == spelling:
+            return key
+    for key, entry in keys.items():
+        if entry.get("identity") == identity:
+            return key
+    if dev_ino:
+        for key, entry in keys.items():
+            if entry.get("dev_ino") == list(dev_ino):
+                return key
+    return None
+
+
+# ── the marker ───────────────────────────────────────────────────────
+
+
+def marker_path(folder):
+    return os.path.join(folder, MARKER_NAME)
+
+
+def read_marker(folder):
+    """The marker in *folder* as a dict, or None (absent or unreadable as
+    ours). OSError other than "not found" propagates (an unreachable share)."""
+    try:
+        with open(marker_path(folder), "r", encoding="utf-8") as handle:
+            text = handle.read(MAX_MARKER_BYTES + 1)
+    except FileNotFoundError:
+        return None
+    except (ValueError, UnicodeDecodeError):
+        return None
+    if len(text) > MAX_MARKER_BYTES:
+        return None
+    try:
+        data = json.loads(text)
+    except ValueError:
+        return None
+    if not isinstance(data, dict) or not isinstance(data.get("id"), str):
+        return None
+    data["keys"] = valid_keys(data.get("keys"))
+    return data
+
+
+def valid_key(key):
+    """Whether *key* can name a source's folder: one folder name of the
+    ``source_key`` form, never a separator, ``..`` or a drive."""
+    return isinstance(key, str) and KEY_PATTERN.fullmatch(key) is not None
+
+
+def valid_keys(keys):
+    """The marker's keys that are safe to join onto the folder; a hand-edited
+    or foreign entry that is not is dropped (and logged), never followed."""
+    if not isinstance(keys, dict):
+        return {}
+    kept = {}
+    for key, entry in keys.items():
+        if valid_key(key) and isinstance(entry, dict):
+            kept[key] = entry
+        else:
+            logger.warning(
+                "Ignoring a thumbnail folder marker entry that is not a source " "key"
+            )
+    return kept
+
+
+def write_marker(folder, mutate):
+    """Re-read the marker, apply ``mutate(marker)`` and write it atomically
+    (a temp sibling renamed over it). Callers hold ``preview_store.write_lock``.
+    Returns the marker written."""
+    current = read_marker(folder) or {}
+    marker = {
+        "version": MARKER_VERSION,
+        "id": current.get("id"),
+        "created_by_pm": bool(current.get("created_by_pm", False)),
+        "keys": dict(current.get("keys") or {}),
+    }
+    mutate(marker)
+    if not marker.get("id"):
+        raise ValueError("a marker needs an id")
+    target = marker_path(folder)
+    temp = os.path.join(
+        folder, f".{MARKER_NAME}.{os.getpid()}-{threading.get_ident()}.tmp"
+    )
+    try:
+        with open(temp, "w", encoding="utf-8") as handle:
+            json.dump(marker, handle, indent=2, sort_keys=True)
+            handle.flush()
+            os.fsync(handle.fileno())
+        os.replace(temp, target)
+    finally:
+        try:
+            os.unlink(temp)
+        except OSError:
+            pass  # gone with the rename; a leftover is swept with the temps
+    return marker
+
+
+def new_id():
+    return str(uuid.uuid4())
+
+
+def init_folder(folder, *, created_by_pm=False):
+    """The marker id of *folder*: an existing marker is adopted; a folder
+    without one gets a new marker (new id, no keys)."""
+    existing = read_marker(folder)
+    if existing is not None:
+        return existing["id"]
+
+    def create(marker):
+        if not marker.get("id"):
+            marker["id"] = new_id()
+            marker["created_by_pm"] = created_by_pm
+
+    return write_marker(folder, create)["id"]
+
+
+def folder_state(path, expected_id):
+    """``(available, problem, marker)`` of a configured thumbnail folder.
+
+    Available: the folder exists, is a directory and not a link, junction or
+    reparse point, and holds our marker with *expected_id*. Raises OSError
+    when the folder cannot be reached (an I/O error other than not found).
+    """
+    try:
+        info = os.lstat(path)
+    except FileNotFoundError:
+        return False, "missing", None
+    except OSError as error:
+        if error.errno in (errno.ENOENT, errno.ENOTDIR):
+            return False, "missing", None
+        raise
+    if is_link_like(path):
+        return False, "invalid", None
+    if not stat_module.S_ISDIR(info.st_mode):
+        return False, "invalid", None
+    marker = read_marker(path)
+    if marker is None:
+        return False, "missing", None
+    if marker["id"] != expected_id:
+        return False, "marker-mismatch", marker
+    return True, None, marker
+
+
+def holds_only_metadata(path):
+    """Whether an existing folder is empty but for OS metadata files."""
+    with os.scandir(path) as entries:
+        for entry in entries:
+            name = entry.name.lower()
+            if name in OS_METADATA or name.startswith("._"):
+                continue
+            return False
+    return True
+
+
+# ── validation ───────────────────────────────────────────────────────
+
+
+@dataclass(frozen=True)
+class Context:
+    """What a location is checked against (gathered by the caller)."""
+
+    roots: tuple = ()  # every image source of every root list
+    contained: tuple = ()  # ComfyUI output/input/temp and the library folder
+    node_folder: str | None = None
+    default_path: str | None = None
+    current_path: str | None = None  # the folder in use (folder mode)
+    home: str | None = None
+    flavor: object = os.path
+    long_paths: bool | None = None
+    environ: object = None
+    # ComfyUI's folder_paths module (None: none; unset: imported).
+    folder_paths: object = library_location._UNSET
+
+
+def _timed(func, timeout):
+    """``func()`` on a daemon thread; raises TimeoutError past *timeout*.
+    A stuck call is left to finish on its own thread."""
+    result = {}
+    done = threading.Event()
+
+    def run():
+        try:
+            result["value"] = func()
+        except BaseException as error:  # handed to the caller
+            result["error"] = error
+        finally:
+            done.set()
+
+    threading.Thread(target=run, name="pm-preview-stat", daemon=True).start()
+    if not done.wait(timeout):
+        raise TimeoutError("the folder did not answer in time")
+    if "error" in result:
+        raise result["error"]
+    return result.get("value")
+
+
+def _inside(path, folders):
+    return library_location._inside_any(path, folders)
+
+
+def _same(first, second):
+    return library_location.same_folder(first, second) or os.path.normcase(
+        os.path.realpath(first)
+    ) == os.path.normcase(os.path.realpath(second))
+
+
+def _served(path, context):
+    environ = os.environ if context.environ is None else context.environ
+    return (
+        library_location.data_dir_problem(
+            path,
+            folder_paths=context.folder_paths,
+            environ=environ,
+            gallery_roots=list(context.roots),
+        )
+        == "data-dir-served"
+    )
+
+
+def _probe_write(path):
+    """Create and delete a probe file in *path*; False when refused."""
+    probe = os.path.join(path, f".pm-probe-{uuid.uuid4().hex[:8]}.tmp")
+    try:
+        with open(probe, "wb") as handle:
+            handle.write(b"")
+    except OSError:
+        return False
+    try:
+        os.unlink(probe)
+    except OSError:
+        return False
+    return True
+
+
+def location_problem(path, *, purpose, context, timeout=VALIDATION_TIMEOUT):
+    """Why *path* cannot be the thumbnail folder, as a code, or None.
+
+    *purpose*: ``move`` (every rule), ``startup`` (no ``same``, ``empty``,
+    ``writable`` or parent rules: availability covers them) or ``recreate``
+    (the folder is missing and is made again).
+    """
+    flavor = context.flavor
+    code = form_problem(path, flavor=flavor)
+    if code is not None:
+        return code
+    path = flavor.abspath(path)
+    default = bool(context.default_path) and flavor.normcase(path) == flavor.normcase(
+        flavor.abspath(context.default_path)
+    )
+    if is_too_broad(path, home=context.home, flavor=flavor):
+        return "location-too-broad"
+    try:
+        return _timed(
+            lambda: _filesystem_problem(path, purpose, context, default), timeout
+        )
+    except (TimeoutError, OSError):
+        return "location-unavailable"
+
+
+def _filesystem_problem(path, purpose, context, default):
+    if _served(path, context):
+        return "location-served"
+    if context.node_folder and not default and _inside(path, [context.node_folder]):
+        return "location-node-folder"
+    inner = list(context.roots) + list(context.contained)
+    if any(folder and _inside(folder, [path]) for folder in inner):
+        if not default:
+            return "location-contains-source"
+    if (
+        purpose == "move"
+        and context.current_path
+        and (
+            os.path.normcase(path) == os.path.normcase(context.current_path)
+            or (os.path.isdir(path) and _same(path, context.current_path))
+        )
+    ):
+        return "location-same"
+    # Windows, or a test injecting its limit (``long_paths=False``).
+    if os.name == "nt" or context.long_paths is False:
+        if too_long_for_windows(path, context.long_paths):
+            return "location-path-too-long"
+    if purpose == "startup":
+        return None
+    if is_link_like(path):
+        # Never through a link: the folder itself must be a real folder.
+        return "location-not-empty"
+    exists = os.path.isdir(path)
+    if not exists:
+        if os.path.lexists(path):
+            return "location-not-empty"
+        if not default and not os.path.isdir(os.path.dirname(path)):
+            return "location-parent-missing"
+        return None
+    if read_marker(path) is None and not holds_only_metadata(path):
+        return "location-not-empty"
+    if not _probe_write(path):
+        return "location-not-writable"
+    return None

py/api/preview_store.py

@@ -23,20 +23,31 @@ no lock (the job registry's ``running()`` is a lock-free snapshot) and
 All functions are blocking: call them off the event loop.
 """
 
+import errno
+import logging
 import os
 import threading
+import time
 from contextlib import contextmanager
-from dataclasses import dataclass
+from dataclasses import dataclass, replace
 from pathlib import Path
 
-from . import thumbnail_settings
+from . import preview_location, thumbnail_settings
 
 try:
     from ...utils.media_folders import PREVIEW_FOLDER
 except ImportError:  # tests import the package top-level
     from utils.media_folders import PREVIEW_FOLDER
 
-SOURCE_MODE = "source"
+logger = logging.getLogger("prompt_manager.thumbnails")
+
+SOURCE_MODE = preview_location.SOURCE_MODE
+FOLDER_MODE = preview_location.FOLDER_MODE
+# The probe checks the folder this often, and a check (or a store call)
+# taking longer than the budget marks it unreachable.
+PROBE_INTERVAL = 5.0
+IO_BUDGET = 2.0
+PROBE_THREAD = "pm-preview-probe"
 
 
 @dataclass(frozen=True)
@@ -52,11 +63,34 @@ class Location:
 
 write_lock = threading.Lock()
 _generation = 0
+# The runtime location: the parsed setting, availability and the marker keys
+# last read. Swapped as a whole; readers take one snapshot.
+_state_lock = threading.Lock()
+_parsed = preview_location.ParsedLocation()
+_available = True
+_problem = None
+_keys = {}
+_key_cache = {}
+_clock = time.monotonic
 
 
 def current():
     """The thumbnail location now."""
-    return Location(generation=_generation)
+    with _state_lock:
+        parsed, available, problem, generation_now = (
+            _parsed,
+            _available,
+            _problem,
+            _generation,
+        )
+    if parsed.mode == SOURCE_MODE:
+        return Location(generation=generation_now)
+    return Location(FOLDER_MODE, parsed.path, available, problem, generation_now)
+
+
+def setting():
+    """The parsed setting in use (``preview_location.ParsedLocation``)."""
+    return _parsed
 
 
 def generation():
@@ -80,23 +114,425 @@ def _within(path, folder):
     return _canonical(path).is_relative_to(_canonical(folder))
 
 
+# ── the setting and availability ─────────────────────────────────────
+
+
+def configure(value, *, context=None, check=True):
+    """Use the ``thumbnails.location`` *value* (config.json) from now on.
+
+    A folder location is validated against *context* (``startup`` rules: a
+    problem other than an unreachable folder makes it ``invalid``), then
+    checked once; the probe keeps checking it. Swapped under ``write_lock``
+    with a generation bump, so no write that started before lands.
+    """
+    global _parsed, _available, _problem, _generation
+    parsed = preview_location.parse_location(value)
+    problem = parsed.problem
+    if parsed.mode == FOLDER_MODE and problem is None and context is not None:
+        code = preview_location.location_problem(
+            parsed.path, purpose="startup", context=context
+        )
+        if code == "location-unavailable":
+            problem = "unreachable"
+        elif code is not None:
+            problem = "invalid"
+            parsed = preview_location.ParsedLocation(
+                mode=FOLDER_MODE,
+                path=parsed.path,
+                id=parsed.id,
+                install=parsed.install,
+                cleanup=parsed.cleanup,
+                problem="invalid",
+            )
+    with write_lock:
+        with _state_lock:
+            _parsed = parsed
+            _available = parsed.mode == SOURCE_MODE
+            _problem = problem if parsed.mode == FOLDER_MODE else None
+            _keys.clear()
+            _key_cache.clear()
+        _generation += 1
+    if parsed.mode == FOLDER_MODE:
+        if check and parsed.problem is None:
+            refresh()
+        _probe.start()
+    else:
+        _probe.stop()
+    return current()
+
+
+def refresh():
+    """Check the folder now (bounded by IO_BUDGET); returns availability."""
+    parsed = _parsed
+    if parsed.mode != FOLDER_MODE or parsed.problem is not None:
+        return parsed.mode == SOURCE_MODE
+    try:
+        available, problem, marker = _probe.check(parsed.path, parsed.id)
+    except (OSError, TimeoutError, RuntimeError):
+        available, problem, marker = False, "unreachable", None
+    _set_state(parsed, available, problem, marker)
+    return available
+
+
+def _set_keys(parsed, marker):
+    """Take the keys of a marker just written (availability unchanged)."""
+    with _state_lock:
+        if _parsed is not parsed:
+            return
+        keys = dict((marker or {}).get("keys") or {})
+        if keys != _keys:
+            _keys.clear()
+            _keys.update(keys)
+            _key_cache.clear()
+
+
+def mark_invalid():
+    """The setting cannot be used (a startup check that failed): folder
+    mode stays in force, unavailable, so nothing is written anywhere."""
+    global _parsed, _available, _problem
+    with _state_lock:
+        if _parsed.mode != FOLDER_MODE:
+            _parsed = preview_location.ParsedLocation(
+                mode=FOLDER_MODE, problem="invalid"
+            )
+        else:
+            _parsed = replace(_parsed, problem="invalid")
+        _available, _problem = False, "invalid"
+
+
+def _set_state(parsed, available, problem, marker):
+    global _available, _problem
+    with _state_lock:
+        if _parsed is not parsed:
+            return  # the setting changed meanwhile
+        _available, _problem = available, problem
+        keys = dict((marker or {}).get("keys") or {})
+        if keys != _keys:
+            _keys.clear()
+            _keys.update(keys)
+            _key_cache.clear()
+
+
+def note_failure():
+    """A store call failed or stalled: unreachable until the probe finds the
+    folder again (requests do no store I/O meanwhile)."""
+    global _available, _problem
+    if _parsed.mode != FOLDER_MODE:
+        return
+    with _state_lock:
+        _available, _problem = False, "unreachable"
+    _probe.wake()
+
+
+@contextmanager
+def store_io():
+    """Around a store call in folder mode: an I/O error (other than a file
+    that is not there) or a call over IO_BUDGET marks the folder
+    unreachable."""
+    if _parsed.mode != FOLDER_MODE:
+        yield
+        return
+    started = _clock()
+    try:
+        yield
+    except OSError as error:
+        if not isinstance(error, (FileNotFoundError, NotADirectoryError)):
+            note_failure()
+        raise
+    finally:
+        if _clock() - started > IO_BUDGET:
+            note_failure()
+
+
+class _Probe:
+    """The background check of the thumbnail folder (one thread, one check
+    at a time; a stuck check is left to finish on its own thread)."""
+
+    def __init__(self):
+        self._lock = threading.Lock()
+        self._wake = threading.Event()
+        self._stop = threading.Event()
+        self._thread = None
+        self._checking = None
+
+    def check(self, path, expected_id):
+        """``folder_state`` within IO_BUDGET; raises TimeoutError. A check
+        already running (the probe's, or a stuck one) is joined, never run
+        twice at once."""
+        with self._lock:
+            pending = self._checking
+            if pending is None or pending[0] != (path, expected_id):
+                pending = self._start_check(path, expected_id)
+        key, done, result = pending
+        if not done.wait(IO_BUDGET):
+            raise TimeoutError("the thumbnail folder did not answer in time")
+        with self._lock:
+            if self._checking is pending:
+                self._checking = None
+        if "error" in result:
+            raise result["error"]
+        return result["value"]
+
+    def _start_check(self, path, expected_id):
+        """Start one check on its own thread (lock held)."""
+        result = {}
+        done = threading.Event()
+
+        def run():
+            try:
+                result["value"] = preview_location.folder_state(path, expected_id)
+            except BaseException as error:  # handed to the caller
+                result["error"] = error
+            finally:
+                done.set()
+
+        pending = ((path, expected_id), done, result)
+        self._checking = pending
+        threading.Thread(target=run, name=PROBE_THREAD + "-check", daemon=True).start()
+        return pending
+
+    def start(self):
+        with self._lock:
+            if self._thread is not None and self._thread.is_alive():
+                return
+            # Each thread has its own stop: a stopped one never resumes.
+            stop = self._stop = threading.Event()
+            thread = threading.Thread(
+                target=self._run, args=(stop,), name=PROBE_THREAD, daemon=True
+            )
+            try:
+                thread.start()
+            except RuntimeError:
+                return  # refused at interpreter exit: no probe
+            self._thread = thread
+
+    def _run(self, stop):
+        while not stop.is_set():
+            self._wake.wait(PROBE_INTERVAL)
+            self._wake.clear()
+            if stop.is_set():
+                return
+            try:
+                refresh()
+            except Exception:  # one failed check never stops the probe
+                logger.debug("Thumbnail folder check failed", exc_info=True)
+
+    def wake(self):
+        self._wake.set()
+
+    def stop(self, timeout=2.0):
+        """Stop the probe; True when its thread ended."""
+        with self._lock:
+            thread, self._thread = self._thread, None
+        self._stop.set()
+        self._wake.set()
+        if thread is None or thread is threading.current_thread():
+            return True
+        thread.join(timeout)
+        return not thread.is_alive()
+
+
+_probe = _Probe()
+
+
+def stop_probe(timeout=2.0):
+    """Stop the folder probe (the package teardown, tests)."""
+    return _probe.stop(timeout)
+
+
 # ── stores ───────────────────────────────────────────────────────────
 
 
-def store_container(root, location=None):
-    """The folder a store of *root* must stay inside: the source itself."""
-    return Path(root)
-
-
-def store_dir(root, location=None):
-    """The store of *root* (``<source>/thumbnails``)."""
+def source_folder(root):
+    """``<source>/thumbnails``: the store in source mode, a leftover in
+    folder mode."""
     return Path(root) / PREVIEW_FOLDER
 
 
-def ensure_store(root, location=None):
-    """``store_dir`` for a writer. Preview folders below it are made by the
-    write itself."""
-    return store_dir(root, location)
+def store_container(root, location=None):
+    """The folder a store of *root* must stay inside: the source itself, or
+    the thumbnail folder in folder mode."""
+    location = location or current()
+    if location.mode == FOLDER_MODE:
+        return Path(location.path) if location.path else None
+    return Path(root)
+
+
+def _identity(root):
+    return os.path.normcase(os.path.realpath(str(root)))
+
+
+def _lookup_key(root):
+    spelling = str(root)
+    with _state_lock:
+        if spelling in _key_cache:
+            return _key_cache[spelling]
+        keys = dict(_keys)
+    key = preview_location.find_key(
+        keys,
+        spelling,
+        _identity(root),
+        preview_location.file_identity(root),
+    )
+    if key is not None:
+        with _state_lock:
+            _key_cache[spelling] = key
+    return key
+
+
+def store_dir(root, location=None):
+    """The store of *root*: ``<source>/thumbnails``, or in folder mode
+    ``<folder>/<key>`` (None while the folder is unavailable or before the
+    source has a key: a reader then finds no preview)."""
+    location = location or current()
+    if location.mode == SOURCE_MODE:
+        return Path(root) / PREVIEW_FOLDER
+    if not location.available or not location.path:
+        return None
+    key = _lookup_key(root)
+    return None if key is None else _key_folder(location.path, key)
+
+
+def _key_folder(folder, key):
+    """``<folder>/<key>``, only for a key that is one folder name directly
+    below the thumbnail folder (a marker is data, never a path)."""
+    if not preview_location.valid_key(key):
+        return None
+    store = Path(folder) / key
+    if store.parent != Path(folder) or store.name != key:
+        return None
+    return store
+
+
+def ensure_store(root, location=None, *, create=False):
+    """``store_dir`` for a writer: in folder mode a source without a key gets
+    one (recorded in the marker); with *create* its key folder is made (never
+    the thumbnail folder itself). None while the folder is unavailable."""
+    location = location or current()
+    store = store_dir(root, location)
+    if location.mode == SOURCE_MODE or (store is None and not location.available):
+        return store
+    if store is None:
+        store = _assign_key(root, location)
+        if store is None:
+            return None
+    if create:
+        try:
+            with store_io():
+                os.mkdir(store)
+        except FileExistsError:
+            pass
+        except OSError:
+            return None
+    return store
+
+
+def _assign_key(root, location):
+    parsed = _parsed
+    if location.path != parsed.path:
+        return None
+    spelling = str(root)
+    identity = _identity(root)
+    dev_ino = preview_location.file_identity(root)
+    assigned = {}
+
+    def add(marker):
+        if marker.get("id") != parsed.id:
+            raise ValueError("another folder's marker")  # marker-mismatch
+        keys = marker["keys"]
+        key = preview_location.find_key(keys, spelling, identity, dev_ino)
+        if key is None:
+            taken = {k: v.get("identity") for k, v in keys.items()}
+            key = preview_location.source_key(spelling, identity, taken)
+            keys[key] = {
+                "path": spelling,
+                "identity": identity,
+                "dev_ino": dev_ino,
+                "install": parsed.install,
+            }
+        assigned["key"] = key
+
+    try:
+        with write_lock, store_io():
+            marker = preview_location.write_marker(location.path, add)
+    except (OSError, ValueError):
+        return None
+    _set_keys(parsed, marker)
+    return _key_folder(location.path, assigned["key"])
+
+
+def may_write(path, location=None):
+    """The one ownership check every preview write and delete passes.
+
+    Source mode: True (the callers' own root rules apply, as before). Folder
+    mode: *path* lies below a source key folder of the thumbnail folder,
+    lexically and resolved, and no link, junction or reparse point sits
+    anywhere from the key folder down to it (the file itself included).
+    """
+    location = location or current()
+    if location.mode == SOURCE_MODE:
+        return True
+    if not location.path:
+        return False
+    folder = Path(os.path.abspath(location.path))
+    path = Path(os.path.abspath(path))
+    try:
+        relative = path.relative_to(folder)
+    except ValueError:
+        return False
+    if len(relative.parts) < 2 or ".." in relative.parts:
+        return False
+    store = _key_folder(folder, relative.parts[0])
+    if store is None or not _within(path, folder):
+        return False
+    try:
+        return _no_link_below(store, path)
+    except OSError:
+        note_failure()
+        return False
+
+
+def make_parent(dst, location=None):
+    """Make the folders a preview *dst* is written into. In folder mode only
+    folders below the thumbnail folder are made, never the folder itself,
+    and only while it holds this install's marker (an unplugged drive or an
+    unmounted mount point raises FileNotFoundError); a *dst* that does not
+    pass ``may_write`` raises PermissionError."""
+    location = location or current()
+    dst = Path(dst)
+    if location.mode == SOURCE_MODE or not location.path:
+        if location.mode != SOURCE_MODE:
+            raise FileNotFoundError(errno.ENOENT, "thumbnail folder unavailable")
+        dst.parent.mkdir(parents=True, exist_ok=True)
+        return
+    folder = Path(os.path.abspath(location.path))
+    with store_io():
+        if not location.available or not folder.is_dir():
+            raise FileNotFoundError(errno.ENOENT, "thumbnail folder unavailable")
+        marker = preview_location.read_marker(str(folder))
+        if marker is None or marker.get("id") != _parsed.id:
+            note_failure()
+            raise FileNotFoundError(errno.ENOENT, "thumbnail folder unavailable")
+        if not may_write(dst, location):
+            raise PermissionError(errno.EACCES, "outside the thumbnail folder")
+        current_folder = folder
+        for part in Path(os.path.abspath(dst.parent)).relative_to(folder).parts:
+            current_folder = current_folder / part
+            try:
+                os.mkdir(current_folder)
+            except FileExistsError:
+                pass
+            if preview_location.is_link_like(current_folder):
+                raise PermissionError(errno.EACCES, "a link in the thumbnail folder")
+
+
+def store_is_link(store, location=None):
+    """Whether a store folder is a link (a junction or reparse point too in
+    folder mode): it is skipped by sweeps and refused by clears."""
+    location = location or current()
+    if location.mode == SOURCE_MODE:
+        return Path(store).is_symlink()
+    return preview_location.is_link_like(store)
 
 
 def given_store(root, store):
@@ -109,10 +545,14 @@ def given_store(root, store):
     return store
 
 
-def preview_path(root, media_rel, ext=None, location=None):
+def preview_path(root, media_rel, ext=None, location=None, *, create=False):
     """Where the preview of *media_rel* (relative to *root*) is, or is to be
-    written, in *ext* (the extension written now when None)."""
-    store = store_dir(root, location)
+    written (*create*: a writer, which gives a new source its key), in *ext*
+    (the extension written now when None)."""
+    if create:
+        store = ensure_store(root, location)
+    else:
+        store = store_dir(root, location)
     if store is None:
         return None
     if ext is None:
@@ -124,8 +564,8 @@ def preview_path(root, media_rel, ext=None, location=None):
 
 def request_path(root, parts, location=None):
     """The disk path of the URL parts ``("thumbnails", ...)``, or None when
-    *parts* is not under an exact ``thumbnails`` first part. The text is
-    checked before (``is_preview_text``)."""
+    *parts* is not under an exact ``thumbnails`` first part or the store has
+    no place for it. The text is checked before (``is_preview_text``)."""
     parts = tuple(parts)
     if len(parts) < 2 or parts[0] != PREVIEW_FOLDER:
         return None
@@ -135,13 +575,38 @@ def request_path(root, parts, location=None):
     return store.joinpath(*parts[1:])
 
 
+def _no_link_below(store, path):
+    """No folder from *store* down to *path* (and *store* itself) is a link,
+    junction or reparse point (lexical walk, one lstat each)."""
+    store = Path(os.path.abspath(store))
+    path = Path(os.path.abspath(path))
+    try:
+        relative = path.relative_to(store)
+    except ValueError:
+        return False
+    current = store
+    for part in ("",) + relative.parts:
+        current = current / part if part else current
+        if preview_location.is_link_like(current):
+            return False
+    return True
+
+
 def owns(root, path, location=None):
     """Whether *path* lies inside the store of *root*, resolved: the store
-    stays inside its container and no link takes *path* out of the store."""
+    stays inside its container and no link takes *path* out of the store.
+    In folder mode no link, junction or reparse point may sit anywhere from
+    the store down to *path*."""
+    location = location or current()
     store = store_dir(root, location)
     if store is None:
         return False
-    return _within(store, store_container(root, location)) and _within(path, store)
+    container = store_container(root, location)
+    if not (_within(store, container) and _within(path, store)):
+        return False
+    if location.mode == SOURCE_MODE:
+        return True
+    return may_write(path, location)
 
 
 # ── preview text ─────────────────────────────────────────────────────

py/api/settings_routes.py

@@ -30,6 +30,7 @@ from . import thumbnail_settings
 from .admin import _current_image_monitor
 from .job_routes import busy_response
 from .jobs import THUMBNAILS_JOB_KIND, JobBusy
+from .thumbnail_location import needs_move_response
 
 try:
     from ...utils.validators import validate_result_timeout, validate_worker_threads
@@ -264,6 +265,12 @@ class SettingsRoutesMixin:
                     "sources": source_status,
                     "monitor": monitor,
                     "thumbnail_encoders": encoders,
+                    # #320: the full path only here (D6A).
+                    "thumbnail_location": await self._run_in_executor(
+                        functools.partial(
+                            self._thumbnail_location_payload, with_path=True
+                        )
+                    ),
                     "settings": {
                         "monitoring_enabled": GalleryConfig.MONITORING_ENABLED,
                         "save_foreign_runs": GalleryConfig.SAVE_FOREIGN_RUNS,
@@ -377,6 +384,9 @@ class SettingsRoutesMixin:
             if error_response is not None:
                 return error_response
 
+            if "thumbnail_location" in data:
+                # #320: only the Move thumbnails job changes the location.
+                return needs_move_response()
             expected_revision = data.get("expected_revision")
             if expected_revision is not None and not isinstance(expected_revision, str):
                 return web.json_response(

py/api/sources.py

@@ -30,6 +30,8 @@ import ntpath
 import os
 import re
 
+from .preview_location import HOLDS_THUMBNAILS_REASON
+
 logger = logging.getLogger("prompt_manager.api")
 
 MAX_SOURCE_PATH = 4096  # characters
@@ -114,7 +116,26 @@ def source_problem(text):
     code = source_form_error(text)
     if code is not None:
         return code, FORM_REASONS[code]
-    return GalleryConfig.gallery_root_problem(text)
+    code, reason = GalleryConfig.gallery_root_problem(text)
+    if code is None and _holds_thumbnails(GalleryConfig.resolve_gallery_root(text)):
+        return "holds-thumbnails", HOLDS_THUMBNAILS_REASON
+    return code, reason
+
+
+def _holds_thumbnails(path):
+    """Whether *path* is inside or around the thumbnail folder in use (#320):
+    a source there would list Prompt Manager's own previews."""
+    from . import preview_store
+
+    try:
+        from ...utils.library_location import _inside_any
+    except ImportError:  # tests import the package top-level
+        from utils.library_location import _inside_any
+
+    setting = preview_store.setting()
+    if setting.mode != preview_store.FOLDER_MODE or not setting.path:
+        return False
+    return _inside_any(path, [setting.path]) or _inside_any(setting.path, [path])
 
 
 def public_spelling(text):

py/api/thumbnail_auto.py

@@ -175,7 +175,7 @@ class DeferredPreview:
 
     def resolve(self):
         """The preview's path now, or None when nothing may be written."""
-        dst = media_preview_path(self.root, self.rel)
+        dst = media_preview_path(self.root, self.rel, create=True)
         if dst is None:
             return None
         if self.strict:
@@ -298,8 +298,12 @@ def _thumbnail_source(root, parts):
     if in_trash_below_roots(found, [root]):
         return None
     rel = Path(*folder_parts, found.name)
-    dst = media_preview_path(root, rel)
-    if dst is None or dst != _current_preview(root, parts):
+    dst = media_preview_path(root, rel, create=True)
+    if dst is None:
+        # Nowhere to write (the thumbnail folder is unavailable): the
+        # original is served and nothing is written.
+        return found, None, found.suffix.lower() in VIDEO_EXTENSIONS
+    if dst != _current_preview(root, parts):
         return None
     if not preview_store.owns(root, dst):
         dst = None
@@ -745,6 +749,8 @@ class AutoThumbnailMixin:
             pool.shutdown(wait=False, cancel_futures=True)
         if final:
             shutdown_encoders()  # kill JPEG XL / ffmpeg encoders still running
+            # The thumbnail folder's probe (#320), bounded like the rest.
+            closed = preview_store.stop_probe() and closed
         return closed
 
     def _auto_thumbnail_failures(self):

py/api/thumbnail_jobs.py

@@ -49,6 +49,7 @@ from .images import (
     _worker_threads,
 )
 from .job_routes import busy_response
+from .thumbnail_location import location_refusal
 from .jobs import THUMBNAIL_GROUP_NAMES, THUMBNAILS_JOB_KIND, JobBusy
 from .thumbnail_convert import CONVERT_JOB_TITLE
 from .thumbnail_encode import writer_settings
@@ -182,6 +183,9 @@ class ThumbnailJobsMixin:
         options, error = thumbnail_options(body)
         if error:
             return _invalid(error)
+        refused = location_refusal()
+        if refused is not None:
+            return refused
         try:
             job = self._start_thumbnail_job(options)
         except JobBusy as busy:
@@ -459,8 +463,12 @@ class ThumbnailJobsMixin:
         or attaches to the running thumbnail job, and streams its frames. A
         legacy hold (the blocking generate or clear) or a record-changing
         job refuses it with the stream's error frame. The job goes on when
-        the client goes away.
+        the client goes away. Refused (409) while the thumbnail folder is
+        unavailable (#320).
         """
+        refused = location_refusal()
+        if refused is not None:
+            return refused
         response = web.StreamResponse(
             status=200,
             headers={

py/api/thumbnail_location.py

@@ -0,0 +1,160 @@
+"""The thumbnail location on the API (#320): startup, the settings payload
+and the refusals while the thumbnail folder is unavailable.
+
+The setting itself is config.json ``thumbnails.location``
+(``preview_location``); the runtime state is ``preview_store``. POST
+``/settings`` never changes it (409 ``location-needs-move``): the Move
+thumbnails job is its only writer.
+"""
+
+import os
+
+from aiohttp import web
+
+from . import preview_location, preview_store
+
+try:
+    from ...utils import library_location
+except ImportError:  # tests import the package top-level
+    from utils import library_location
+
+# The node folder (its web/ is served and a reinstall wipes it).
+NODE_FOLDER = os.path.dirname(
+    os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+)
+
+
+def default_path():
+    """``<data folder>/data/thumbnails`` (D7B), or None when unknown."""
+    try:
+        return library_location.cache_dir("thumbnails")
+    except Exception:  # no data folder can be decided: no default
+        return None
+
+
+def location_refusal(location=None):
+    """The 409 ``location-unavailable`` response while the thumbnail folder
+    is unavailable, else None (thumbnail generate, convert, clear, prune)."""
+    location = location or preview_store.current()
+    if location.mode != preview_store.FOLDER_MODE or location.available:
+        return None
+    status, text = preview_location.message("location-unavailable")
+    return web.json_response(
+        {"success": False, "code": "location-unavailable", "error": text},
+        status=status,
+    )
+
+
+def needs_move_response():
+    status, text = preview_location.message("location-needs-move")
+    return web.json_response(
+        {"success": False, "code": "location-needs-move", "error": text},
+        status=status,
+    )
+
+
+class ThumbnailLocationMixin:
+    """Startup check and payloads of the thumbnail location."""
+
+    def _all_source_roots(self):
+        """Every image source of every root list: the Images roots, the
+        configured sources and the extra gallery roots (as strings)."""
+        from ..config import GalleryConfig, _extra_gallery_roots
+
+        roots = [str(root) for root in self._get_all_output_dirs()]
+        roots += [str(root) for root in GalleryConfig.MONITORING_DIRECTORIES]
+        roots += list(_extra_gallery_roots())
+        return tuple(dict.fromkeys(root for root in roots if root))
+
+    def _thumbnail_location_context(self):
+        """What a thumbnail folder is validated against (blocking)."""
+        contained = []
+        try:
+            import folder_paths
+
+            for name in (
+                "get_output_directory",
+                "get_input_directory",
+                "get_temp_directory",
+            ):
+                getter = getattr(folder_paths, name, None)
+                folder = getter() if getter is not None else None
+                if folder:
+                    contained.append(str(folder))
+        except ImportError:
+            pass
+        try:
+            contained.append(library_location.config_location()[1])
+        except Exception:  # no data folder can be decided
+            self.logger.debug("No library folder for the thumbnail check")
+        setting = preview_store.setting()
+        return preview_location.Context(
+            roots=self._all_source_roots(),
+            contained=tuple(contained),
+            node_folder=NODE_FOLDER,
+            default_path=default_path(),
+            current_path=(
+                setting.path if setting.mode == preview_store.FOLDER_MODE else None
+            ),
+        )
+
+    def use_preview_location(self):
+        """Put the configured location in force at once, unchecked (the
+        package's import, before any request): folder mode is unavailable
+        until ``start_preview_location`` has checked it, so nothing is ever
+        written to the sources meanwhile. No filesystem access."""
+        from ..config import PromptManagerConfig
+
+        try:
+            return preview_store.configure(
+                PromptManagerConfig.THUMBNAIL_LOCATION, check=False
+            )
+        except Exception:
+            self.logger.warning("The thumbnail location is unusable", exc_info=True)
+            preview_store.mark_invalid()
+            return preview_store.current()
+
+    def start_preview_location(self):
+        """Parse and check the configured location once (startup thread);
+        the probe takes over. Never raises; a check that fails leaves a
+        folder location unavailable (``invalid``), never the sources."""
+        from ..config import PromptManagerConfig
+
+        try:
+            value = PromptManagerConfig.THUMBNAIL_LOCATION
+            parsed = preview_location.parse_location(value)
+            context = None
+            if parsed.mode == preview_store.FOLDER_MODE:
+                context = self._thumbnail_location_context()
+            location = preview_store.configure(value, context=context)
+        except Exception:
+            self.logger.warning("The thumbnail location check failed", exc_info=True)
+            if (
+                preview_location.parse_location(
+                    PromptManagerConfig.THUMBNAIL_LOCATION
+                ).mode
+                == preview_store.FOLDER_MODE
+            ):
+                preview_store.mark_invalid()
+            return preview_store.current()
+        if location.mode == preview_store.FOLDER_MODE and not location.available:
+            self.logger.warning(
+                "The thumbnail folder is unavailable "
+                f"({location.problem or 'checking'}): originals are shown and "
+                "no thumbnails are written until it is back"
+            )
+        return location
+
+    def _thumbnail_location_payload(self, *, with_path):
+        """``{mode, available, problem}`` plus, for the settings GET only,
+        ``path`` and ``default_path`` (D6A)."""
+        location = preview_store.current()
+        payload = {
+            "mode": location.mode,
+            "available": location.available,
+            "problem": location.problem,
+        }
+        if with_path:
+            payload["path"] = location.path
+            payload["default_path"] = default_path()
+        return payload

py/api/thumbnail_settings.py

@@ -221,7 +221,13 @@ def _owned(root, rel_path, preferred):
     if store is None:
         return iter(())
     stem = store / preview_stem(rel_path)
-    return owned_in_folder(stem.parent, stem.name, rel_path.suffix, preferred)
+    found = owned_in_folder(stem.parent, stem.name, rel_path.suffix, preferred)
+    if preview_store.current().mode == preview_store.SOURCE_MODE:
+        return found
+    # Folder mode: timed, so a stalled share marks the folder unreachable,
+    # and only previews no link takes out of the key folder (deletes use it).
+    with preview_store.store_io():
+        return iter([path for path in found if preview_store.may_write(path)])
 
 
 def owned_thumbnails(root, rel_path, preferred=None):

py/config.py

@@ -767,6 +767,8 @@ class PromptManagerConfig:
     THUMBNAIL_QUALITY = 80
     THUMBNAIL_EDGE = 512
     THUMBNAIL_EDGE_RANGE = (128, 2048)
+    # #320 thumbnail location (config.json ``thumbnails.location``), as read.
+    THUMBNAIL_LOCATION = None
 
     @classmethod
     def max_worker_threads(cls) -> int:
@@ -828,6 +830,12 @@ class PromptManagerConfig:
                 "format": cls.THUMBNAIL_FORMAT,
                 "quality": cls.THUMBNAIL_QUALITY,
                 "edge": cls.THUMBNAIL_EDGE,
+                # #320: as read; only the Move thumbnails job writes it.
+                **(
+                    {"location": cls.THUMBNAIL_LOCATION}
+                    if cls.THUMBNAIL_LOCATION is not None
+                    else {}
+                ),
             },
             "gallery": GalleryConfig.get_config(),
             "integrations": IntegrationConfig.get_config(),
@@ -1002,6 +1010,11 @@ class PromptManagerConfig:
         low, high = cls.THUMBNAIL_EDGE_RANGE
         if type(edge) is int and low <= edge <= high:
             cls.THUMBNAIL_EDGE = edge
+        # #320: kept as read (``preview_location.parse_location`` checks it;
+        # a bad value becomes an unavailable ``invalid`` location, never a
+        # silent fall-back to the sources).
+        if "location" in section:
+            cls.THUMBNAIL_LOCATION = section["location"]
 
 
 # Load configuration on import

tests/browser/server.py

@@ -22,6 +22,9 @@ from PIL.PngImagePlugin import PngInfo
 from tests.browser.folder_trash import close_trash, replace_folder
 from tests.browser.malloc_tuning import pin_mmap_threshold
 from tests.browser.request_hold import register_request_holds
+from tests.browser.thumbnail_location_fixture import (
+    register_thumbnail_location_commands,
+)
 from tests.browser import slow_reset_watch
 from tests.browser.fast_reset import SEED_ON_LIVE_ENV, ScratchSeeder
 from tests.browser.workspace_activity_fixture import register_activity_commands
@@ -240,6 +243,7 @@ def create_app(root: Path) -> web.Application:
     library_editor_commands = register_library_editor_commands(routes, api, root)
     register_vocabulary_commands(routes, api, root)
     restore_sources = register_source_change_commands(routes, api)
+    reset_thumbnail_location = register_thumbnail_location_commands(routes, api, root)
     register_tagging_commands(routes, api, root, model, on_reset=restore_sources)
     register_review_commands(routes, api, root)
     register_collections_commands(routes, api, root)
@@ -451,6 +455,8 @@ def create_app(root: Path) -> web.Application:
         restore_sources()
         # No preview queued by the last test may land in the new fixture.
         await api._run_in_executor(api.close_auto_thumbnails)
+        # Previews inside each source again (#320).
+        await api._run_in_executor(reset_thumbnail_location)
         library_editor_commands.reset()
         await api._run_in_executor(reset_logs)
         data = await request.json() if request.can_read_body else {}
@@ -595,6 +601,7 @@ def create_app(root: Path) -> web.Application:
         history_commands.reset()
         library_editor_commands.reset()
         reveal_stub.reset()
+        await api._run_in_executor(reset_thumbnail_location)
         history_fixture = None
         manifest = await api._run_in_executor(
             seeder.run,

tests/browser/thumbnail_location_fixture.py

@@ -0,0 +1,92 @@
+"""The thumbnail location on the test server (#320), undone by every reset.
+
+Folder mode is reachable from config.json and tests only until the Move
+thumbnails job lands, so page tests set it here, inside the server's root:
+
+- ``POST /__test__/thumbnail-location`` ``{"mode": "folder"}`` (or
+  ``"source"``): previews live in ``<root>/thumbs/<key>`` from now on (a new
+  folder with Prompt Manager's marker).
+- ``GET /__test__/thumbnail-files``: the preview files in the thumbnail
+  folder and whether any source grew a ``thumbnails/`` folder.
+- ``POST /__test__/thumbnail-folder/remove``: the thumbnail folder goes, as
+  when its drive is unplugged; the server notices it on its next check.
+"""
+
+from __future__ import annotations
+
+import shutil
+from pathlib import Path
+from typing import TYPE_CHECKING, Callable
+
+from aiohttp import web
+
+if TYPE_CHECKING:
+    from py.api import PromptManagerAPI
+
+FOLDER_NAME = "thumbs"
+
+
+def register_thumbnail_location_commands(
+    routes: web.RouteTableDef, api: PromptManagerAPI, root: Path
+) -> Callable[[], None]:
+    """Register the commands; returns the reset every reset runs."""
+    from py.api import preview_location, preview_store
+
+    folder = root / FOLDER_NAME
+
+    def reset() -> None:
+        preview_store.configure(None)
+        shutil.rmtree(folder, ignore_errors=True)
+
+    def use(mode: str) -> dict:
+        reset()
+        if mode == "source":
+            return {"mode": "source"}
+        folder.mkdir()
+        marker_id = preview_location.init_folder(str(folder), created_by_pm=True)
+        location = preview_store.configure(
+            {
+                "mode": "folder",
+                "path": str(folder),
+                "id": marker_id,
+                "install": "browser-tests",
+            }
+        )
+        return {"mode": location.mode, "available": location.available}
+
+    @routes.post("/__test__/thumbnail-location")
+    async def set_location(request: web.Request) -> web.Response:
+        data = await request.json()
+        mode = data.get("mode") if isinstance(data, dict) else None
+        if mode not in ("source", "folder"):
+            raise web.HTTPBadRequest()
+        return web.json_response(await api._run_in_executor(use, mode))
+
+    def files() -> dict:
+        previews = sorted(
+            path.relative_to(folder).as_posix()
+            for path in folder.rglob("*_thumb.*")
+            if path.is_file()
+        )
+        sources = [str(path) for path in api._get_all_output_dirs()]
+        return {
+            "folder": previews,
+            "source_thumbnails": any(
+                (Path(source) / "thumbnails").exists() for source in sources
+            ),
+        }
+
+    @routes.get("/__test__/thumbnail-files")
+    async def thumbnail_files(_request: web.Request) -> web.Response:
+        return web.json_response(await api._run_in_executor(files))
+
+    def remove() -> bool:
+        shutil.rmtree(folder, ignore_errors=True)
+        return preview_store.refresh()
+
+    @routes.post("/__test__/thumbnail-folder/remove")
+    async def remove_folder(_request: web.Request) -> web.Response:
+        available = await api._run_in_executor(remove)
+        return web.json_response({"available": available})
+
+    return reset

tests/browser/workflows.spec.mjs

@@ -261,3 +261,20 @@ test('prompt deletion requires confirmation and can retry after a network failur
         await expect(await request.get(url)).toBeOK();
     }
 });
+
+// #320: with previews in one folder, the legacy gallery still shows them, made
+// and served from that folder (the fixture's own previews inside the sources
+// are never served then).
+test('the legacy gallery shows thumbnails kept in one folder', async ({ page, request }) => {
+    await expect(await request.post('/__test__/thumbnail-location', { data: { mode: 'folder' } })).toBeOK();
+    const served = page.waitForResponse(response =>
+        response.url().includes('/prompt_manager/images/serve/thumbnails/') && response.status() === 200);
+    await page.goto('/prompt_manager/gallery');
+    await expect(page.locator('#galleryGrid .image-item')).toHaveCount(2);
+    const response = await served; // made in the folder before it was served
+    const name = decodeURIComponent(new URL(response.url()).pathname.split('/serve/thumbnails/')[1]);
+    const files = await (await request.get('/__test__/thumbnail-files')).json();
+    const stem = name.replace(/\.[^.]+$/, '');
+    expect(files.folder.some(file => /^[a-z0-9_-]+-[0-9a-f]{8}\//.test(file)
+        && file.slice(file.indexOf('/') + 1).replace(/\.[^.]+$/, '') === stem)).toBe(true);
+});

tests/fixtures/legacy_contract_snapshot.json

@@ -212,6 +212,12 @@
     "thumbnail_encoders",
     "thumbnail_encoders.jpeg",
     "thumbnail_encoders.jxl",
-    "thumbnail_encoders.webp"
+    "thumbnail_encoders.webp",
+    "thumbnail_location",
+    "thumbnail_location.available",
+    "thumbnail_location.default_path",
+    "thumbnail_location.mode",
+    "thumbnail_location.path",
+    "thumbnail_location.problem"
   ]
 }

tests/test_api_thumbnail_location.py

@@ -0,0 +1,415 @@
+"""The thumbnail location setting (#320 PR 2): parsing, validation, source
+keys, the marker, the settings payloads and reverse containment.
+
+Spelling rules are tested on both path flavors (``ntpath`` and
+``posixpath``) as text; filesystem rules run on the native one.
+"""
+
+import json
+import ntpath
+import os
+import posixpath
+import re
+import sys
+import tempfile
+import threading
+import unittest
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+sys.modules.setdefault("server", MagicMock())
+
+from py.api import preview_location as pl  # noqa: E402
+from py.api import preview_store  # noqa: E402
+
+REPO = Path(__file__).resolve().parent.parent
+
+
+class TestParse(unittest.TestCase):
+    def test_missing_or_source_is_inside_each_source(self):
+        for value in (None, {"mode": "source"}, {}):
+            with self.subTest(value=value):
+                parsed = pl.parse_location(value)
+                self.assertEqual((parsed.mode, parsed.problem), ("source", None))
+
+    def test_a_folder_location_is_kept_as_spelled(self):
+        parsed = pl.parse_location(
+            {"mode": "folder", "path": "D:\\Thumbs", "id": "x", "install": "i"},
+            flavor=ntpath,
+        )
+        self.assertEqual(
+            (parsed.mode, parsed.path, parsed.id, parsed.problem),
+            ("folder", "D:\\Thumbs", "x", None),
+        )
+
+    def test_invalid_values_never_fall_back_to_the_sources(self):
+        for value in (
+            "folder",
+            {"mode": "elsewhere"},
+            {"mode": "folder", "path": "relative", "id": "x"},
+            {"mode": "folder", "path": "/abs"},
+            {"mode": "folder", "path": "/abs", "id": ""},
+            {"mode": "folder", "path": 7, "id": "x"},
+        ):
+            with self.subTest(value=value):
+                parsed = pl.parse_location(value, flavor=posixpath)
+                self.assertEqual((parsed.mode, parsed.problem), ("folder", "invalid"))
+
+
+class TestForm(unittest.TestCase):
+    def test_windows_spellings(self):
+        cases = {
+            "D:\\Thumbs": None,
+            "\\\\server\\share\\thumbs": None,
+            "C:thumbs": "location-not-absolute",
+            "\\thumbs": "location-not-absolute",
+            "thumbs": "location-not-absolute",
+            "": "location-not-absolute",
+            " D:\\Thumbs": "location-not-absolute",
+            "\\\\?\\D:\\Thumbs": "location-device-path",
+            "\\\\.\\D:\\Thumbs": "location-device-path",
+            "D:\\CON\\x": "location-invalid-name",
+            "D:\\com1.txt": "location-invalid-name",
+            "D:\\thumbs.": "location-invalid-name",
+            "D:\\thumbs ": "location-invalid-name",
+            "D:\\a|b": "location-invalid-name",
+            "D:\\a?b": "location-invalid-name",
+            "D:\\a\x01b": "location-invalid-name",
+        }
+        for path, code in cases.items():
+            with self.subTest(path=path):
+                self.assertEqual(pl.form_problem(path, flavor=ntpath), code)
+
+    def test_posix_spellings(self):
+        cases = {
+            "/mnt/fast/thumbs": None,
+            "mnt/thumbs": "location-not-absolute",
+            "/mnt/nul": "location-invalid-name",
+            "/mnt/a:b": "location-invalid-name",
+            "//?/x": "location-device-path",
+        }
+        for path, code in cases.items():
+            with self.subTest(path=path):
+                self.assertEqual(pl.form_problem(path, flavor=posixpath), code)
+
+    def test_roots_and_home_are_too_broad(self):
+        self.assertTrue(pl.is_too_broad("C:\\", home="C:\\Users\\a", flavor=ntpath))
+        self.assertTrue(
+            pl.is_too_broad("\\\\server\\share", home="C:\\Users\\a", flavor=ntpath)
+        )
+        self.assertTrue(
+            pl.is_too_broad("C:\\users\\A", home="C:\\Users\\a", flavor=ntpath)
+        )
+        self.assertTrue(pl.is_too_broad("/", home="/home/a", flavor=posixpath))
+        self.assertTrue(pl.is_too_broad("/home/a/", home="/home/a", flavor=posixpath))
+        self.assertFalse(
+            pl.is_too_broad("/home/a/thumbs", home="/home/a", flavor=posixpath)
+        )
+
+    def test_long_windows_paths_are_refused_without_long_paths(self):
+        short = "D:\\" + "a" * 100
+        long = "D:\\" + "a" * 200
+        self.assertFalse(pl.too_long_for_windows(short, long_paths=False))
+        self.assertTrue(pl.too_long_for_windows(long, long_paths=False))
+        self.assertFalse(pl.too_long_for_windows(long, long_paths=True))
+
+
+class TestSourceKeys(unittest.TestCase):
+    def test_slug(self):
+        self.assertEqual(pl.slug("ComfyUI Output"), "comfyui-output")
+        self.assertEqual(pl.slug("Ünïcödé_Fôlder"), "unicode_folder")
+        self.assertEqual(pl.slug(""), "source")
+        self.assertEqual(pl.slug("!!!"), "source")
+        self.assertEqual(len(pl.slug("x" * 80)), 24)
+
+    def test_keys_are_a_slug_plus_a_hash(self):
+        key = pl.source_key("D:\\AI\\Output", "d:\\ai\\output", {})
+        self.assertRegex(key, r"^output-[0-9a-f]{8}$")
+        self.assertRegex(pl.source_key("D:\\", "d:\\", {}), r"^source-[0-9a-f]{8}$")
+        self.assertRegex(
+            pl.source_key("\\\\nas\\share\\Out", "\\\\nas\\share\\out", {}),
+            r"^out-[0-9a-f]{8}$",
+        )
+
+    def test_a_taken_key_grows_its_hash(self):
+        first = pl.source_key("/a/out", "/a/out", {})
+        taken = {first: "/someone/else"}
+        second = pl.source_key("/a/out", "/a/out", taken)
+        self.assertRegex(second, r"^out-[0-9a-f]{12}$")
+        taken[second] = "/another"
+        self.assertRegex(
+            pl.source_key("/a/out", "/a/out", taken), r"^out-[0-9a-f]{16}$"
+        )
+        self.assertEqual(pl.source_key("/a/out", "/a/out", {first: "/a/out"}), first)
+
+    def test_lookup_by_spelling_identity_then_file_id(self):
+        keys = {
+            "out-0123abcd": {
+                "path": "D:\\Out",
+                "identity": "d:\\out",
+                "dev_ino": [1, 2],
+            },
+            "lib-0123abcd": {"path": "E:\\Lib", "identity": "e:\\lib", "dev_ino": None},
+        }
+        self.assertEqual(pl.find_key(keys, "D:\\Out", "x", None), "out-0123abcd")
+        self.assertEqual(pl.find_key(keys, "Z:\\Out", "d:\\out", None), "out-0123abcd")
+        # A letter spelling and its UNC spelling: the same file id.
+        self.assertEqual(
+            pl.find_key(keys, "\\\\nas\\out", "\\\\nas\\out", (1, 2)), "out-0123abcd"
+        )
+        self.assertIsNone(pl.find_key(keys, "F:\\New", "f:\\new", (9, 9)))
+
+
+class TestMarker(unittest.TestCase):
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self._tmp.cleanup)
+        self.folder = self._tmp.name
+
+    def test_init_creates_once_and_adopts_after(self):
+        first = pl.init_folder(self.folder, created_by_pm=True)
+        self.assertEqual(pl.init_folder(self.folder), first)
+        marker = pl.read_marker(self.folder)
+        self.assertEqual(marker["id"], first)
+        self.assertTrue(marker["created_by_pm"])
+        self.assertEqual(marker["version"], 1)
+
+    def test_writes_merge_what_another_writer_added(self):
+        pl.init_folder(self.folder)
+        pl.write_marker(
+            self.folder, lambda m: m["keys"].update({"a-0123abcd": {"install": "1"}})
+        )
+        pl.write_marker(
+            self.folder, lambda m: m["keys"].update({"b-0123abcd": {"install": "2"}})
+        )
+        self.assertEqual(
+            sorted(pl.read_marker(self.folder)["keys"]), ["a-0123abcd", "b-0123abcd"]
+        )
+        # No temp file is left beside it.
+        self.assertEqual(os.listdir(self.folder), [pl.MARKER_NAME])
+
+    def test_a_folder_without_the_marker_is_missing(self):
+        self.assertEqual(pl.folder_state(self.folder, "x")[:2], (False, "missing"))
+        missing = os.path.join(self.folder, "gone")
+        self.assertEqual(pl.folder_state(missing, "x")[:2], (False, "missing"))
+        marker_id = pl.init_folder(self.folder)
+        self.assertEqual(pl.folder_state(self.folder, marker_id)[:2], (True, None))
+        self.assertEqual(
+            pl.folder_state(self.folder, "other")[:2], (False, "marker-mismatch")
+        )
+
+    @unittest.skipUnless(hasattr(os, "symlink"), "needs symlinks")
+    def test_a_linked_folder_is_not_available(self):
+        real = os.path.join(self.folder, "real")
+        os.mkdir(real)
+        marker_id = pl.init_folder(real)
+        link = os.path.join(self.folder, "link")
+        try:
+            os.symlink(real, link, target_is_directory=True)
+        except OSError:
+            self.skipTest("symlinks not permitted")
+        self.assertEqual(pl.folder_state(link, marker_id)[:2], (False, "invalid"))
+
+
+class TestValidation(unittest.TestCase):
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self._tmp.cleanup)
+        self.base = os.path.realpath(self._tmp.name)
+        self.source = os.path.join(self.base, "output")
+        self.node = os.path.join(self.base, "node")
+        os.mkdir(self.source)
+        os.mkdir(self.node)
+        self.context = pl.Context(
+            roots=(self.source,),
+            node_folder=self.node,
+            default_path=os.path.join(self.node, "data", "thumbnails"),
+            home=os.path.join(self.base, "home"),
+            environ={},
+            long_paths=True,
+            folder_paths=None,
+        )
+
+    def problem(self, path, purpose="move", **changes):
+        context = self.context
+        if changes:
+            context = pl.Context(**{**context.__dict__, **changes})
+        return pl.location_problem(path, purpose=purpose, context=context)
+
+    def test_an_empty_new_folder_beside_the_sources_is_accepted(self):
+        self.assertIsNone(self.problem(os.path.join(self.base, "thumbs")))
+        os.mkdir(os.path.join(self.base, "empty"))
+        Path(self.base, "empty", "Thumbs.db").write_bytes(b"")
+        self.assertIsNone(self.problem(os.path.join(self.base, "empty")))
+
+    def test_codes(self):
+        cases = {
+            "relative": "location-not-absolute",
+            os.path.join(self.source, "thumbs"): "location-served",
+            os.path.join(self.node, "thumbs"): "location-node-folder",
+            self.base: "location-contains-source",
+            os.path.join(self.base, "no", "parent"): "location-parent-missing",
+        }
+        for path, code in cases.items():
+            with self.subTest(path=path):
+                self.assertEqual(self.problem(path), code)
+
+    def test_the_default_may_sit_in_the_node_folder_and_make_its_parents(self):
+        self.assertIsNone(self.problem(self.context.default_path))
+
+    def test_the_folder_in_use_is_the_same(self):
+        folder = os.path.join(self.base, "thumbs")
+        os.mkdir(folder)
+        self.assertEqual(self.problem(folder, current_path=folder), "location-same")
+
+    def test_a_folder_with_other_files_is_not_empty_unless_it_is_ours(self):
+        folder = os.path.join(self.base, "full")
+        os.mkdir(folder)
+        Path(folder, "photo.jpg").write_bytes(b"x")
+        self.assertEqual(self.problem(folder), "location-not-empty")
+        pl.init_folder(folder)
+        self.assertIsNone(self.problem(folder))
+
+    @unittest.skipIf(os.name == "nt" or os.geteuid() == 0, "needs POSIX modes")
+    def test_a_folder_that_refuses_writes(self):
+        folder = os.path.join(self.base, "ro")
+        os.mkdir(folder)
+        os.chmod(folder, 0o500)
+        self.addCleanup(os.chmod, folder, 0o700)
+        self.assertEqual(self.problem(folder), "location-not-writable")
+
+    def test_a_stat_that_does_not_answer_is_unavailable(self):
+        release = threading.Event()
+        self.addCleanup(release.set)
+
+        def stuck(*_args):
+            release.wait(10)
+
+        with patch.object(pl, "_filesystem_problem", stuck):
+            code = pl.location_problem(
+                os.path.join(self.base, "thumbs"),
+                purpose="move",
+                context=self.context,
+                timeout=0.1,
+            )
+        self.assertEqual(code, "location-unavailable")
+
+    def test_a_long_windows_path(self):
+        path = os.path.join(self.base, "t" * 220)
+        self.assertEqual(self.problem(path, long_paths=False), "location-path-too-long")
+
+    def test_every_code_has_a_message_without_a_path(self):
+        for code, (status, text) in pl.MESSAGES.items():
+            with self.subTest(code=code):
+                self.assertIn(status, (400, 409))
+                self.assertTrue(text.endswith("."))
+
+
+class TestNoLegacySaveCallers(unittest.TestCase):
+    def test_save_to_file_has_no_callers(self):
+        offenders = []
+        for folder in ("py", "utils", "database"):
+            for path in (REPO / folder).rglob("*.py"):
+                for number, line in enumerate(
+                    path.read_text(encoding="utf-8").splitlines(), 1
+                ):
+                    code = line.split("#", 1)[0]
+                    if (
+                        re.search(r"\bsave_to_file\(", code)
+                        and "def save_to_file" not in code
+                    ):
+                        if "PromptManagerConfig.save_to_file('" in line:
+                            continue  # its own docstring example
+                        offenders.append(f"{path.relative_to(REPO)}:{number}")
+        self.assertEqual(offenders, [])
+
+
+try:
+    from tests.test_api_images import ImageAPITestCase
+except ImportError:  # discovered with tests/ as the top-level directory
+    from test_api_images import ImageAPITestCase
+
+
+class TestLocationRoutes(ImageAPITestCase):
+    async def asyncSetUp(self):
+        await super().asyncSetUp()
+        self.addCleanup(preview_store.configure, None)
+
+    def use_folder(self):
+        folder = self.comfy_dir / "thumbs"
+        folder.mkdir()
+        marker_id = pl.init_folder(str(folder), created_by_pm=True)
+        preview_store.configure(
+            {"mode": "folder", "path": str(folder), "id": marker_id}
+        )
+        return folder
+
+    async def test_settings_get_names_the_location_with_its_path(self):
+        folder = self.use_folder()
+        resp = await self.client.get("/prompt_manager/settings")
+        location = (await resp.json())["thumbnail_location"]
+        self.assertEqual(location["mode"], "folder")
+        self.assertEqual(location["path"], str(folder))
+        self.assertTrue(location["available"])
+        self.assertIsNone(location["problem"])
+        self.assertIn("default_path", location)
+
+    async def test_settings_post_never_changes_the_location(self):
+        resp = await self.client.post(
+            "/prompt_manager/settings",
+            json={"thumbnail_location": {"mode": "folder", "path": "/x"}},
+        )
+        self.assertEqual(resp.status, 409)
+        body = await resp.json()
+        self.assertEqual(body["code"], "location-needs-move")
+        self.assertNotIn("/x", body["error"])
+
+    async def test_status_has_the_location_but_no_path(self):
+        folder = self.use_folder()
+        resp = await self.client.get("/prompt_manager/images/thumbnails/status")
+        body = await resp.text()
+        self.assertNotIn(str(folder), body)
+        self.assertEqual(
+            json.loads(body)["location"],
+            {"mode": "folder", "available": True, "problem": None},
+        )
+
+    async def test_a_source_inside_or_around_the_folder_is_refused(self):
+        from py.api import sources
+
+        folder = self.use_folder()
+        inside = folder / "sub"
+        inside.mkdir()
+        with patch(
+            "py.config.GalleryConfig.gallery_root_problem",
+            classmethod(lambda cls, path: (None, "")),
+        ):
+            for path in (inside, folder, self.comfy_dir):
+                with self.subTest(path=path):
+                    code, reason = sources.source_problem(str(path))
+                    self.assertEqual(code, "holds-thumbnails")
+                    self.assertEqual(reason, pl.HOLDS_THUMBNAILS_REASON)
+            self.assertEqual(sources.source_problem(str(self.output_dir))[0], None)
+
+    async def test_a_library_move_into_the_folder_is_refused(self):
+        from py.api import library_location as routes
+
+        folder = self.use_folder()
+        self.assertTrue(routes._inside_thumbnail_folder(str(folder / "lib")))
+        self.assertFalse(routes._inside_thumbnail_folder(str(self.output_dir)))
+
+    def test_startup_makes_a_location_inside_a_source_invalid(self):
+        inside = self.output_dir / "thumbs"
+        inside.mkdir()
+        marker_id = pl.init_folder(str(inside))
+        from py.config import PromptManagerConfig
+
+        value = {"mode": "folder", "path": str(inside), "id": marker_id}
+        with patch.object(PromptManagerConfig, "THUMBNAIL_LOCATION", value):
+            location = self.api.start_preview_location()
+        self.assertEqual((location.available, location.problem), (False, "invalid"))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_native_memory.py

@@ -30,8 +30,12 @@ class ReleaseFreeMemoryTests(unittest.TestCase):
     def test_musl_does_nothing(self):
         with (
             mock.patch.object(sys, "platform", "linux"),
+            # Windows has no os.confstr: this test fakes Linux there too.
             mock.patch.object(
-                native_memory.os, "confstr", side_effect=ValueError("unknown name")
+                native_memory.os,
+                "confstr",
+                side_effect=ValueError("unknown name"),
+                create=True,
             ),
             mock.patch.object(native_memory.ctypes, "CDLL") as cdll,
         ):

tests/test_preview_store_folder.py

@@ -0,0 +1,539 @@
+"""One folder mode (#320 PR 2): previews in ``<folder>/<source key>``.
+
+Folder mode is reachable from config.json and tests only (D4B). These tests
+set it on the resolver directly (``preview_store.configure``) and drive
+every route that reads, writes, lists or removes previews: they must touch
+only ``<folder>/<key>``, never ``<source>/thumbnails``, never through a link,
+and nothing at all while the folder is unavailable. Waits are events; no
+sleeps.
+"""
+
+import asyncio
+import json
+import os
+import sys
+import threading
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+try:
+    from tests.test_thumbnail_auto import AutoThumbnailCase, _entry, WAIT
+    from tests.test_api_images import make_png
+except ImportError:  # discovered with tests/ as the top-level directory
+    from test_thumbnail_auto import AutoThumbnailCase, _entry, WAIT
+    from test_api_images import make_png
+
+from py.api import preview_location, preview_store, thumbnail_settings  # noqa: E402
+
+SERVE = "/prompt_manager/images/serve/"
+
+
+class FolderModeCase(AutoThumbnailCase):
+    """The output root's previews live in ``<comfy>/thumbs/<key>``."""
+
+    async def asyncSetUp(self):
+        await super().asyncSetUp()
+        self.folder = self.use_folder(self.comfy_dir / "thumbs")
+        self.old_store = self.output_dir / "thumbnails"
+
+    def use_folder(self, folder, install="install-a"):
+        folder.mkdir(exist_ok=True)
+        marker_id = preview_location.init_folder(str(folder), created_by_pm=True)
+        self.addCleanup(preview_store.configure, None)
+        preview_store.configure(
+            {"mode": "folder", "path": str(folder), "id": marker_id, "install": install}
+        )
+        self.assertTrue(preview_store.current().available)
+        return folder
+
+    def key_folder(self):
+        store = preview_store.ensure_store(self.output_dir, create=True)
+        self.assertIsNotNone(store)
+        return store
+
+    def preview(self, rel, ext=".webp"):
+        return self.key_folder() / (thumbnail_settings.preview_stem(rel) + ext)
+
+    def marker(self):
+        return json.loads(
+            (self.folder / preview_location.MARKER_NAME).read_text(encoding="utf-8")
+        )
+
+
+class TestFolderModeRoutes(FolderModeCase):
+    async def test_the_key_is_a_readable_name_plus_a_hash_in_the_marker(self):
+        key = self.key_folder().name
+        self.assertRegex(key, r"^output-[0-9a-f]{8}$")
+        entry = self.marker()["keys"][key]
+        self.assertEqual(entry["path"], str(self.output_dir))
+        self.assertEqual(entry["install"], "install-a")
+
+    async def test_another_case_of_the_source_finds_the_same_key(self):
+        key = self.key_folder()
+        other = Path(str(self.output_dir).swapcase())
+        if not other.is_dir():
+            self.skipTest("a case-sensitive volume")
+        self.assertEqual(preview_store.store_dir(other), key)
+
+    @unittest.skipUnless(hasattr(os, "symlink"), "needs symlinks")
+    async def test_a_linked_spelling_of_the_source_finds_the_same_key(self):
+        key = self.key_folder()
+        link = self.comfy_dir / "output-link"
+        try:
+            link.symlink_to(self.output_dir, target_is_directory=True)
+        except OSError:
+            self.skipTest("symlinks not permitted")
+        self.assertEqual(preview_store.store_dir(link), key)
+
+    async def test_an_existing_preview_is_listed_and_served_from_the_key_folder(self):
+        make_png(self.output_dir / "sub" / "a.png", size=(64, 64))
+        preview = self.preview("sub/a.png")
+        make_png(preview, size=(32, 32))
+        url = _entry(self.api, self.output_dir, "sub/a.png")["thumbnail_url"]
+        self.assertIn("v=", url)
+        self.assertTrue(url.startswith(SERVE + "thumbnails/sub/a_thumb.webp"))
+        resp = await self.client.get(url)
+        self.assertEqual(resp.status, 200)
+        self.assertEqual(await resp.read(), preview.read_bytes())
+
+    async def test_a_missing_preview_is_made_in_the_key_folder(self):
+        make_png(self.output_dir / "b.png", size=(640, 480))
+        url = _entry(self.api, self.output_dir, "b.png")["thumbnail_url"]
+        resp = await self.client.get(url)
+        self.assertEqual(resp.status, 200)
+        self.assertTrue(self.preview("b.png").is_file())
+        self.assertFalse(self.old_store.exists())
+
+    async def test_a_leftover_in_the_source_folder_is_never_served(self):
+        make_png(self.output_dir / "c.png", size=(640, 480))
+        self.old_store.mkdir()
+        (self.old_store / "c_thumb.webp").write_bytes(b"leftover")
+        resp = await self.client.get(SERVE + "thumbnails/c_thumb.webp?root=0")
+        self.assertEqual(resp.status, 200)
+        self.assertNotEqual(await resp.read(), b"leftover")
+
+    async def test_other_text_under_thumbnails_is_refused(self):
+        self.old_store.mkdir()
+        make_png(self.old_store / "notes.png", size=(8, 8))
+        resp = await self.client.get(SERVE + "thumbnails/notes.png?root=0")
+        self.assertEqual(resp.status, 403)
+
+    async def test_traversal_through_thumbnails_is_refused(self):
+        for text in (
+            "thumbnails/../../thumbs/x_thumb.webp",
+            "thumbnails/..%2F..%2Fx_thumb.webp",
+            "thumbnails/%2e%2e/x_thumb.webp",
+        ):
+            with self.subTest(text=text):
+                resp = await self.client.get(SERVE + text + "?root=0")
+                self.assertIn(resp.status, (403, 404))
+
+    @unittest.skipUnless(hasattr(os, "symlink"), "needs symlinks")
+    async def test_a_link_out_of_the_key_folder_is_never_served_or_written(self):
+        outside = self.comfy_dir / "outside"
+        outside.mkdir()
+        make_png(outside / "d_thumb.webp", size=(8, 8))
+        make_png(self.output_dir / "sub" / "d.png", size=(64, 64))
+        try:
+            (self.key_folder() / "sub").symlink_to(outside, target_is_directory=True)
+        except OSError:
+            self.skipTest("symlinks not permitted")
+        resp = await self.client.get(SERVE + "thumbnails/sub/d_thumb.webp?root=0")
+        self.assertEqual(resp.status, 403)
+        self.assertFalse(preview_store.owns(self.output_dir, outside / "d_thumb.webp"))
+        self.assertFalse(
+            preview_store.owns(self.output_dir, self.key_folder() / "sub" / "x.webp")
+        )
+
+    @unittest.skipUnless(os.name == "nt", "junctions are Windows only")
+    async def test_a_junction_inside_a_store_is_refused(self):
+        import _winapi
+
+        outside = self.comfy_dir / "outside"
+        outside.mkdir()
+        (outside / "e_thumb.webp").write_bytes(b"x")
+        _winapi.CreateJunction(str(outside), str(self.key_folder() / "sub"))
+        self.assertTrue(preview_location.is_link_like(self.key_folder() / "sub"))
+        resp = await self.client.get(SERVE + "thumbnails/sub/e_thumb.webp?root=0")
+        self.assertEqual(resp.status, 403)
+
+    async def test_capture_writes_into_the_key_folder(self):
+        path = make_png(self.output_dir / "run" / "f.png", size=(64, 64))
+        self.assertEqual(self.api.queue_capture_thumbnail(str(path)), "queued")
+        await self._idle()
+        self.assertTrue(self.preview("run/f.png").is_file())
+        self.assertFalse(self.old_store.exists())
+
+    async def test_clear_removes_previews_only(self):
+        make_png(self.output_dir / "g.png", size=(64, 64))
+        payload = await asyncio.get_running_loop().run_in_executor(
+            None, self.api._generate_gallery_thumbnails_sync, [self.output_dir]
+        )
+        self.assertTrue(payload["success"])
+        keep = self.key_folder() / "readme.txt"
+        keep.write_text("not a preview", encoding="utf-8")
+        resp = await self.client.post("/prompt_manager/images/clear-thumbnails")
+        self.assertEqual(resp.status, 200)
+        self.assertFalse(self.preview("g.png").exists())
+        self.assertTrue(keep.exists())
+        self.assertTrue((self.folder / preview_location.MARKER_NAME).is_file())
+        self.assertTrue(self.output_dir.joinpath("g.png").is_file())
+
+    async def test_removing_a_file_removes_its_preview_in_the_key_folder(self):
+        from py.api.admin import _remove_thumbnail
+
+        make_png(self.output_dir / "h.png", size=(64, 64))
+        preview = self.preview("h.png")
+        preview.write_bytes(b"x")
+        self.assertTrue(_remove_thumbnail(self.output_dir, Path("h.png")))
+        self.assertFalse(preview.exists())
+
+    async def test_history_cover_is_authorized_by_the_store(self):
+        from py.api import history_current
+
+        preview = self.preview("i.png")
+        preview.write_bytes(b"x")
+        target = history_current._canonical(preview)
+        self.assertTrue(
+            history_current._target_authorized(
+                target, self.output_dir, None, self.output_dir
+            )
+        )
+        stray = self.comfy_dir / "stray_thumb.webp"
+        stray.write_bytes(b"x")
+        self.assertFalse(
+            history_current._target_authorized(
+                history_current._canonical(stray),
+                self.output_dir,
+                None,
+                self.output_dir,
+            )
+        )
+
+    async def test_the_thumbnail_folder_is_never_recreated_by_a_write(self):
+        make_png(self.output_dir / "j.png", size=(64, 64))
+        self.key_folder()
+        for child in sorted(self.folder.rglob("*"), reverse=True):
+            child.rmdir() if child.is_dir() else child.unlink()
+        self.folder.rmdir()
+        with self.assertRaises(FileNotFoundError):
+            preview_store.make_parent(self.folder / "output-x" / "sub" / "a.webp")
+        self.assertFalse(self.folder.exists())
+
+
+class TestUnavailableFolder(FolderModeCase):
+    def go_missing(self):
+        (self.folder / preview_location.MARKER_NAME).unlink()
+        self.assertFalse(preview_store.refresh())
+        location = preview_store.current()
+        self.assertEqual((location.available, location.problem), (False, "missing"))
+
+    async def test_a_folder_without_its_marker_is_missing_and_writes_nothing(self):
+        make_png(self.output_dir / "k.png", size=(640, 480))
+        self.go_missing()
+        entry = _entry(self.api, self.output_dir, "k.png")
+        self.assertFalse(entry.get("thumbnail_ready", False))
+        resp = await self.client.get(SERVE + "thumbnails/k_thumb.webp?root=0")
+        self.assertEqual(resp.status, 200)
+        self.assertEqual(resp.headers["Cache-Control"], "no-store")
+        self.assertEqual(
+            await resp.read(), (self.output_dir / "k.png").read_bytes()
+        )  # the original
+        self.assertEqual(
+            sorted(p.name for p in self.folder.rglob("*") if p.is_file()), []
+        )
+        self.assertFalse(self.old_store.exists())
+
+    async def test_thumbnail_work_is_refused_while_unavailable(self):
+        self.go_missing()
+        for method, url, body in (
+            ("POST", "/prompt_manager/images/generate-thumbnails", {}),
+            ("POST", "/prompt_manager/images/generate-thumbnails/progress", None),
+            ("POST", "/prompt_manager/images/clear-thumbnails", None),
+            ("POST", "/prompt_manager/jobs/thumbnails", {"mode": "missing"}),
+        ):
+            with self.subTest(url=url):
+                resp = await self.client.request(method, url, json=body)
+                self.assertEqual(resp.status, 409)
+                self.assertEqual((await resp.json())["code"], "location-unavailable")
+
+    async def test_status_reports_the_location_without_a_path(self):
+        self.go_missing()
+        resp = await self.client.get("/prompt_manager/images/thumbnails/status")
+        location = (await resp.json())["location"]
+        self.assertEqual(
+            location, {"mode": "folder", "available": False, "problem": "missing"}
+        )
+
+    async def test_another_folders_marker_is_a_mismatch(self):
+        marker = self.marker()
+        marker["id"] = "someone-else"
+        (self.folder / preview_location.MARKER_NAME).write_text(
+            json.dumps(marker), encoding="utf-8"
+        )
+        self.assertFalse(preview_store.refresh())
+        self.assertEqual(preview_store.current().problem, "marker-mismatch")
+
+    async def test_the_temp_sweep_skips_an_unavailable_folder(self):
+        stray = self.key_folder() / ".l_thumb.part-1-1.webp"
+        stray.write_bytes(b"x")
+        old = os.path.getmtime(stray) - 10 * 24 * 3600
+        os.utime(stray, (old, old))
+        self.go_missing()
+        self.assertEqual(self.api.sweep_thumbnail_temps(), 0)
+        self.assertTrue(stray.exists())
+
+    async def test_a_stalled_store_call_marks_it_unreachable_and_stops_io(self):
+        make_png(self.output_dir / "m.png", size=(64, 64))
+        self.preview("m.png").write_bytes(b"x")
+        ticks = iter([0.0, preview_store.IO_BUDGET + 1.0])
+        with patch.object(preview_store, "_clock", lambda: next(ticks, 99.0)):
+            _entry(self.api, self.output_dir, "m.png")  # one slow lookup
+        self.assertEqual(preview_store.current().problem, "unreachable")
+        calls = []
+        real = thumbnail_settings.owned_in_folder
+        with patch.object(
+            thumbnail_settings,
+            "owned_in_folder",
+            lambda *a, **k: calls.append(a) or real(*a, **k),
+        ):
+            entry = _entry(self.api, self.output_dir, "m.png")
+        self.assertEqual(calls, [])  # no store I/O while unreachable
+        self.assertNotIn("v=", entry["thumbnail_url"])
+        self.assertTrue(preview_store.refresh())  # the probe finds it again
+
+    async def test_a_check_that_blocks_is_unreachable_within_the_budget(self):
+        entered = threading.Event()
+        release = threading.Event()
+        self.addCleanup(release.set)
+
+        def blocking(path, expected_id):
+            entered.set()
+            release.wait(WAIT)
+            return True, None, None
+
+        with (
+            patch.object(preview_store, "IO_BUDGET", 0.2),
+            patch.object(preview_location, "folder_state", blocking),
+        ):
+            self.assertFalse(preview_store.refresh())
+        self.assertTrue(entered.is_set())
+        self.assertEqual(preview_store.current().problem, "unreachable")
+        release.set()
+
+    async def test_an_invalid_config_value_writes_nothing_anywhere(self):
+        preview_store.configure({"mode": "folder", "path": "relative/thumbs"})
+        location = preview_store.current()
+        self.assertEqual((location.available, location.problem), (False, "invalid"))
+        make_png(self.output_dir / "n.png", size=(640, 480))
+        resp = await self.client.get(SERVE + "thumbnails/n_thumb.webp?root=0")
+        self.assertEqual(resp.status, 200)
+        self.assertFalse(self.old_store.exists())
+
+
+class TestSharedFolders(FolderModeCase):
+    async def test_two_installs_keep_each_others_keys(self):
+        self.key_folder()
+        other_root = self.comfy_dir / "other"
+        other_root.mkdir()
+        setting = preview_store.setting()
+        preview_store.configure(
+            {
+                "mode": "folder",
+                "path": setting.path,
+                "id": setting.id,
+                "install": "install-b",
+            }
+        )
+        self.assertIsNotNone(preview_store.ensure_store(other_root))
+        keys = self.marker()["keys"]
+        installs = sorted(entry["install"] for entry in keys.values())
+        self.assertEqual(installs, ["install-a", "install-b"])
+
+    async def test_init_folder_adopts_an_existing_marker(self):
+        marker_id = self.marker()["id"]
+        self.assertEqual(preview_location.init_folder(str(self.folder)), marker_id)
+
+
+if __name__ == "__main__":
+    unittest.main()
+
+
+class TestHostileMarker(FolderModeCase):
+    """A hand-edited or foreign marker never takes a write or a delete out of
+    the thumbnail folder (shared and UNC folders can be edited by others)."""
+
+    def plant(self, keys):
+        marker = self.marker()
+        marker["keys"] = keys
+        (self.folder / preview_location.MARKER_NAME).write_text(
+            json.dumps(marker), encoding="utf-8"
+        )
+        self.assertTrue(preview_store.refresh())
+
+    def outside_files(self):
+        return sorted(
+            p.relative_to(self.comfy_dir).as_posix()
+            for p in self.comfy_dir.rglob("*")
+            if p.is_file()
+            and not p.is_relative_to(self.folder)
+            and not p.is_relative_to(self.output_dir)
+            and p.suffix in (".webp", ".jpg", ".png")
+        )
+
+    async def test_keys_that_are_not_one_folder_name_are_dropped(self):
+        entry = {"path": str(self.output_dir), "identity": "x", "install": "i"}
+        self.plant(
+            {
+                "../escape-0123abcd": entry,
+                str(self.comfy_dir / "abs-0123abcd"): entry,
+                "a/b-0123abcd": entry,
+                "a\\b-0123abcd": entry,
+                "Upper-0123abcd": entry,
+                "nohash": entry,
+                "fine-0123abcd": "not a dict",
+            }
+        )
+        self.assertEqual(preview_location.read_marker(str(self.folder))["keys"], {})
+        store = self.key_folder()
+        self.assertEqual(store.parent, self.folder)
+        make_png(self.output_dir / "p.png", size=(640, 480))
+        resp = await self.client.get(SERVE + "thumbnails/p_thumb.webp?root=0")
+        self.assertEqual(resp.status, 200)
+        self.assertTrue((store / "p_thumb.webp").is_file())
+        self.assertEqual(self.outside_files(), [])
+
+    @unittest.skipUnless(hasattr(os, "symlink"), "needs symlinks")
+    async def test_a_key_folder_that_is_a_link_is_never_written_or_cleared(self):
+        outside = self.comfy_dir / "outside"
+        outside.mkdir()
+        victim = outside / "q_thumb.webp"
+        victim.write_bytes(b"keep me")
+        store = self.key_folder()
+        store.rmdir()
+        try:
+            store.symlink_to(outside, target_is_directory=True)
+        except OSError:
+            self.skipTest("symlinks not permitted")
+        make_png(self.output_dir / "q.png", size=(640, 480))
+        from py.api.admin import _remove_thumbnail
+
+        self.assertFalse(_remove_thumbnail(self.output_dir, Path("q.png")))
+        resp = await self.client.post("/prompt_manager/images/clear-thumbnails")
+        self.assertIn(resp.status, (200, 400))
+        resp = await self.client.get(SERVE + "thumbnails/q_thumb.webp?root=0")
+        self.assertIn(resp.status, (200, 403))
+        self.assertEqual(victim.read_bytes(), b"keep me")
+        self.assertEqual(sorted(p.name for p in outside.iterdir()), ["q_thumb.webp"])
+
+    @unittest.skipUnless(hasattr(os, "symlink"), "needs symlinks")
+    async def test_a_linked_folder_below_a_key_is_never_deleted_through(self):
+        outside = self.comfy_dir / "outside"
+        outside.mkdir()
+        victim = outside / "r_thumb.webp"
+        victim.write_bytes(b"keep me")
+        try:
+            (self.key_folder() / "sub").symlink_to(outside, target_is_directory=True)
+        except OSError:
+            self.skipTest("symlinks not permitted")
+        make_png(self.output_dir / "sub" / "r.png", size=(64, 64))
+        from py.api.admin import _remove_thumbnail
+
+        self.assertFalse(_remove_thumbnail(self.output_dir, Path("sub/r.png")))
+        resp = await self.client.post("/prompt_manager/images/clear-thumbnails")
+        self.assertEqual(resp.status, 200)
+        self.assertEqual(victim.read_bytes(), b"keep me")
+
+    @unittest.skipUnless(os.name == "nt", "junctions are Windows only")
+    async def test_a_key_folder_that_is_a_junction_is_never_written_or_cleared(self):
+        import _winapi
+
+        outside = self.comfy_dir / "outside"
+        outside.mkdir()
+        victim = outside / "v_thumb.webp"
+        victim.write_bytes(b"keep me")
+        store = self.key_folder()
+        store.rmdir()
+        _winapi.CreateJunction(str(outside), str(store))
+        make_png(self.output_dir / "v.png", size=(640, 480))
+        from py.api.admin import _remove_thumbnail
+
+        self.assertFalse(_remove_thumbnail(self.output_dir, Path("v.png")))
+        resp = await self.client.post("/prompt_manager/images/clear-thumbnails")
+        self.assertIn(resp.status, (200, 400))
+        resp = await self.client.get(SERVE + "thumbnails/v_thumb.webp?root=0")
+        self.assertIn(resp.status, (200, 403))
+        self.assertEqual(victim.read_bytes(), b"keep me")
+        self.assertEqual(sorted(p.name for p in outside.iterdir()), ["v_thumb.webp"])
+
+    async def test_a_write_outside_the_folder_is_refused(self):
+        with self.assertRaises(OSError):
+            preview_store.make_parent(self.comfy_dir / "elsewhere" / "a_thumb.webp")
+        self.assertFalse((self.comfy_dir / "elsewhere").exists())
+
+    async def test_an_unmounted_folder_takes_no_write_before_the_next_check(self):
+        """An unmounted mount point is an empty folder: the flag may still
+        say available, but without the marker nothing is made there."""
+        store = self.key_folder()
+        store.rmdir()
+        (self.folder / preview_location.MARKER_NAME).unlink()
+        self.assertTrue(preview_store.current().available)  # not checked yet
+        with self.assertRaises(OSError):
+            preview_store.make_parent(store / "sub" / "s_thumb.webp")
+        self.assertEqual(list(self.folder.iterdir()), [])
+
+
+class TestStartupNeverFallsBackToTheSources(AutoThumbnailCase):
+    async def test_a_folder_setting_is_in_force_before_its_check(self):
+        self.addCleanup(preview_store.configure, None)
+        folder = self.comfy_dir / "thumbs"
+        value = {"mode": "folder", "path": str(folder), "id": "x"}
+        preview_store.configure(value, check=False)
+        location = preview_store.current()
+        self.assertEqual((location.mode, location.available), ("folder", False))
+        make_png(self.output_dir / "t.png", size=(640, 480))
+        resp = await self.client.get(SERVE + "thumbnails/t_thumb.webp?root=0")
+        self.assertEqual(resp.status, 200)
+        self.assertFalse((self.output_dir / "thumbnails").exists())
+
+    async def test_a_failed_startup_check_leaves_the_folder_invalid(self):
+        from py.config import PromptManagerConfig
+
+        self.addCleanup(preview_store.configure, None)
+        value = {"mode": "folder", "path": str(self.comfy_dir / "thumbs"), "id": "x"}
+        with (
+            patch.object(PromptManagerConfig, "THUMBNAIL_LOCATION", value),
+            patch.object(
+                type(self.api),
+                "_thumbnail_location_context",
+                side_effect=RuntimeError("boom"),
+            ),
+        ):
+            self.api.start_preview_location()
+        location = preview_store.current()
+        self.assertEqual(
+            (location.mode, location.available, location.problem),
+            ("folder", False, "invalid"),
+        )
+
+
+class TestNoStoreIoOnTheEventLoop(FolderModeCase):
+    async def test_legacy_listings_do_not_stat_the_folder(self):
+        make_png(self.output_dir / "u.png", size=(64, 64))
+        self.preview("u.png").write_bytes(b"x")
+        calls = []
+        real = thumbnail_settings.owned_in_folder
+        with patch.object(
+            thumbnail_settings,
+            "owned_in_folder",
+            lambda *a, **k: calls.append(threading.current_thread().name)
+            or real(*a, **k),
+        ):
+            images = self.api._enrich_images(
+                [{"image_path": str(self.output_dir / "u.png")}]
+            )
+        self.assertEqual(calls, [])
+        self.assertEqual(images[0]["thumbnail_url"], SERVE + "thumbnails/u_thumb.webp")